From 2e8668b5171106f6d0dfa17d750bf7c01e7ef061 Mon Sep 17 00:00:00 2001 From: Isaries Date: Tue, 27 Oct 2026 20:02:24 +0800 Subject: [PATCH] fix(ClassroomMonitor): Recover student statuses missed while the socket was down The simple broker keeps nothing for a subscriber that is not connected, so any student status sent while the teacher's socket was reconnecting was lost until the page was reloaded. Student statuses are now fetched again on every connect, including the first one, which also covers statuses posted between the initial request and the socket connecting. The response is merged rather than replacing the list: a workgroup whose status arrived over the socket while any refresh was in flight keeps that status, a fetched status never replaces one with a newer server timestamp, workgroups missing from the loaded config are skipped as retrieveStudentStatuses() does, and only statuses that changed are broadcast. The classroom monitor config reload on newWorkgroupJoinedRun was never subscribed, so new workgroups did not appear. It is now subscribed, debounced and restartable, and an error does not stop later reloads. Re-initializing now deactivates the previous STOMP client and tears down all of its subscriptions. --- .../services/classroomStatusService.spec.ts | 70 ++++++++++ .../wise5/services/classroomStatusService.ts | 61 ++++++++- .../services/teacherWebSocketService.spec.ts | 126 +++++++++++++++++ .../wise5/services/teacherWebSocketService.ts | 129 +++++++++++++----- 4 files changed, 348 insertions(+), 38 deletions(-) create mode 100644 src/assets/wise5/services/teacherWebSocketService.spec.ts 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() + ); + } }