diff --git a/src/app/services/classroomStatusService.spec.ts b/src/app/services/classroomStatusService.spec.ts index 4fbbbb305a3..9af2ccbdc35 100644 --- a/src/app/services/classroomStatusService.spec.ts +++ b/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(); diff --git a/src/assets/wise5/services/classroomStatusService.ts b/src/assets/wise5/services/classroomStatusService.ts index a52909c159e..41b46ade1c5 100644 --- a/src/assets/wise5/services/classroomStatusService.ts +++ b/src/assets/wise5/services/classroomStatusService.ts @@ -3,7 +3,7 @@ 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'; @@ -11,6 +11,7 @@ import { CompletionCounter } from '../common/CompletionCounter'; @Injectable() export class ClassroomStatusService { studentStatuses = []; + private workgroupIdsReceivedDuringRefreshes: Set> = new Set(); private studentStatusReceivedSource: Subject = new Subject(); public studentStatusReceived$: Observable = 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 { + return defer(() => { + const received = new Set(); + this.workgroupIdsReceivedDuringRefreshes.add(received); + return this.http + .get(`/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): 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++) { diff --git a/src/assets/wise5/services/teacherWebSocketService.spec.ts b/src/assets/wise5/services/teacherWebSocketService.spec.ts new file mode 100644 index 00000000000..bb1d8f6e124 --- /dev/null +++ b/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>; + +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(); + 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(); + }); + }); +} diff --git a/src/assets/wise5/services/teacherWebSocketService.ts b/src/assets/wise5/services/teacherWebSocketService.ts index 33cd8eb9346..a38968bedb8 100644 --- a/src/assets/wise5/services/teacherWebSocketService.ts +++ b/src/assets/wise5/services/teacherWebSocketService.ts @@ -4,7 +4,15 @@ import { Injectable } from '@angular/core'; import { ConfigService } from './configService'; import { ClassroomStatusService } from './classroomStatusService'; import { NotificationService } from './notificationService'; -import { Observable, Subject } from 'rxjs'; +import { + catchError, + debounceTime, + EMPTY, + Observable, + Subject, + Subscription, + switchMap +} from 'rxjs'; import { AchievementService } from './achievementService'; import { RxStomp } from '@stomp/rx-stomp'; import { Message } from '@stomp/stompjs'; @@ -15,9 +23,13 @@ export class TeacherWebSocketService { runId: number; rxStomp: RxStomp; private newAnnotationReceivedSource: Subject = new Subject(); - public newAnnotationReceived$: Observable = this.newAnnotationReceivedSource.asObservable(); + public newAnnotationReceived$: Observable = + this.newAnnotationReceivedSource.asObservable(); private newStudentWorkReceivedSource: Subject = new Subject(); - public newStudentWorkReceived$: Observable = this.newStudentWorkReceivedSource.asObservable(); + public newStudentWorkReceived$: Observable = + this.newStudentWorkReceivedSource.asObservable(); + private subscriptions: Subscription = new Subscription(); + private workgroupJoinedSource: Subject = new Subject(); constructor( private AchievementService: AchievementService, @@ -27,11 +39,19 @@ export class TeacherWebSocketService { ) {} initialize() { + this.disconnect(); this.runId = this.configService.getRunId(); this.initializeStomp(); this.subscribeToTeacherTopic(); this.subscribeToTeacherWorkgroupTopic(); this.subscribeToClassroomTopics(); + this.subscribeToWorkgroupJoined(); + } + + private disconnect(): void { + this.subscriptions.unsubscribe(); + this.subscriptions = new Subscription(); + this.rxStomp?.deactivate(); } initializeStomp() { @@ -39,26 +59,44 @@ export class TeacherWebSocketService { this.rxStomp.configure({ brokerURL: this.configService.getWebSocketURL() }); + this.subscriptions.add(this.rxStomp.connected$.subscribe(() => this.refreshStudentStatuses())); this.rxStomp.activate(); } + // The broker keeps nothing for a disconnected subscriber, so statuses sent while the socket was + // down can only be recovered by fetching them again. + private refreshStudentStatuses(): void { + this.subscriptions.add( + this.classroomStatusService.refreshStudentStatuses().subscribe({ + next: (changedStatuses) => { + for (const studentStatus of changedStatuses) { + this.classroomStatusService.broadcastStudentStatusReceived({ studentStatus }); + } + }, + error: () => {} + }) + ); + } + subscribeToTeacherTopic() { - this.rxStomp.watch(`/topic/teacher/${this.runId}`).subscribe((message: Message) => { - const body = JSON.parse(message.body); - if (body.type === 'studentWork') { - const studentWork = JSON.parse(body.content); - this.broadcastNewStudentWorkReceived({ studentWork: studentWork }); - } else if (body.type === 'studentStatus') { - const status = JSON.parse(body.content); - this.classroomStatusService.setStudentStatus(status); - this.classroomStatusService.broadcastStudentStatusReceived({ studentStatus: status }); - } else if (body.type === 'newStudentAchievement') { - const achievement = JSON.parse(body.content); - this.AchievementService.broadcastNewStudentAchievement(achievement); - } else if (body.type === 'annotation') { - this.broadcastNewAnnotationReceived(JSON.parse(body.content)); - } - }); + this.subscriptions.add( + this.rxStomp.watch(`/topic/teacher/${this.runId}`).subscribe((message: Message) => { + const body = JSON.parse(message.body); + if (body.type === 'studentWork') { + const studentWork = JSON.parse(body.content); + this.broadcastNewStudentWorkReceived({ studentWork: studentWork }); + } else if (body.type === 'studentStatus') { + const status = JSON.parse(body.content); + this.classroomStatusService.setStudentStatus(status); + this.classroomStatusService.broadcastStudentStatusReceived({ studentStatus: status }); + } else if (body.type === 'newStudentAchievement') { + const achievement = JSON.parse(body.content); + this.AchievementService.broadcastNewStudentAchievement(achievement); + } else if (body.type === 'annotation') { + this.broadcastNewAnnotationReceived(JSON.parse(body.content)); + } + }) + ); } broadcastNewStudentWorkReceived(args: any) { @@ -70,14 +108,16 @@ export class TeacherWebSocketService { } subscribeToTeacherWorkgroupTopic() { - this.rxStomp - .watch(`/topic/workgroup/${this.configService.getWorkgroupId()}`) - .subscribe((message: Message) => { - const body = JSON.parse(message.body); - if (body.type === 'notification') { - this.NotificationService.addNotification(JSON.parse(body.content)); - } - }); + this.subscriptions.add( + this.rxStomp + .watch(`/topic/workgroup/${this.configService.getWorkgroupId()}`) + .subscribe((message: Message) => { + const body = JSON.parse(message.body); + if (body.type === 'notification') { + this.NotificationService.addNotification(JSON.parse(body.content)); + } + }) + ); } pauseScreens(periodId: number): void { @@ -97,16 +137,31 @@ export class TeacherWebSocketService { private subscribeToClassroomTopics(): void { for (const period of this.configService.getPeriods()) { - this.rxStomp - .watch(`/topic/classroom/${this.configService.getRunId()}/${period.periodId}`) - .subscribe((message: Message) => { - const body = JSON.parse(message.body); - if (body.type === 'newWorkgroupJoinedRun') { - this.configService.retrieveConfig( - `/api/config/classroomMonitor/${this.configService.getRunId()}` - ); - } - }); + this.subscriptions.add( + this.rxStomp + .watch(`/topic/classroom/${this.configService.getRunId()}/${period.periodId}`) + .subscribe((message: Message) => { + const body = JSON.parse(message.body); + if (body.type === 'newWorkgroupJoinedRun') { + this.workgroupJoinedSource.next(); + } + }) + ); } } + + private subscribeToWorkgroupJoined(): void { + this.subscriptions.add( + this.workgroupJoinedSource + .pipe( + debounceTime(2000), + switchMap(() => + this.configService + .retrieveConfig(`/api/config/classroomMonitor/${this.runId}`) + .pipe(catchError(() => EMPTY)) + ) + ) + .subscribe() + ); + } }