From 2960a3e2ee7561413f6659c4b748d16fc45ee816 Mon Sep 17 00:00:00 2001 From: Javier Date: Thu, 15 Oct 2020 09:51:15 +0200 Subject: [PATCH 1/3] Fix cleanup observer blocking unsubscribe (2) (#6985) --- src/core/__tests__/QueryManager/index.ts | 67 ++++++++++++++++++- .../__tests__/client/Mutation.test.tsx | 2 +- src/utilities/observables/Concast.ts | 26 +++---- .../testing/mocking/MockedProvider.tsx | 2 +- src/utilities/testing/mocking/mockLink.ts | 21 ++++-- 5 files changed, 99 insertions(+), 19 deletions(-) diff --git a/src/core/__tests__/QueryManager/index.ts b/src/core/__tests__/QueryManager/index.ts index 8f861def4e4..ab1f1fb1c79 100644 --- a/src/core/__tests__/QueryManager/index.ts +++ b/src/core/__tests__/QueryManager/index.ts @@ -16,7 +16,7 @@ import { // mocks import mockQueryManager from '../../../utilities/testing/mocking/mockQueryManager'; import mockWatchQuery from '../../../utilities/testing/mocking/mockWatchQuery'; -import { MockApolloLink, mockSingleLink } from '../../../utilities/testing/mocking/mockLink'; +import { MockApolloLink, mockSingleLink, MockLink } from '../../../utilities/testing/mocking/mockLink'; // core import { ApolloQueryResult } from '../../types'; @@ -460,6 +460,71 @@ describe('QueryManager', () => { expect(subscription.unsubscribe).not.toThrow(); }); + // Query should be aborted on last .unsubscribe() + itAsync('causes immediate link unsubscription if unsubscribed', (resolve, reject) => { + const expResult = { + data: { + allPeople: { + people: [ + { + name: 'Luke Skywalker', + }, + ], + }, + }, + }; + + const request = { + query: gql` + query people { + allPeople(first: 1) { + people { + name + } + } + } + `, + variables: undefined + }; + + const mockedResponse = { + request, + result: expResult + }; + + const onRequestSubscribe = jest.fn(); + const onRequestUnsubscribe = jest.fn(); + + const mockedSingleLink = new MockLink([mockedResponse], { + addTypename: true, + onSubscribe: onRequestSubscribe, + onUnsubscribe: onRequestUnsubscribe + }); + + const mockedQueryManger = new QueryManager({ + link: mockedSingleLink, + cache: new InMemoryCache({ addTypename: false }), + }); + + const observableQuery = mockedQueryManger.watchQuery({ + query: request.query, + variables: request.variables, + notifyOnNetworkStatusChange: false + }); + + const subscription = observableQuery.subscribe({ + next: wrap(reject, () => { + reject(new Error('Link subscriptions should have been cancelled')); + }), + }); + + subscription.unsubscribe(); + + expect(onRequestSubscribe).toHaveBeenCalledTimes(1) + expect(onRequestUnsubscribe).toHaveBeenCalledTimes(1) + resolve(); + }); + itAsync('supports interoperability with other Observable implementations like RxJS', (resolve, reject) => { const expResult = { data: { diff --git a/src/react/components/__tests__/client/Mutation.test.tsx b/src/react/components/__tests__/client/Mutation.test.tsx index e3c3861d3d8..d70f8e16617 100644 --- a/src/react/components/__tests__/client/Mutation.test.tsx +++ b/src/react/components/__tests__/client/Mutation.test.tsx @@ -108,7 +108,7 @@ describe('General Mutation testing', () => { function mockClient(m: any) { return new ApolloClient({ - link: new MockLink(m, false), + link: new MockLink(m, { addTypename: false }), cache: new Cache({ addTypename: false }) }); } diff --git a/src/utilities/observables/Concast.ts b/src/utilities/observables/Concast.ts index 932fc159ab5..51bc22dbce2 100644 --- a/src/utilities/observables/Concast.ts +++ b/src/utilities/observables/Concast.ts @@ -130,16 +130,19 @@ export class Concast extends Observable { observer: Observer, quietly?: boolean, ) { - if (this.observers.delete(observer) && - this.observers.size < 1) { - if (quietly) return; - if (this.sub) { - this.sub.unsubscribe(); - // In case anyone happens to be listening to this.promise, after - // this.observers has become empty. - this.reject(new Error("Observable cancelled prematurely")); + if (this.observers.delete(observer)) { + --this.addCount; + + if (this.addCount < 1) { + if (quietly) return; + if (this.sub) { + this.sub.unsubscribe(); + // In case anyone happens to be listening to this.promise, after + // this.observers has become empty. + this.reject(new Error("Observable cancelled prematurely")); + } + this.sub = null; } - this.sub = null; } } @@ -209,13 +212,12 @@ export class Concast extends Observable { const once = () => { if (!called) { called = true; - // If there have been no other (non-cleanup) observers added, pass - // true for the quietly argument, so the removal of the cleanup + // Pass true for the quietly argument, so the removal of the cleanup // observer does not call this.sub.unsubscribe. If a cleanup // observer is added and removed before any other observers // subscribe, we do not want to prevent other observers from // subscribing later. - this.removeObserver(observer, !this.addCount); + this.removeObserver(observer, true); callback(); } } diff --git a/src/utilities/testing/mocking/MockedProvider.tsx b/src/utilities/testing/mocking/MockedProvider.tsx index 5612a0f53c5..63e3ece1998 100644 --- a/src/utilities/testing/mocking/MockedProvider.tsx +++ b/src/utilities/testing/mocking/MockedProvider.tsx @@ -47,7 +47,7 @@ export class MockedProvider extends React.Component< defaultOptions, link: link || new MockLink( mocks || [], - addTypename, + { addTypename }, ), resolvers, }); diff --git a/src/utilities/testing/mocking/mockLink.ts b/src/utilities/testing/mocking/mockLink.ts index a754c3640bc..741b64e69bd 100644 --- a/src/utilities/testing/mocking/mockLink.ts +++ b/src/utilities/testing/mocking/mockLink.ts @@ -35,17 +35,25 @@ function requestToKey(request: GraphQLRequest, addTypename: Boolean): string { return JSON.stringify(requestKey); } +interface MockLinkOptions { + addTypename?: boolean; + onSubscribe?: () => void; + onUnsubscribe?: () => void; +} + export class MockLink extends ApolloLink { public operation: Operation; public addTypename: Boolean = true; private mockedResponsesByKey: { [key: string]: MockedResponse[] } = {}; + private options: MockLinkOptions; constructor( mockedResponses: ReadonlyArray, - addTypename: Boolean = true + options: MockLinkOptions = {} ) { super(); - this.addTypename = addTypename; + this.options = options; + this.addTypename = options.addTypename ?? true; if (mockedResponses) { mockedResponses.forEach(mockedResponse => { this.addMockedResponse(mockedResponse); @@ -109,7 +117,9 @@ export class MockLink extends ApolloLink { } } - return new Observable(observer => { + const requestObservable = new Observable(observer => { + this.options.onSubscribe?.(); + const timer = setTimeout(() => { if (configError) { try { @@ -141,9 +151,12 @@ export class MockLink extends ApolloLink { }, response && response.delay || 0); return () => { + this.options.onUnsubscribe?.(); clearTimeout(timer); }; }); + + return requestObservable; } private normalizeMockedResponse( @@ -183,5 +196,5 @@ export function mockSingleLink( maybeTypename = true; } - return new MockLink(mocks, maybeTypename); + return new MockLink(mocks, { addTypename: maybeTypename }); } From 4ce6195e41d29f171c94bacb8b64cbd456e83325 Mon Sep 17 00:00:00 2001 From: Ben Newman Date: Thu, 15 Oct 2020 15:10:58 -0400 Subject: [PATCH 2/3] Simplify concast.addCount logic and unsubscribe asynchronously. The test failure in #7170 was a case of unsubscribing the last observer and then immediately resubscribing another observer, which should keep the subscription alive, for backwards compatibility. Delaying the unsubscribe to a later microtask tick enables this use case, while still unsubscribing very soon after the last non-cleanup Concast observer unsubscribes. In case there are still any cleanup observers in this.observers, and no error or completion has been broadcast yet, we call this.handlers.error to make sure those cleanup observers receive an error that terminates them. --- src/core/__tests__/QueryManager/index.ts | 14 +++++-- src/utilities/observables/Concast.ts | 48 +++++++++++------------- 2 files changed, 32 insertions(+), 30 deletions(-) diff --git a/src/core/__tests__/QueryManager/index.ts b/src/core/__tests__/QueryManager/index.ts index ab1f1fb1c79..59aa36e0f1b 100644 --- a/src/core/__tests__/QueryManager/index.ts +++ b/src/core/__tests__/QueryManager/index.ts @@ -461,7 +461,7 @@ describe('QueryManager', () => { }); // Query should be aborted on last .unsubscribe() - itAsync('causes immediate link unsubscription if unsubscribed', (resolve, reject) => { + itAsync('causes link unsubscription if unsubscribed', (resolve, reject) => { const expResult = { data: { allPeople: { @@ -520,9 +520,15 @@ describe('QueryManager', () => { subscription.unsubscribe(); - expect(onRequestSubscribe).toHaveBeenCalledTimes(1) - expect(onRequestUnsubscribe).toHaveBeenCalledTimes(1) - resolve(); + return new Promise( + // Unsubscribing from the link happens after a microtask + // (Promise.resolve().then) delay, so we need to wait at least that + // long before verifying onRequestUnsubscribe was called. + resolve => setTimeout(resolve, 0) + ).then(() => { + expect(onRequestSubscribe).toHaveBeenCalledTimes(1); + expect(onRequestUnsubscribe).toHaveBeenCalledTimes(1); + }).then(resolve, reject); }); itAsync('supports interoperability with other Observable implementations like RxJS', (resolve, reject) => { diff --git a/src/utilities/observables/Concast.ts b/src/utilities/observables/Concast.ts index 51bc22dbce2..b0d04939f77 100644 --- a/src/utilities/observables/Concast.ts +++ b/src/utilities/observables/Concast.ts @@ -92,7 +92,7 @@ export class Concast extends Observable { // source observable. It's tempting to do this step lazily in // addObserver, but this.promise can be accessed without calling // addObserver, so consumption needs to begin eagerly. - this.handlers.complete!(); + this.handlers.complete(); } private deliverLastMessage(observer: Observer) { @@ -130,19 +130,13 @@ export class Concast extends Observable { observer: Observer, quietly?: boolean, ) { - if (this.observers.delete(observer)) { - --this.addCount; - - if (this.addCount < 1) { - if (quietly) return; - if (this.sub) { - this.sub.unsubscribe(); - // In case anyone happens to be listening to this.promise, after - // this.observers has become empty. - this.reject(new Error("Observable cancelled prematurely")); - } - this.sub = null; - } + if (this.observers.delete(observer) && + --this.addCount < 1 && + !quietly) { + // In case there are still any cleanup observers in this.observers, + // and no error or completion has been broadcast yet, make sure + // those observers receive an error that terminates them. + this.handlers.error(new Error("Observable cancelled prematurely")); } } @@ -162,17 +156,21 @@ export class Concast extends Observable { // Bound handler functions that can be reused for every internal // subscription. - private handlers: Observer = { - next: result => { + private handlers = { + next: (result: T) => { if (this.sub !== null) { this.latest = ["next", result]; iterateObserversSafely(this.observers, "next", result); } }, - error: error => { - if (this.sub !== null) { - if (this.sub) this.sub.unsubscribe(); + error: (error: any) => { + const { sub } = this; + if (sub !== null) { + // Delay unsubscribing from the underlying subscription slightly, + // so that immediately subscribing another observer can keep the + // subscription active. + if (sub) Promise.resolve().then(() => sub.unsubscribe()); this.sub = null; this.latest = ["error", error]; this.reject(error); @@ -212,12 +210,10 @@ export class Concast extends Observable { const once = () => { if (!called) { called = true; - // Pass true for the quietly argument, so the removal of the cleanup - // observer does not call this.sub.unsubscribe. If a cleanup - // observer is added and removed before any other observers - // subscribe, we do not want to prevent other observers from - // subscribing later. - this.removeObserver(observer, true); + // Removing a cleanup observer should not unsubscribe from the + // underlying Observable, so the only removeObserver behavior we + // need here is to delete observer from this.observers. + this.observers.delete(observer); callback(); } } @@ -238,7 +234,7 @@ export class Concast extends Observable { public cancel = (reason: any) => { this.reject(reason); this.sources = []; - this.handlers.complete!(); + this.handlers.complete(); } } From ef936eda89bec97be1e41cdfb5cda96183e60333 Mon Sep 17 00:00:00 2001 From: Javier Date: Fri, 16 Oct 2020 08:35:49 +0200 Subject: [PATCH 3/3] Maintain MockLink public API --- src/core/__tests__/QueryManager/index.ts | 30 ++++++++++++++----- .../__tests__/client/Mutation.test.tsx | 2 +- .../testing/mocking/MockedProvider.tsx | 2 +- src/utilities/testing/mocking/mockLink.ts | 21 +++---------- 4 files changed, 28 insertions(+), 27 deletions(-) diff --git a/src/core/__tests__/QueryManager/index.ts b/src/core/__tests__/QueryManager/index.ts index 59aa36e0f1b..f2b32b8bd51 100644 --- a/src/core/__tests__/QueryManager/index.ts +++ b/src/core/__tests__/QueryManager/index.ts @@ -16,7 +16,7 @@ import { // mocks import mockQueryManager from '../../../utilities/testing/mocking/mockQueryManager'; import mockWatchQuery from '../../../utilities/testing/mocking/mockWatchQuery'; -import { MockApolloLink, mockSingleLink, MockLink } from '../../../utilities/testing/mocking/mockLink'; +import { MockApolloLink, mockSingleLink } from '../../../utilities/testing/mocking/mockLink'; // core import { ApolloQueryResult } from '../../types'; @@ -495,10 +495,20 @@ describe('QueryManager', () => { const onRequestSubscribe = jest.fn(); const onRequestUnsubscribe = jest.fn(); - const mockedSingleLink = new MockLink([mockedResponse], { - addTypename: true, - onSubscribe: onRequestSubscribe, - onUnsubscribe: onRequestUnsubscribe + const mockedSingleLink = new ApolloLink(() => { + return new Observable(observer => { + onRequestSubscribe(); + + const timer = setTimeout(() => { + observer.next(mockedResponse.result); + observer.complete(); + }, 0); + + return () => { + onRequestUnsubscribe(); + clearTimeout(timer); + }; + }); }); const mockedQueryManger = new QueryManager({ @@ -512,10 +522,14 @@ describe('QueryManager', () => { notifyOnNetworkStatusChange: false }); + const observerCallback = wrap(reject, () => { + reject(new Error('Link subscription should have been cancelled')); + }); + const subscription = observableQuery.subscribe({ - next: wrap(reject, () => { - reject(new Error('Link subscriptions should have been cancelled')); - }), + next: observerCallback, + error: observerCallback, + complete: observerCallback }); subscription.unsubscribe(); diff --git a/src/react/components/__tests__/client/Mutation.test.tsx b/src/react/components/__tests__/client/Mutation.test.tsx index d70f8e16617..e3c3861d3d8 100644 --- a/src/react/components/__tests__/client/Mutation.test.tsx +++ b/src/react/components/__tests__/client/Mutation.test.tsx @@ -108,7 +108,7 @@ describe('General Mutation testing', () => { function mockClient(m: any) { return new ApolloClient({ - link: new MockLink(m, { addTypename: false }), + link: new MockLink(m, false), cache: new Cache({ addTypename: false }) }); } diff --git a/src/utilities/testing/mocking/MockedProvider.tsx b/src/utilities/testing/mocking/MockedProvider.tsx index 63e3ece1998..5612a0f53c5 100644 --- a/src/utilities/testing/mocking/MockedProvider.tsx +++ b/src/utilities/testing/mocking/MockedProvider.tsx @@ -47,7 +47,7 @@ export class MockedProvider extends React.Component< defaultOptions, link: link || new MockLink( mocks || [], - { addTypename }, + addTypename, ), resolvers, }); diff --git a/src/utilities/testing/mocking/mockLink.ts b/src/utilities/testing/mocking/mockLink.ts index 741b64e69bd..a754c3640bc 100644 --- a/src/utilities/testing/mocking/mockLink.ts +++ b/src/utilities/testing/mocking/mockLink.ts @@ -35,25 +35,17 @@ function requestToKey(request: GraphQLRequest, addTypename: Boolean): string { return JSON.stringify(requestKey); } -interface MockLinkOptions { - addTypename?: boolean; - onSubscribe?: () => void; - onUnsubscribe?: () => void; -} - export class MockLink extends ApolloLink { public operation: Operation; public addTypename: Boolean = true; private mockedResponsesByKey: { [key: string]: MockedResponse[] } = {}; - private options: MockLinkOptions; constructor( mockedResponses: ReadonlyArray, - options: MockLinkOptions = {} + addTypename: Boolean = true ) { super(); - this.options = options; - this.addTypename = options.addTypename ?? true; + this.addTypename = addTypename; if (mockedResponses) { mockedResponses.forEach(mockedResponse => { this.addMockedResponse(mockedResponse); @@ -117,9 +109,7 @@ export class MockLink extends ApolloLink { } } - const requestObservable = new Observable(observer => { - this.options.onSubscribe?.(); - + return new Observable(observer => { const timer = setTimeout(() => { if (configError) { try { @@ -151,12 +141,9 @@ export class MockLink extends ApolloLink { }, response && response.delay || 0); return () => { - this.options.onUnsubscribe?.(); clearTimeout(timer); }; }); - - return requestObservable; } private normalizeMockedResponse( @@ -196,5 +183,5 @@ export function mockSingleLink( maybeTypename = true; } - return new MockLink(mocks, { addTypename: maybeTypename }); + return new MockLink(mocks, maybeTypename); }