fix(ClassroomMonitor): Recover student statuses missed while the socket was down

src/app/services/classroomStatusService.spec.ts

@@ -50,9 +50,79 @@ describe('ClassroomStatusService', () => {
     projectService.setProject(copy(project));
   });
   retrieveStudentStatuses();
+  refreshStudentStatuses();
   getNodeCompletion();
 });
 
+function refreshStudentStatuses() {
+  describe('refreshStudentStatuses', () => {
+    const row = (workgroupId: number, nodeId: string) => ({
+      status: JSON.stringify({ workgroupId: workgroupId, currentNodeId: nodeId }),
+      timestamp: 2,
+      workgroupId: workgroupId
+    });
+    let changed: any[];
+    beforeEach(() => {
+      spyOn(configService, 'getRunId').and.returnValue(runId1);
+      spyOn(configService, 'getClassmateUserInfos').and.returnValue([
+        { workgroupId: 1 },
+        { workgroupId: 2 },
+        { workgroupId: 3 }
+      ]);
+      service.studentStatuses = [
+        { workgroupId: 1, currentNodeId: nodeId1, postTimestamp: 1 },
+        { workgroupId: 2, currentNodeId: nodeId1, postTimestamp: 1 },
+        { workgroupId: 4, currentNodeId: nodeId1, postTimestamp: 1 }
+      ];
+      service.refreshStudentStatuses().subscribe((statuses) => (changed = statuses));
+    });
+
+    it('merges newer statuses and returns only those that changed', () => {
+      http
+        .expectOne(`/api/teacher/run/${runId1}/student-status`)
+        .flush([row(1, nodeId1), row(2, nodeId2), row(3, nodeId3), row(99, nodeId1)]);
+      expect(changed.map((s) => s.workgroupId)).toEqual([2, 3]);
+      expect(service.getStudentStatusForWorkgroupId(2).currentNodeId).toEqual(nodeId2);
+      expect(service.getStudentStatusForWorkgroupId(3).currentNodeId).toEqual(nodeId3);
+      expect(service.hasStudentStatus(4)).toBeTrue();
+      expect(service.hasStudentStatus(99)).toBeFalse();
+    });
+
+    it('keeps a status that arrived over the socket during the request', () => {
+      service.setStudentStatus({ workgroupId: 2, currentNodeId: nodeId4 });
+      http.expectOne(`/api/teacher/run/${runId1}/student-status`).flush([row(2, nodeId2)]);
+      expect(changed).toEqual([]);
+      expect(service.getStudentStatusForWorkgroupId(2).currentNodeId).toEqual(nodeId4);
+    });
+
+    it('stops keeping socket statuses once the request has finished', () => {
+      http.expectOne(`/api/teacher/run/${runId1}/student-status`).flush([]);
+      service.setStudentStatus({ workgroupId: 2, currentNodeId: nodeId4 });
+      service.refreshStudentStatuses().subscribe((statuses) => (changed = statuses));
+      http.expectOne(`/api/teacher/run/${runId1}/student-status`).flush([row(2, nodeId2)]);
+      expect(service.getStudentStatusForWorkgroupId(2).currentNodeId).toEqual(nodeId2);
+    });
+
+    it('does not let an older response that returns last overwrite a newer one', () => {
+      service.refreshStudentStatuses().subscribe();
+      const [first, second] = http.match(`/api/teacher/run/${runId1}/student-status`);
+      second.flush([{ ...row(2, nodeId4), timestamp: 3 }]);
+      first.flush([row(2, nodeId2)]);
+      expect(service.getStudentStatusForWorkgroupId(2).currentNodeId).toEqual(nodeId4);
+    });
+
+    it('keeps a status that arrived during an earlier request that was still in flight', () => {
+      service.refreshStudentStatuses().subscribe();
+      service.setStudentStatus({ workgroupId: 2, currentNodeId: nodeId4 });
+      const [first, second] = http.match(`/api/teacher/run/${runId1}/student-status`);
+      first.flush([row(2, nodeId2)]);
+      second.flush([row(2, nodeId2)]);
+      expect(changed).toEqual([]);
+      expect(service.getStudentStatusForWorkgroupId(2).currentNodeId).toEqual(nodeId4);
+    });
+  });
+}
+
 function retrieveStudentStatuses() {
   describe('retrieveStudentStatuses', () => {
     retrieveStudentStatuses_SetStudentStatuses();

src/assets/wise5/services/classroomStatusService.ts

@@ -3,14 +3,15 @@ import { HttpClient } from '@angular/common/http';
 import { AnnotationService } from './annotationService';
 import { ConfigService } from './configService';
 import { ProjectService } from './projectService';
-import { Observable, Subject, tap } from 'rxjs';
+import { defer, finalize, map, Observable, Subject, tap } from 'rxjs';
 import { ProjectCompletion } from '../common/ProjectCompletion';
 import { StudentStatus } from '../common/StudentStatus';
 import { CompletionCounter } from '../common/CompletionCounter';
 
 @Injectable()
 export class ClassroomStatusService {
   studentStatuses = [];
+  private workgroupIdsReceivedDuringRefreshes: Set<Set<number>> = new Set();
   private studentStatusReceivedSource: Subject<any> = new Subject<any>();
   public studentStatusReceived$: Observable<any> = this.studentStatusReceivedSource.asObservable();
 
@@ -77,7 +78,65 @@ export class ClassroomStatusService {
     return this.getStudentStatusForWorkgroupId(workgroupId) != null;
   }
 
+  /**
+   * Unlike retrieveStudentStatuses(), never replaces a status with an older one and keeps statuses
+   * that arrived over the socket while the request was in flight, since the response may be older
+   * than they are.
+   * @returns the statuses that changed
+   */
+  refreshStudentStatuses(): Observable<any[]> {
+    return defer(() => {
+      const received = new Set<number>();
+      this.workgroupIdsReceivedDuringRefreshes.add(received);
+      return this.http
+        .get<any[]>(`/api/teacher/run/${this.configService.getRunId()}/student-status`)
+        .pipe(
+          map((rows: any[]) => this.mergeStudentStatuses(rows, received)),
+          finalize(() => this.workgroupIdsReceivedDuringRefreshes.delete(received))
+        );
+    });
+  }
+
+  private mergeStudentStatuses(rows: any[], received: Set<number>): any[] {
+    const changed = [];
+    for (const status of this.parseStudentStatuses(rows)) {
+      const current = this.getStudentStatusForWorkgroupId(status.workgroupId);
+      if (
+        received.has(status.workgroupId) ||
+        this.isNotNewer(status, current) ||
+        (current != null && this.isSameStatus(current, status))
+      ) {
+        continue;
+      }
+      this.replaceStudentStatus(status);
+      changed.push(status);
+    }
+    return changed;
+  }
+
+  // Only statuses fetched from the server carry a timestamp; those received over the socket do not.
+  private isNotNewer(status: any, current: any): boolean {
+    return (
+      current?.postTimestamp != null &&
+      new Date(status.postTimestamp).getTime() <= new Date(current.postTimestamp).getTime()
+    );
+  }
+
+  private isSameStatus(a: any, b: any): boolean {
+    return (
+      JSON.stringify({ ...a, postTimestamp: undefined }) ===
+      JSON.stringify({ ...b, postTimestamp: undefined })
+    );
+  }
+
   setStudentStatus(studentStatus: any): void {
+    this.workgroupIdsReceivedDuringRefreshes.forEach((received) =>
+      received.add(studentStatus.workgroupId)
+    );
+    this.replaceStudentStatus(studentStatus);
+  }
+
+  private replaceStudentStatus(studentStatus: any): void {
     let isStudentStatusFound = false;
     const studentStatuses = this.getStudentStatuses();
     for (let x = 0; x < studentStatuses.length; x++) {

src/assets/wise5/services/teacherWebSocketService.spec.ts

@@ -0,0 +1,126 @@
+import { discardPeriodicTasks, fakeAsync, TestBed, tick } from '@angular/core/testing';
+import { RxStomp, RxStompState } from '@stomp/rx-stomp';
+import { MockProvider } from 'ng-mocks';
+import { defer, of, Subject, throwError } from 'rxjs';
+import { AchievementService } from './achievementService';
+import { ClassroomStatusService } from './classroomStatusService';
+import { ConfigService } from './configService';
+import { NotificationService } from './notificationService';
+import { TeacherWebSocketService } from './teacherWebSocketService';
+
+const runId = 1;
+let service: TeacherWebSocketService;
+let classroomStatusService: ClassroomStatusService;
+let configService: ConfigService;
+let topics: Map<string, Subject<any>>;
+
+describe('TeacherWebSocketService', () => {
+  beforeEach(() => {
+    topics = new Map();
+    spyOn(RxStomp.prototype, 'activate');
+    spyOn(RxStomp.prototype, 'deactivate').and.returnValue(Promise.resolve());
+    spyOn(RxStomp.prototype, 'watch').and.callFake((destination: any) => {
+      const topic = new Subject<any>();
+      topics.set(destination, topic);
+      return topic;
+    });
+    TestBed.configureTestingModule({
+      providers: [
+        MockProvider(AchievementService),
+        MockProvider(ClassroomStatusService),
+        MockProvider(ConfigService, {
+          getRunId: () => runId,
+          getWorkgroupId: () => 10,
+          getPeriods: () => [{ periodId: 1 }],
+          getWebSocketURL: () => 'ws://localhost/websocket'
+        }),
+        MockProvider(NotificationService),
+        TeacherWebSocketService
+      ]
+    });
+    service = TestBed.inject(TeacherWebSocketService);
+    classroomStatusService = TestBed.inject(ClassroomStatusService);
+    configService = TestBed.inject(ConfigService);
+  });
+  refreshStudentStatusesOnConnect();
+  reloadConfigWhenWorkgroupsJoin();
+  initializeAgain();
+});
+
+function connect(): void {
+  service.rxStomp.connectionState$.next(RxStompState.OPEN);
+}
+
+function sendWorkgroupJoined(): void {
+  topics
+    .get(`/topic/classroom/${runId}/1`)
+    .next({ body: JSON.stringify({ type: 'newWorkgroupJoinedRun' }) });
+}
+
+function refreshStudentStatusesOnConnect() {
+  describe('on connect', () => {
+    it('refreshes student statuses and broadcasts only those that changed', () => {
+      const changed = [{ workgroupId: 2 }];
+      spyOn(classroomStatusService, 'refreshStudentStatuses').and.returnValue(of(changed));
+      const broadcast = spyOn(classroomStatusService, 'broadcastStudentStatusReceived');
+      service.initialize();
+      expect(classroomStatusService.refreshStudentStatuses).not.toHaveBeenCalled();
+      connect();
+      expect(classroomStatusService.refreshStudentStatuses).toHaveBeenCalledTimes(1);
+      expect(broadcast.calls.allArgs()).toEqual([[{ studentStatus: changed[0] }]]);
+      service.rxStomp.connectionState$.next(RxStompState.CLOSED);
+      connect();
+      expect(classroomStatusService.refreshStudentStatuses).toHaveBeenCalledTimes(2);
+    });
+  });
+}
+
+function reloadConfigWhenWorkgroupsJoin() {
+  describe('when workgroups join the run', () => {
+    let requests: number;
+    beforeEach(() => {
+      requests = 0;
+      spyOn(configService, 'retrieveConfig').and.callFake(() =>
+        defer(() => {
+          requests++;
+          return requests === 1 ? throwError(() => new Error()) : of({});
+        })
+      );
+      service.initialize();
+    });
+
+    it('reloads the config once per burst and keeps working after a failed reload', fakeAsync(() => {
+      sendWorkgroupJoined();
+      sendWorkgroupJoined();
+      sendWorkgroupJoined();
+      tick(2000);
+      expect(configService.retrieveConfig).toHaveBeenCalledOnceWith(
+        `/api/config/classroomMonitor/${runId}`
+      );
+      expect(requests).toEqual(1);
+      sendWorkgroupJoined();
+      tick(2000);
+      expect(requests).toEqual(2);
+      discardPeriodicTasks();
+    }));
+  });
+}
+
+function initializeAgain() {
+  describe('initialize called again', () => {
+    it('deactivates the previous client and stops its subscriptions', () => {
+      spyOn(classroomStatusService, 'refreshStudentStatuses').and.returnValue(of([]));
+      const setStudentStatus = spyOn(classroomStatusService, 'setStudentStatus');
+      service.initialize();
+      const firstClient = service.rxStomp;
+      const firstTeacherTopic = topics.get(`/topic/teacher/${runId}`);
+      service.initialize();
+      expect(RxStomp.prototype.deactivate).toHaveBeenCalledTimes(1);
+      expect(service.rxStomp).not.toBe(firstClient);
+      expect(firstTeacherTopic.observed).toBeFalse();
+      firstClient.connectionState$.next(RxStompState.OPEN);
+      expect(classroomStatusService.refreshStudentStatuses).not.toHaveBeenCalled();
+      expect(setStudentStatus).not.toHaveBeenCalled();
+    });
+  });
+}