implement deduplication options

README.md

@@ -51,3 +51,47 @@ this.link = ApolloLink.from([
     new HttpLink({ uri: URI_TO_YOUR_GRAPHQL_SERVER }),
 ]);
 ```
+
+### Deduplicate
+
+It's possible to deduplicate operations while the queue is closed.
+This for example allows to only save the newest changes without the steps between.
+
+```js
+import { ApolloLink } from 'apollo-link';
+import { BatchHttpLink } from 'apollo-link-batch-http';
+import QueueLink from 'apollo-link-queue';
+
+const offlineLink = new QueueLink({
+    /**
+     * Decides which entry to keep in the queue in case of duplicate entries.
+     * Possible values:
+     *  - last: removes existing duplicate
+     *  - first: ignores new duplicate
+     *  - all: doesn't deduplicate
+     *
+     * Defaults to 'all'.
+     */
+    keepPolicy: 'last',
+
+    /**
+     * Specifies which entries are considered duplicates
+     *
+     * Defaults to comparing operation operationA.toKey() === operationB.toKey()
+     * https://www.apollographql.com/docs/link/overview.html
+     */
+    isDuplicate: (a, b) => (a.operationName === 'save' &&
+        a.operationName === b.operationName && a.variables.id === b.variables.id)
+});
+
+this.link = ApolloLink.from([
+    new RetryLink({
+      attempts: {
+        // We don't want to retry deduped operations
+        retryIf: (error, _operation) => !!error && error.name !== 'DedupedByQueueError'
+      }
+    }),
+    offlineLink,
+    new BatchHttpLink({ uri: URI_TO_YOUR_GRAPHQL_SERVER }),
+]);
+```

src/QueueLink.test.ts

@@ -1,6 +1,9 @@
-import QueueLink from './QueueLink';
+import QueueLink, {
+    DedupedByQueueError
+} from './QueueLink';
 import {
     assertObservableSequence,
+    executeMultiple,
     TestLink,
 } from './TestUtils';
 import {
@@ -21,13 +24,26 @@ describe('OnOffLink', () => {
         },
     };
 
+    const testResponse2 = {
+        data: {
+            hello2: 'World',
+        },
+    };
+
     const op: GraphQLRequest = {
-        query: gql`{ hello }`,
+        query: gql`query hello { hello }`,
         context: {
             testResponse,
         },
     };
 
+    const op2: GraphQLRequest = {
+        query: gql`query hello2 { hello }`,
+        context: {
+            testResponse: testResponse2,
+        },
+    };
+
     beforeEach(() => {
         jest.useFakeTimers();
         testLink = new TestLink();
@@ -51,7 +67,7 @@ describe('OnOffLink', () => {
     });
     it('skips the queue when asked to', () => {
         const opWithSkipQueue: GraphQLRequest = {
-            query: gql`{ hello }`,
+            query: gql`query hello { hello }`,
             context: {
                 skipQueue: true,
             },
@@ -94,6 +110,7 @@ describe('OnOffLink', () => {
         expect(testLink.operations.length).toBe(0);
         sub.unsubscribe();
     });
+
     it('releases held requests when you open it', () => {
         onOffLink.close();
         return assertObservableSequence(
@@ -105,6 +122,70 @@ describe('OnOffLink', () => {
             () => {
                 expect(testLink.operations.length).toBe(0);
                 onOffLink.open();
+                expect(testLink.operations.length).toBe(1);
+                jest.runAllTimers();
+            },
+        );
+    });
+
+    it('releases held deduplicated requests when you open it (last)', () => {
+        const dedupOnOffLink = new QueueLink({keepPolicy: "last"});
+        const myLink = ApolloLink.from([dedupOnOffLink, testLink]);
+        dedupOnOffLink.close();
+        return assertObservableSequence(
+            executeMultiple(myLink, op, op2, op),
+            [
+                { type: 'error', value: new DedupedByQueueError()},
+                { type: 'next', value: testResponse2 },
+                { type: 'next', value: testResponse },
+                { type: 'complete' },
+            ],
+            () => {
+                expect(testLink.operations.length).toBe(0);
+                dedupOnOffLink.open();
+                expect(testLink.operations.length).toBe(2);
+                jest.runAllTimers();
+            },
+        );
+    });
+
+    it('releases held deduplicated requests when you open it (first)', () => {
+        const dedupOnOffLink = new QueueLink({keepPolicy: "first"});
+        const myLink = ApolloLink.from([dedupOnOffLink, testLink]);
+        dedupOnOffLink.close();
+        return assertObservableSequence(
+            executeMultiple(myLink, op, op2, op),
+            [
+                { type: 'error', value: new DedupedByQueueError()},
+                { type: 'next', value: testResponse },
+                { type: 'next', value: testResponse2 },
+                { type: 'complete' },
+            ],
+            () => {
+                expect(testLink.operations.length).toBe(0);
+                dedupOnOffLink.open();
+                expect(testLink.operations.length).toBe(2);
+                jest.runAllTimers();
+            },
+        );
+    });
+
+    it('releases held deduplicated requests when you open it (all)', () => {
+        const dedupOnOffLink = new QueueLink({keepPolicy: "all"});
+        const myLink = ApolloLink.from([dedupOnOffLink, testLink]);
+        dedupOnOffLink.close();
+        return assertObservableSequence(
+            executeMultiple(myLink, op, op2, op),
+            [
+                { type: 'next', value: testResponse },
+                { type: 'next', value: testResponse2 },
+                { type: 'next', value: testResponse },
+                { type: 'complete' },
+            ],
+            () => {
+                expect(testLink.operations.length).toBe(0);
+                dedupOnOffLink.open();
+                expect(testLink.operations.length).toBe(3);
                 jest.runAllTimers();
             },
         );

src/QueueLink.ts

@@ -14,9 +14,57 @@ interface OperationQueueEntry {
     subscription?: { unsubscribe: () => void };
 }
 
+export type KeepPolicy =
+    | 'first'
+    | 'last'
+    | 'all';
+
+export namespace QueueLink {
+    export interface Options {
+        /**
+         * Decides which entry to keep in the queue in case of duplicate entries.
+         *
+         * Defaults to 'all'.
+         */
+        keepPolicy?: KeepPolicy;
+
+        /**
+         * Specifies which entries are considered duplicates
+         *
+         * Defaults to comparing operation operationA.toKey() === operationB.toKey()
+         * https://www.apollographql.com/docs/link/overview.html
+         */
+        isDuplicate?: (operationA: Operation, operationB: Operation) => boolean;
+    }
+}
+
+const defaultOptions: QueueLink.Options = {
+    keepPolicy: 'all',
+    isDuplicate: (a: Operation, b: Operation) => a.toKey() === b.toKey()
+};
+
+export class DedupedByQueueError extends Error {
+    constructor() {
+        super('Operation got deduplicated by apollo-link-queue.');
+        Object.defineProperty(this, 'name', { value: 'DedupedByQueueError' });
+    }
+}
+
 export default class QueueLink extends ApolloLink {
     private opQueue: OperationQueueEntry[] = [];
     private isOpen: boolean = true;
+    private readonly keepPolicy: KeepPolicy;
+    private readonly isDuplicate: (operationA: Operation, operationB: Operation) => boolean;
+
+    constructor(options: QueueLink.Options = defaultOptions) {
+        super();
+        const {
+            keepPolicy = defaultOptions.keepPolicy,
+            isDuplicate = defaultOptions.isDuplicate
+        } = options;
+        this.keepPolicy = keepPolicy;
+        this.isDuplicate = isDuplicate;
+    }
 
     public open() {
         this.isOpen = true;
@@ -49,6 +97,29 @@ export default class QueueLink extends ApolloLink {
     }
 
     private enqueue(entry: OperationQueueEntry) {
-        this.opQueue.push(entry);
+        const isDuplicate = ({operation}: OperationQueueEntry) => this.isDuplicate(operation, entry.operation);
+        switch (this.keepPolicy) {
+            case "first":
+                const alreadyInQueue = this.opQueue.some(isDuplicate);
+                if (alreadyInQueue) {
+                    // if there is already a duplicate entry the new one gets ignored
+                    entry.observer.error(new DedupedByQueueError());
+                } else {
+                    this.opQueue.push(entry);
+                }
+                break;
+            case "last":
+                const index = this.opQueue.findIndex(isDuplicate);
+                if (index !== -1) {
+                    // if there is already a duplicate entry it gets removed
+                    const [duplicate] = this.opQueue.splice(index, 1);
+                    duplicate.observer.error(new DedupedByQueueError());
+                }
+                this.opQueue.push(entry);
+                break;
+            case "all":
+                this.opQueue.push(entry);
+                break;
+        }
     }
 }

src/TestUtils.ts

@@ -2,10 +2,14 @@ import {
     ApolloLink,
     Operation,
     Observable,
+    execute,
 } from 'apollo-link';
 import {
     ExecutionResult,
 } from 'graphql';
+import {
+    GraphQLRequest
+} from 'apollo-link/src/types';
 
 export class TestLink extends ApolloLink {
     public operations: Operation[];
@@ -39,7 +43,7 @@ export interface Unsubscribable {
 }
 
 export const assertObservableSequence = (
-    observable: Observable<ExecutionResult>,
+    observable: Observable<ExecutionResult | Error>,
     sequence: ObservableValue[],
     initializer: (sub: Unsubscribable) => void = () => undefined,
 ): Promise<boolean | Error> => {
@@ -50,7 +54,8 @@ export const assertObservableSequence = (
     return new Promise((resolve, reject) => {
         const sub = observable.subscribe({
             next: (value) => {
-                expect({ type: 'next', value }).toEqual(sequence[index]);
+                const type = value instanceof Error ? 'error' : 'next';
+                expect({ type, value }).toEqual(sequence[index]);
                 index++;
                 if (index === sequence.length) {
                     resolve(true);
@@ -76,3 +81,17 @@ export const assertObservableSequence = (
         initializer(sub);
     });
 };
+
+export function executeMultiple(link: ApolloLink, ...operations: GraphQLRequest[]) {
+    return Observable.from(operations.map(op => execute(link, op)))
+        .flatMap(o => new Observable(sub => {
+            o.subscribe({
+                next: (v: any) => sub.next(v),
+                error: (e: any) => {
+                    sub.next(e);
+                    sub.complete();
+                },
+                complete: () => sub.complete()
+            })
+        }));
+}