From f56c2a27bb4cf140c266259d560e2f83f570ddb5 Mon Sep 17 00:00:00 2001 From: Ehsan Rezaei Date: Thu, 3 Sep 2026 13:11:24 +0200 Subject: [PATCH] AAE-51358 Cleaning up websocket service from old subscription protocol (#12212) * AAE-51358 Cleaning up websocket service from old subscription protocol * AAE-51358 Fixing retry on error bug --- .../lib/services/web-socket.service.spec.ts | 111 +++++++++++++++++- .../src/lib/services/web-socket.service.ts | 36 +++--- 2 files changed, 126 insertions(+), 21 deletions(-) diff --git a/lib/process-services-cloud/src/lib/services/web-socket.service.spec.ts b/lib/process-services-cloud/src/lib/services/web-socket.service.spec.ts index 9f47740424..402b1ae380 100644 --- a/lib/process-services-cloud/src/lib/services/web-socket.service.spec.ts +++ b/lib/process-services-cloud/src/lib/services/web-socket.service.spec.ts @@ -16,18 +16,39 @@ */ import { TestBed } from '@angular/core/testing'; +import { Injectable } from '@angular/core'; import { Apollo, gql } from 'apollo-angular'; import { lastValueFrom, of, Subject } from 'rxjs'; import { WebSocketService } from './web-socket.service'; -import { SubscriptionOptions } from '@apollo/client/core'; +import { ApolloLink, execute, FetchResult, Observable as ApolloObservable, SubscriptionOptions } from '@apollo/client/core'; import { provideHttpClientTesting } from '@angular/common/http/testing'; import { AuthenticationService, AppConfigService } from '@alfresco/adf-core'; +import { Client, ClientOptions, Sink, SubscribePayload } from 'graphql-ws'; +import { HttpLink } from 'apollo-angular/http'; + +@Injectable() +class TestWebSocketService extends WebSocketService { + public capturedOnError: (() => void) | undefined; + + protected override createWsClient(clientOptions: ClientOptions): Client { + this.capturedOnError = clientOptions.on?.error as (() => void) | undefined; + + return { + on: () => () => undefined, + subscribe: (_payload: SubscribePayload, _sink: Sink) => () => undefined, + async *iterate() {}, + terminate: () => undefined, + dispose: () => undefined + }; + } +} describe('WebSocketService', () => { - let service: WebSocketService; + let service: TestWebSocketService; const onLogoutSubject: Subject = new Subject(); - const apolloMock = jasmine.createSpyObj('Apollo', ['use', 'createNamed']); + const apolloMock = jasmine.createSpyObj('Apollo', ['use', 'createNamed', 'removeClient']); + const httpLinkMock = jasmine.createSpyObj('HttpLink', ['create']); beforeEach(() => { TestBed.configureTestingModule({ @@ -37,6 +58,14 @@ describe('WebSocketService', () => { provide: Apollo, useValue: apolloMock }, + { + provide: WebSocketService, + useClass: TestWebSocketService + }, + { + provide: HttpLink, + useValue: httpLinkMock + }, { provide: AppConfigService, useValue: { @@ -52,13 +81,15 @@ describe('WebSocketService', () => { } ] }); - service = TestBed.inject(WebSocketService); + service = TestBed.inject(WebSocketService) as TestWebSocketService; apolloMock.use.and.returnValues(undefined, { subscribe: () => of({}) }); }); afterEach(() => { apolloMock.use.calls.reset(); apolloMock.createNamed.calls.reset(); + apolloMock.removeClient.calls.reset(); + httpLinkMock.create.calls.reset(); }); it('should not create a new Apollo client if it is already in use', async () => { @@ -95,7 +126,7 @@ describe('WebSocketService', () => { const apolloClientName = 'testClient'; const subscriptionOptions: SubscriptionOptions = { query: gql(`subscription {testQuery}`) }; const wsOptions = { apolloClientName, wsUrl: 'testUrl', subscriptionOptions }; - apolloMock.createNamed.and.callFake((_, options) => { + apolloMock.createNamed.and.callFake((_: any, options: { headers: {} }) => { headers = options.headers; }); @@ -105,4 +136,74 @@ describe('WebSocketService', () => { expect(apolloMock.createNamed).toHaveBeenCalled(); expect(headers).toEqual(expectedHeaders); }); + + it('should recreate the subscription client when the websocket connection errors', async () => { + const apolloClientName = 'testClient'; + const subscriptionOptions: SubscriptionOptions = { query: gql(`subscription {testQuery}`) }; + const wsOptions = { apolloClientName, wsUrl: 'testUrl', subscriptionOptions }; + + await lastValueFrom(service.getSubscription(wsOptions)); + + expect(apolloMock.createNamed).toHaveBeenCalledTimes(1); + expect(apolloMock.removeClient).not.toHaveBeenCalled(); + + if (!service.capturedOnError) { + fail('Expected websocket error handler to be registered'); + return; + } + + service.capturedOnError(); + + expect(apolloMock.removeClient).toHaveBeenCalledWith(apolloClientName); + expect(apolloMock.createNamed).toHaveBeenCalledTimes(2); + expect(apolloMock.createNamed).toHaveBeenCalledWith(apolloClientName, jasmine.any(Object)); + }); + + it('should retry the operation when a GraphQL error is unauthenticated', async () => { + const apolloClientName = 'testClient'; + const subscriptionOptions: SubscriptionOptions = { query: gql(`subscription {testQuery}`) }; + const wsOptions = { apolloClientName, wsUrl: 'testUrl', httpUrl: 'testHttpUrl', subscriptionOptions }; + const expectedResult: FetchResult = { data: { retried: true } }; + let createdLink: ApolloLink | undefined; + let requestCount = 0; + + httpLinkMock.create.and.returnValue( + new ApolloLink( + () => + new ApolloObservable((observer) => { + requestCount++; + + if (requestCount === 1) { + observer.next({ + errors: [{ message: 'Unauthorized', extensions: { code: 'UNAUTHENTICATED' } }] + }); + } else { + observer.next(expectedResult); + } + + observer.complete(); + }) + ) + ); + apolloMock.createNamed.and.callFake((_clientName: any, options: { link: ApolloLink | undefined }) => { + createdLink = options.link; + }); + + await lastValueFrom(service.getSubscription(wsOptions)); + + if (!createdLink) { + fail('Expected Apollo link to be created'); + return; + } + + const result = await new Promise((resolve, reject) => { + execute(createdLink!, { query: gql(`query { testQuery }`) }).subscribe({ + next: resolve, + error: reject + }); + }); + + expect(requestCount).toBe(2); + expect(result).toEqual(expectedResult); + }); }); diff --git a/lib/process-services-cloud/src/lib/services/web-socket.service.ts b/lib/process-services-cloud/src/lib/services/web-socket.service.ts index 4b9ebf95a9..c9e0e36076 100644 --- a/lib/process-services-cloud/src/lib/services/web-socket.service.ts +++ b/lib/process-services-cloud/src/lib/services/web-socket.service.ts @@ -15,10 +15,9 @@ * limitations under the License. */ -import { createClient } from 'graphql-ws'; +import { Client, ClientOptions, createClient } from 'graphql-ws'; import { inject, Injectable } from '@angular/core'; import { GraphQLWsLink } from '@apollo/client/link/subscriptions'; -import { WebSocketLink } from '@apollo/client/link/ws'; import { DefaultContext, FetchResult, @@ -56,9 +55,8 @@ export class WebSocketService { private readonly authService = inject(AuthenticationService); private readonly appConfigService = inject(AppConfigService); - private readonly subscriptionProtocol: 'graphql-ws' | 'transport-ws' = 'graphql-ws'; - private wsLink: GraphQLWsLink | WebSocketLink; - private httpLinkHandler: HttpLinkHandler; + private wsLink!: GraphQLWsLink; + private httpLinkHandler: HttpLinkHandler | undefined; public getSubscription(options: serviceOptions): Observable> { const { apolloClientName, subscriptionOptions } = options; @@ -110,8 +108,7 @@ export class WebSocketService { operation.setContext(({ headers }: DefaultContext) => ({ headers: { ...headers, - ...(this.subscriptionProtocol === 'graphql-ws' && { Authorization: `Bearer ${this.authService.getToken()}` }), - ...(this.subscriptionProtocol === 'transport-ws' && { 'X-Authorization': `Bearer ${this.authService.getToken()}` }) + Authorization: `Bearer ${this.authService.getToken()}` } })); return forward(operation); @@ -120,8 +117,8 @@ export class WebSocketService { const errorLink = onError(({ graphQLErrors, networkError, operation, forward }) => { if (graphQLErrors) { for (const error of graphQLErrors) { - if (error.extensions && error.extensions['code'] === 'UNAUTHENTICATED') { - authLink(operation, forward); + if (error.extensions?.['code'] === 'UNAUTHENTICATED') { + return authLink(operation, forward); } } } @@ -129,6 +126,8 @@ export class WebSocketService { if (networkError) { console.error(`[Network error]: ${networkError}`); } + + return undefined; }); const retryLink = new RetryLink({ @@ -145,8 +144,7 @@ export class WebSocketService { this.apollo.createNamed(options.apolloClientName, { headers: { - ...(this.subscriptionProtocol === 'graphql-ws' && { Authorization: `Bearer ${this.authService.getToken()}` }), - ...(this.subscriptionProtocol === 'transport-ws' && { 'X-Authorization': `Bearer ${this.authService.getToken()}` }) + Authorization: `Bearer ${this.authService.getToken()}` }, link: from([authLink, retryLink, errorLink, link]), cache: new InMemoryCache({ merge: true } as InMemoryCacheConfig) @@ -155,22 +153,28 @@ export class WebSocketService { private createGraphQLWsLink(options: serviceOptions): void { this.wsLink = new GraphQLWsLink( - createClient({ + this.createWsClient({ url: this.createWsUrl(options.wsUrl) + '/v2/ws/graphql', connectionParams: () => ({ Authorization: 'Bearer ' + this.authService.getToken() }), on: { - error: () => { - this.apollo.removeClient(options.apolloClientName); - this.initSubscriptions(options); - } + error: () => this.reconnect(options) }, lazy: true }) ); } + protected createWsClient(clientOptions: ClientOptions): Client { + return createClient(clientOptions); + } + + private reconnect(options: serviceOptions): void { + this.apollo.removeClient(options.apolloClientName); + this.initSubscriptions(options); + } + private createHttpLinkHandler(options: serviceOptions): void { this.httpLinkHandler = options.httpUrl ? this.httpLink.create({