apollo-link-retry - npm Package Compare versions

Comparing version 1.0.2 to 2.0.0

CHANGELOG.md

 
 ### vNext
 
+### 2.0.0
+- Entirely rewritten to address a number of flaws including a new API to prevent DOSing your own server when it may be down. Thanks @nevir for the amazing work!
+
 ### 1.0.2
@@ -5,0 +8,0 @@ - changed peer-dependency of apollo-link to actual dependency

lib/bundle.umd.js

@@ -7,2 +7,29 @@ (function (global, factory) {
 
+function buildDelayFunction(_a) {
+    var _b = _a === void 0 ? {} : _a, _c = _b.initial, initial = _c === void 0 ? 300 : _c, _d = _b.max, max = _d === void 0 ? Infinity : _d, _e = _b.jitter, jitter = _e === void 0 ? true : _e;
+    var baseDelay;
+    if (jitter) {
+        baseDelay = initial;
+    }
+    else {
+        baseDelay = initial / 2;
+    }
+    return function delayFunction(count) {
+        var delay = Math.min(max, baseDelay * Math.pow(2, count));
+        if (jitter) {
+            delay = Math.random() * delay;
+        }
+        return delay;
+    };
+}
+
+function buildRetryFunction(_a) {
+    var _b = _a === void 0 ? {} : _a, _c = _b.max, max = _c === void 0 ? 5 : _c, retryIf = _b.retryIf;
+    return function retryFunction(count, operation, error) {
+        if (count >= max)
+            return false;
+        return retryIf ? retryIf(error) : !!error;
+    };
+}
+
 var __extends = (undefined && undefined.__extends) || (function () {
@@ -18,48 +45,119 @@ var extendStatics = Object.setPrototypeOf ||
 })();
-var operationFnOrNumber = function (prop) {
-    return typeof prop === 'number' ? function () { return prop; } : prop;
-};
-var defaultInterval = function (delay) { return delay; };
+var RetryableOperation = (function () {
+    function RetryableOperation(operation, nextLink, delayFor, retryIf) {
+        var _this = this;
+        this.operation = operation;
+        this.nextLink = nextLink;
+        this.delayFor = delayFor;
+        this.retryIf = retryIf;
+        this.retryCount = 0;
+        this.values = [];
+        this.complete = false;
+        this.canceled = false;
+        this.observers = [];
+        this.currentSubscription = null;
+        this.onNext = function (value) {
+            _this.values.push(value);
+            for (var _i = 0, _a = _this.observers; _i < _a.length; _i++) {
+                var observer = _a[_i];
+                observer.next(value);
+            }
+        };
+        this.onComplete = function () {
+            _this.complete = true;
+            for (var _i = 0, _a = _this.observers; _i < _a.length; _i++) {
+                var observer = _a[_i];
+                observer.complete();
+            }
+        };
+        this.onError = function (error) {
+            _this.retryCount += 1;
+            if (_this.retryIf(_this.retryCount, _this.operation, error)) {
+                _this.scheduleRetry(_this.delayFor(_this.retryCount, _this.operation, error));
+                return;
+            }
+            _this.error = error;
+            for (var _i = 0, _a = _this.observers; _i < _a.length; _i++) {
+                var observer = _a[_i];
+                observer.error(error);
+            }
+        };
+    }
+    RetryableOperation.prototype.subscribe = function (observer) {
+        if (this.canceled) {
+            throw new Error("Subscribing to a retryable link that was canceled is not supported");
+        }
+        this.observers.push(observer);
+        for (var _i = 0, _a = this.values; _i < _a.length; _i++) {
+            var value = _a[_i];
+            observer.next(value);
+        }
+        if (this.complete) {
+            observer.complete();
+        }
+        else if (this.error) {
+            observer.error(this.error);
+        }
+    };
+    RetryableOperation.prototype.unsubscribe = function (observer) {
+        var index = this.observers.indexOf(observer);
+        if (index < 0) {
+            throw new Error("RetryLink BUG! Attempting to unsubscribe unknown observer!");
+        }
+        this.observers[index] = null;
+        if (this.observers.every(function (o) { return o === null; })) {
+            this.cancel();
+        }
+    };
+    RetryableOperation.prototype.start = function () {
+        if (this.currentSubscription)
+            return;
+        this.try();
+    };
+    RetryableOperation.prototype.cancel = function () {
+        if (this.currentSubscription) {
+            this.currentSubscription.unsubscribe();
+        }
+        clearTimeout(this.timerId);
+        this.timerId = null;
+        this.currentSubscription = null;
+        this.canceled = true;
+    };
+    RetryableOperation.prototype.try = function () {
+        this.currentSubscription = this.nextLink(this.operation).subscribe({
+            next: this.onNext,
+            error: this.onError,
+            complete: this.onComplete,
+        });
+    };
+    RetryableOperation.prototype.scheduleRetry = function (delay) {
+        var _this = this;
+        if (this.timerId) {
+            throw new Error("RetryLink BUG! Encountered overlapping retries");
+        }
+        this.timerId = setTimeout(function () {
+            _this.timerId = null;
+            _this.try();
+        }, delay);
+    };
+    return RetryableOperation;
+}());
 var RetryLink = (function (_super) {
     __extends(RetryLink, _super);
-    function RetryLink(params) {
+    function RetryLink(_a) {
+        var _b = _a === void 0 ? {} : _a, delay = _b.delay, attempts = _b.attempts;
         var _this = _super.call(this) || this;
-        _this.subscriptions = {};
-        _this.timers = {};
-        _this.counts = {};
-        _this.max = operationFnOrNumber((params && params.max) || 10);
-        _this.delay = operationFnOrNumber((params && params.delay) || 300);
-        _this.interval = (params && params.interval) || defaultInterval;
+        _this.delayFor =
+            typeof delay === 'function' ? delay : buildDelayFunction(delay);
+        _this.retryIf =
+            typeof attempts === 'function' ? attempts : buildRetryFunction(attempts);
         return _this;
     }
-    RetryLink.prototype.request = function (operation, forward) {
-        var _this = this;
-        var key = operation.toKey();
-        if (!this.counts[key])
-            this.counts[key] = 0;
+    RetryLink.prototype.request = function (operation, nextLink) {
+        var retryable = new RetryableOperation(operation, nextLink, this.delayFor, this.retryIf);
+        retryable.start();
         return new apolloLink.Observable(function (observer) {
-            var subscriber = {
-                next: function (data) {
-                    _this.counts[key] = 0;
-                    observer.next(data);
-                },
-                error: function (error) {
-                    _this.counts[key]++;
-                    if (_this.counts[key] < _this.max(operation)) {
-                        _this.timers[key] = setTimeout(function () {
-                            var observable = forward(operation);
-                            _this.subscriptions[key] = observable.subscribe(subscriber);
-                        }, _this.interval(_this.delay(operation), _this.counts[key]));
-                    }
-                    else {
-                        observer.error(error);
-                    }
-                },
-                complete: observer.complete.bind(observer),
-            };
-            _this.subscriptions[key] = forward(operation).subscribe(subscriber);
+            retryable.subscribe(observer);
             return function () {
-                _this.subscriptions[key].unsubscribe();
-                if (_this.timers[key])
-                    clearTimeout(_this.timers[key]);
+                retryable.unsubscribe(observer);
             };
@@ -66,0 +164,0 @@ });

lib/retryLink.d.ts

 /// <reference types="zen-observable" />
 import { ApolloLink, Observable, Operation, NextLink, FetchResult } from 'apollo-link';
-export declare type ParamFnOrNumber = (operation: Operation) => number | number;
+import { DelayFunction, DelayFunctionOptions } from './delayFunction';
+import { RetryFunction, RetryFunctionOptions } from './retryFunction';
+export declare namespace RetryLink {
+    interface Options {
+        delay?: DelayFunctionOptions | DelayFunction;
+        attempts?: RetryFunctionOptions | RetryFunction;
+    }
+}
 export declare class RetryLink extends ApolloLink {
-    private delay;
-    private max;
-    private interval;
-    private subscriptions;
-    private timers;
-    private counts;
-    constructor(params?: {
-        max?: ParamFnOrNumber;
-        delay?: ParamFnOrNumber;
-        interval?: (delay: number, count: number) => number;
-    });
-    request(operation: Operation, forward: NextLink): Observable<FetchResult>;
+    private delayFor;
+    private retryIf;
+    constructor({delay, attempts}?: RetryLink.Options);
+    request(operation: Operation, nextLink: NextLink): Observable<FetchResult>;
 }

lib/retryLink.js

@@ -12,48 +12,121 @@ var __extends = (this && this.__extends) || (function () {
 import { ApolloLink, Observable, } from 'apollo-link';
-var operationFnOrNumber = function (prop) {
-    return typeof prop === 'number' ? function () { return prop; } : prop;
-};
-var defaultInterval = function (delay) { return delay; };
+import { buildDelayFunction, } from './delayFunction';
+import { buildRetryFunction, } from './retryFunction';
+var RetryableOperation = (function () {
+    function RetryableOperation(operation, nextLink, delayFor, retryIf) {
+        var _this = this;
+        this.operation = operation;
+        this.nextLink = nextLink;
+        this.delayFor = delayFor;
+        this.retryIf = retryIf;
+        this.retryCount = 0;
+        this.values = [];
+        this.complete = false;
+        this.canceled = false;
+        this.observers = [];
+        this.currentSubscription = null;
+        this.onNext = function (value) {
+            _this.values.push(value);
+            for (var _i = 0, _a = _this.observers; _i < _a.length; _i++) {
+                var observer = _a[_i];
+                observer.next(value);
+            }
+        };
+        this.onComplete = function () {
+            _this.complete = true;
+            for (var _i = 0, _a = _this.observers; _i < _a.length; _i++) {
+                var observer = _a[_i];
+                observer.complete();
+            }
+        };
+        this.onError = function (error) {
+            _this.retryCount += 1;
+            if (_this.retryIf(_this.retryCount, _this.operation, error)) {
+                _this.scheduleRetry(_this.delayFor(_this.retryCount, _this.operation, error));
+                return;
+            }
+            _this.error = error;
+            for (var _i = 0, _a = _this.observers; _i < _a.length; _i++) {
+                var observer = _a[_i];
+                observer.error(error);
+            }
+        };
+    }
+    RetryableOperation.prototype.subscribe = function (observer) {
+        if (this.canceled) {
+            throw new Error("Subscribing to a retryable link that was canceled is not supported");
+        }
+        this.observers.push(observer);
+        for (var _i = 0, _a = this.values; _i < _a.length; _i++) {
+            var value = _a[_i];
+            observer.next(value);
+        }
+        if (this.complete) {
+            observer.complete();
+        }
+        else if (this.error) {
+            observer.error(this.error);
+        }
+    };
+    RetryableOperation.prototype.unsubscribe = function (observer) {
+        var index = this.observers.indexOf(observer);
+        if (index < 0) {
+            throw new Error("RetryLink BUG! Attempting to unsubscribe unknown observer!");
+        }
+        this.observers[index] = null;
+        if (this.observers.every(function (o) { return o === null; })) {
+            this.cancel();
+        }
+    };
+    RetryableOperation.prototype.start = function () {
+        if (this.currentSubscription)
+            return;
+        this.try();
+    };
+    RetryableOperation.prototype.cancel = function () {
+        if (this.currentSubscription) {
+            this.currentSubscription.unsubscribe();
+        }
+        clearTimeout(this.timerId);
+        this.timerId = null;
+        this.currentSubscription = null;
+        this.canceled = true;
+    };
+    RetryableOperation.prototype.try = function () {
+        this.currentSubscription = this.nextLink(this.operation).subscribe({
+            next: this.onNext,
+            error: this.onError,
+            complete: this.onComplete,
+        });
+    };
+    RetryableOperation.prototype.scheduleRetry = function (delay) {
+        var _this = this;
+        if (this.timerId) {
+            throw new Error("RetryLink BUG! Encountered overlapping retries");
+        }
+        this.timerId = setTimeout(function () {
+            _this.timerId = null;
+            _this.try();
+        }, delay);
+    };
+    return RetryableOperation;
+}());
 var RetryLink = (function (_super) {
     __extends(RetryLink, _super);
-    function RetryLink(params) {
+    function RetryLink(_a) {
+        var _b = _a === void 0 ? {} : _a, delay = _b.delay, attempts = _b.attempts;
         var _this = _super.call(this) || this;
-        _this.subscriptions = {};
-        _this.timers = {};
-        _this.counts = {};
-        _this.max = operationFnOrNumber((params && params.max) || 10);
-        _this.delay = operationFnOrNumber((params && params.delay) || 300);
-        _this.interval = (params && params.interval) || defaultInterval;
+        _this.delayFor =
+            typeof delay === 'function' ? delay : buildDelayFunction(delay);
+        _this.retryIf =
+            typeof attempts === 'function' ? attempts : buildRetryFunction(attempts);
         return _this;
     }
-    RetryLink.prototype.request = function (operation, forward) {
-        var _this = this;
-        var key = operation.toKey();
-        if (!this.counts[key])
-            this.counts[key] = 0;
+    RetryLink.prototype.request = function (operation, nextLink) {
+        var retryable = new RetryableOperation(operation, nextLink, this.delayFor, this.retryIf);
+        retryable.start();
         return new Observable(function (observer) {
-            var subscriber = {
-                next: function (data) {
-                    _this.counts[key] = 0;
-                    observer.next(data);
-                },
-                error: function (error) {
-                    _this.counts[key]++;
-                    if (_this.counts[key] < _this.max(operation)) {
-                        _this.timers[key] = setTimeout(function () {
-                            var observable = forward(operation);
-                            _this.subscriptions[key] = observable.subscribe(subscriber);
-                        }, _this.interval(_this.delay(operation), _this.counts[key]));
-                    }
-                    else {
-                        observer.error(error);
-                    }
-                },
-                complete: observer.complete.bind(observer),
-            };
-            _this.subscriptions[key] = forward(operation).subscribe(subscriber);
+            retryable.subscribe(observer);
             return function () {
-                _this.subscriptions[key].unsubscribe();
-                if (_this.timers[key])
-                    clearTimeout(_this.timers[key]);
+                retryable.unsubscribe(observer);
             };
@@ -60,0 +133,0 @@ });

package.json

 {
   "name": "apollo-link-retry",
-  "version": "1.0.2",
+  "version": "2.0.0",
   "description": "Retry Apollo Link for GraphQL Network Stack",
@@ -45,3 +45,3 @@ "author": "Evans Hauser <[e-mail]>",
     "@types/zen-observable": "0.5.3",
-    "apollo-link": "^1.0.4"
+    "apollo-link": "^1.0.6"
   },
@@ -56,7 +56,8 @@ "devDependencies": {
     "rimraf": "2.6.1",
-    "rollup": "0.52.0",
-    "ts-jest": "21.2.3",
+    "rollup": "0.52.1",
+    "ts-jest": "21.2.4",
     "tslint": "5.8.0",
     "typescript": "2.6.2",
-    "uglify-js": "3.2.0"
+    "uglify-js": "3.2.1",
+    "wait-for-observables": "1.0.3"
   },
@@ -63,0 +64,0 @@ "jest": {

README.md

@@ -6,10 +6,16 @@ ---
 ## Purpose
-An Apollo Link to allow multiple attempts when an operation has failed. One such use case is to try a request while a network connection is offline and retry until it comes back online. You can configure a RetryLink to vary the number of times it retries and how long it waits between retries through its configuration.
 
+An Apollo Link to allow multiple attempts when an operation has failed, due to network or server errors. `RetryLink` provides exponential backoff, and jitters delays between attempts by default. It does not (currently) support retries for GraphQL errors.
+
+One such use case is to try a request while a network connection is offline and retry until it comes back online.
+
 ## Installation
 
-`npm install apollo-link-retry --save`
+```sh
+npm install apollo-link-retry --save
+```
 
 ## Usage
-```js
+
+```ts
 import { RetryLink } from "apollo-link-retry";
@@ -21,29 +27,59 @@
 ## Options
-Retry Link takes an object with three options on it to customize the behavior of the link.
 
-Retry Link retries on network errors only, not on GraphQL errors.
+The standard retry strategy provides exponential backoff with jittering, and takes the following options, grouped into `delay` and `attempt` strategies:
 
-The default delay algorithm is to wait `delay` ms between each retry. You can customize the algorithm (eg, replacing with exponential backoff) with the `interval` option. The possible values for the configuration object are as follow:
-- `max`: a number or function matching (Operation => number) to determine the max number of times to try a single operation before giving up. It defaults to 10
-- `delay`: a number or function matching (Operation => number) to input to the interval function below: Defaults to 300 ms
-- `interval`: a function matching (delay: number, count: number) => number which is the amount of time (in ms) to wait before the next attempt; count is the number of requests previously tried
+- `delay.initial`: The number of milliseconds to wait before attempting the first retry.
 
-```js
-import { RetryLink } from "apollo-link-retry";
+- `delay.max`: The maximum number of milliseconds that the link should wait for any retry.
 
-const max = (operation) => operation.getContext().max;
-const delay = 5000;
-const interval = (delay, count) => {
-  if (count > 5) return 10000;
-  return delay;
-}
+- `delay.jitter`: Whether delays between attempts should be randomized.
 
-const link = new RetryLink({
-  max,
-  delay,
-  interval
+- `attempts.max`: The max number of times to try a single operation before giving up.
+
+- `attempts.retryIf`: A predicate function that can determine whether a particular response should be retried.
+
+The default configuration is equivalent to:
+
+```ts
+new RetryLink({
+  delay: {
+    initial: 300,
+    max: Infinity,
+    jitter: true,
+  },
+  attempts: {
+    max: 5,
+    retryIf: (_count, _operation, error) => !!error,
+  },
 });
 ```
 
+### On Exponential Backoff & Jitter
+
+Starting with `initialDelay`, the delay of each subsequent retry is increased exponentially (by a power of 2).  For example, if `initialDelay` is 100, additional retries will occur after delays of 200, 400, 800, etc.
+
+Additionally, with `jitter` enabled, delays are randomized anywhere between 0ms (instant), and 2x the configured delay so that, on average, they should occur at the same intervals.
+
+These two features combined help alleviate [the thundering herd problem](https://en.wikipedia.org/wiki/Thundering_herd_problem), by distributing load during major outages.
+
+### Custom Strategies
+
+Instead of the options object, you may pass a function for `delay` and/or `attempts`, which implement custom strategies for each.  In both cases the function is given the same arguments (`count`, `operation`, `error`).
+
+The `attempts` function should return a boolean indicating whether the response should be retried.  If yes, the `delay` function is then called, and should return the number of milliseconds to delay by.
+
+```ts
+import { RetryLink } from "apollo-link-retry";
+
+const link = new RetryLink(
+  attempts: (count, operation, error) => {
+    return !!error && operation.operationName != 'specialCase';
+  },
+  delay: (count, operation, error) => {
+    return count * 1000 * Math.random();
+  },
+});
+```
+
 ## Context
 The Retry Link does not use the context for anything.

src/__tests__/retryLink.ts

 import gql from 'graphql-tag';
 import { execute, ApolloLink, Observable, FetchResult } from 'apollo-link';
+import waitFor from 'wait-for-observables';
 
@@ -14,77 +15,153 @@ import { RetryLink } from '../retryLink';
 
+const standardError = new Error('I never work');
+
 describe('RetryLink', () => {
-  it('should fail with unreachable endpoint', done => {
+  it('fails for unreachable endpoints', async () => {
     const max = 10;
-    const retry = new RetryLink({ delay: 1, max });
-    const error = new Error('I never work');
-    const stub = jest.fn(() => {
-      return new Observable(observer => observer.error(error));
-    });
+    const retry = new RetryLink({ delay: { initial: 1 }, attempts: { max } });
+    const stub = jest.fn(() => new Observable(o => o.error(standardError)));
+    const link = ApolloLink.from([retry, stub]);
 
+    const [{ error }] = await waitFor(execute(link, { query }));
+    expect(error).toEqual(standardError);
+    expect(stub).toHaveBeenCalledTimes(max);
+  });
+
+  it('returns data from the underlying link on a successful operation', async () => {
+    const retry = new RetryLink();
+    const data = { data: { hello: 'world' } };
+    const stub = jest.fn(() => Observable.of(data));
     const link = ApolloLink.from([retry, stub]);
 
-    execute(link, { query }).subscribe(
-      () => {
-        throw new Error();
-      },
-      actualError => {
-        expect(stub).toHaveBeenCalledTimes(max);
-        expect(error).toEqual(actualError);
-        done();
-      },
-      () => {
-        throw new Error();
-      },
-    );
+    const [{ values }] = await waitFor(execute(link, { query }));
+    expect(values).toEqual([data]);
+    expect(stub).toHaveBeenCalledTimes(1);
   });
 
-  it('should return data from the underlying link on a successful operation', done => {
-    const retry = new RetryLink();
-    const data = <FetchResult>{
-      data: {
-        hello: 'world',
-      },
-    };
+  it('returns data from the underlying link on a successful retry', async () => {
+    const retry = new RetryLink({
+      delay: { initial: 1 },
+      attempts: { max: 2 },
+    });
+    const data = { data: { hello: 'world' } };
     const stub = jest.fn();
-    stub.mockReturnValue(Observable.of(data));
+    stub.mockReturnValueOnce(new Observable(o => o.error(standardError)));
+    stub.mockReturnValueOnce(Observable.of(data));
+    const link = ApolloLink.from([retry, stub]);
 
+    const [{ values }] = await waitFor(execute(link, { query }));
+    expect(values).toEqual([data]);
+    expect(stub).toHaveBeenCalledTimes(2);
+  });
+
+  it('calls unsubscribe on the appropriate downstream observable', async () => {
+    const retry = new RetryLink({
+      delay: { initial: 1 },
+      attempts: { max: 2 },
+    });
+    const data = { data: { hello: 'world' } };
+    const unsubscribeStub = jest.fn();
+
+    const firstTry = new Observable(o => o.error(standardError));
+    // Hold the test hostage until we're hit
+    let secondTry;
+    const untilSecondTry = new Promise(resolve => {
+      secondTry = {
+        subscribe(observer) {
+          resolve(); // Release hold on test.
+
+          Promise.resolve().then(() => {
+            observer.next(data);
+            observer.complete();
+          });
+          return { unsubscribe: unsubscribeStub };
+        },
+      };
+    });
+
+    const stub = jest.fn();
+    stub.mockReturnValueOnce(firstTry);
+    stub.mockReturnValueOnce(secondTry);
     const link = ApolloLink.from([retry, stub]);
 
-    execute(link, { query }).subscribe(
-      actualData => {
-        expect(stub).toHaveBeenCalledTimes(1);
-        expect(data).toEqual(actualData);
-      },
-      () => {
-        throw new Error();
-      },
-      done,
-    );
+    const subscription = execute(link, { query }).subscribe({});
+    await untilSecondTry;
+    subscription.unsubscribe();
+    expect(unsubscribeStub).toHaveBeenCalledTimes(1);
   });
 
-  it('should return data from the underlying link on a successful retry', done => {
-    const retry = new RetryLink({ delay: 1, max: 2 });
-    const error = new Error('I never work');
-    const data = <FetchResult>{
-      data: {
-        hello: 'world',
-      },
-    };
+  it('supports multiple subscribers to the same request', async () => {
+    const retry = new RetryLink({
+      delay: { initial: 1 },
+      attempts: { max: 5 },
+    });
+    const data = { data: { hello: 'world' } };
     const stub = jest.fn();
-    stub.mockReturnValueOnce(new Observable(observer => observer.error(error)));
+    stub.mockReturnValueOnce(new Observable(o => o.error(standardError)));
+    stub.mockReturnValueOnce(new Observable(o => o.error(standardError)));
     stub.mockReturnValueOnce(Observable.of(data));
+    const link = ApolloLink.from([retry, stub]);
 
+    const observable = execute(link, { query });
+    const [result1, result2] = await waitFor(observable, observable);
+    expect(result1.values).toEqual([data]);
+    expect(result2.values).toEqual([data]);
+    expect(stub).toHaveBeenCalledTimes(3);
+  });
+
+  it('retries independently for concurrent requests', async () => {
+    const retry = new RetryLink({
+      delay: { initial: 1 },
+      attempts: { max: 5 },
+    });
+    const data = { data: { hello: 'world' } };
+    const stub = jest.fn(() => new Observable(o => o.error(standardError)));
     const link = ApolloLink.from([retry, stub]);
 
-    execute(link, { query }).subscribe(
-      actualData => {
-        expect(stub).toHaveBeenCalledTimes(2);
-        expect(data).toEqual(actualData);
-      },
-      () => {
-        throw new Error();
-      },
-      done,
+    const [result1, result2] = await waitFor(
+      execute(link, { query }),
+      execute(link, { query }),
     );
+    expect(result1.error).toEqual(standardError);
+    expect(result2.error).toEqual(standardError);
+    expect(stub).toHaveBeenCalledTimes(10);
   });
+
+  it('supports custom delay functions', async () => {
+    const delayStub = jest.fn(() => 1);
+    const retry = new RetryLink({ delay: delayStub, attempts: { max: 3 } });
+    const linkStub = jest.fn(() => new Observable(o => o.error(standardError)));
+    const link = ApolloLink.from([retry, linkStub]);
+    const [{ error }] = await waitFor(execute(link, { query }));
+
+    expect(error).toEqual(standardError);
+    const operation = delayStub.mock.calls[0][1];
+    expect(delayStub.mock.calls).toEqual([
+      [1, operation, standardError],
+      [2, operation, standardError],
+    ]);
+  });
+
+  it('supports custom attempt functions', async () => {
+    const attemptStub = jest.fn();
+    attemptStub.mockReturnValueOnce(true);
+    attemptStub.mockReturnValueOnce(true);
+    attemptStub.mockReturnValueOnce(false);
+
+    const retry = new RetryLink({
+      delay: { initial: 1 },
+      attempts: attemptStub,
+    });
+    const linkStub = jest.fn(() => new Observable(o => o.error(standardError)));
+    const link = ApolloLink.from([retry, linkStub]);
+    const [{ error }] = await waitFor(execute(link, { query }));
+
+    expect(error).toEqual(standardError);
+    const operation = attemptStub.mock.calls[0][1];
+    expect(attemptStub.mock.calls).toEqual([
+      [1, operation, standardError],
+      [2, operation, standardError],
+      [3, operation, standardError],
+    ]);
+  });
 });

src/retryLink.ts

@@ -9,26 +9,177 @@ import {
 
-const operationFnOrNumber = prop =>
-  typeof prop === 'number' ? () => prop : prop;
+import {
+  DelayFunction,
+  DelayFunctionOptions,
+  buildDelayFunction,
+} from './delayFunction';
+import {
+  RetryFunction,
+  RetryFunctionOptions,
+  buildRetryFunction,
+} from './retryFunction';
 
-const defaultInterval = delay => delay;
+export namespace RetryLink {
+  export interface Options {
+    /**
+     * Configuration for the delay strategy to use, or a custom delay strategy.
+     */
+    delay?: DelayFunctionOptions | DelayFunction;
 
-export type ParamFnOrNumber = (operation: Operation) => number | number;
+    /**
+     * Configuration for the retry strategy to use, or a custom retry strategy.
+     */
+    attempts?: RetryFunctionOptions | RetryFunction;
+  }
+}
 
+/**
+ * Tracking and management of operations that may be (or currently are) retried.
+ */
+class RetryableOperation<TValue = any> {
+  private retryCount: number = 0;
+  private values: any[] = [];
+  private error: any;
+  private complete = false;
+  private canceled = false;
+  private observers: ZenObservable.Observer<TValue>[] = [];
+  private currentSubscription: ZenObservable.Subscription = null;
+  private timerId: number;
+
+  constructor(
+    private operation: Operation,
+    private nextLink: NextLink,
+    private delayFor: DelayFunction,
+    private retryIf: RetryFunction,
+  ) {}
+
+  /**
+   * Register a new observer for this operation.
+   *
+   * If the operation has previously emitted other events, they will be
+   * immediately triggered for the observer.
+   */
+  subscribe(observer: ZenObservable.Observer<TValue>) {
+    if (this.canceled) {
+      throw new Error(
+        `Subscribing to a retryable link that was canceled is not supported`,
+      );
+    }
+    this.observers.push(observer);
+
+    // If we've already begun, catch this observer up.
+    for (const value of this.values) {
+      observer.next(value);
+    }
+
+    if (this.complete) {
+      observer.complete();
+    } else if (this.error) {
+      observer.error(this.error);
+    }
+  }
+
+  /**
+   * Remove a previously registered observer from this operation.
+   *
+   * If no observers remain, the operation will stop retrying, and unsubscribe
+   * from its downstream link.
+   */
+  unsubscribe(observer: ZenObservable.Observer<TValue>) {
+    const index = this.observers.indexOf(observer);
+    if (index < 0) {
+      throw new Error(
+        `RetryLink BUG! Attempting to unsubscribe unknown observer!`,
+      );
+    }
+    // Note that we are careful not to change the order of length of the array,
+    // as we are often mid-iteration when calling this method.
+    this.observers[index] = null;
+
+    // If this is the last observer, we're done.
+    if (this.observers.every(o => o === null)) {
+      this.cancel();
+    }
+  }
+
+  /**
+   * Start the initial request.
+   */
+  start() {
+    if (this.currentSubscription) return; // Already started.
+
+    this.try();
+  }
+
+  /**
+   * Stop retrying for the operation, and cancel any in-progress requests.
+   */
+  cancel() {
+    if (this.currentSubscription) {
+      this.currentSubscription.unsubscribe();
+    }
+    clearTimeout(this.timerId);
+    this.timerId = null;
+    this.currentSubscription = null;
+    this.canceled = true;
+  }
+
+  private try() {
+    this.currentSubscription = this.nextLink(this.operation).subscribe({
+      next: this.onNext,
+      error: this.onError,
+      complete: this.onComplete,
+    });
+  }
+
+  private onNext = (value: any) => {
+    this.values.push(value);
+    for (const observer of this.observers) {
+      observer.next(value);
+    }
+  };
+
+  private onComplete = () => {
+    this.complete = true;
+    for (const observer of this.observers) {
+      observer.complete();
+    }
+  };
+
+  private onError = error => {
+    this.retryCount += 1;
+
+    // Should we retry?
+    if (this.retryIf(this.retryCount, this.operation, error)) {
+      this.scheduleRetry(this.delayFor(this.retryCount, this.operation, error));
+      return;
+    }
+
+    this.error = error;
+    for (const observer of this.observers) {
+      observer.error(error);
+    }
+  };
+
+  private scheduleRetry(delay) {
+    if (this.timerId) {
+      throw new Error(`RetryLink BUG! Encountered overlapping retries`);
+    }
+
+    this.timerId = setTimeout(() => {
+      this.timerId = null;
+      this.try();
+    }, delay);
+  }
+}
+
 export class RetryLink extends ApolloLink {
-  private delay: ParamFnOrNumber;
-  private max: ParamFnOrNumber;
-  private interval: (delay: number, count: number) => number;
-  private subscriptions: { [key: string]: ZenObservable.Subscription } = {};
-  private timers = {};
-  private counts: { [key: string]: number } = {};
+  private delayFor: DelayFunction;
+  private retryIf: RetryFunction;
 
-  constructor(params?: {
-    max?: ParamFnOrNumber;
-    delay?: ParamFnOrNumber;
-    interval?: (delay: number, count: number) => number;
-  }) {
+  constructor({ delay, attempts }: RetryLink.Options = {}) {
     super();
-    this.max = operationFnOrNumber((params && params.max) || 10);
-    this.delay = operationFnOrNumber((params && params.delay) || 300);
-    this.interval = (params && params.interval) || defaultInterval;
+    this.delayFor =
+      typeof delay === 'function' ? delay : buildDelayFunction(delay);
+    this.retryIf =
+      typeof attempts === 'function' ? attempts : buildRetryFunction(attempts);
   }
@@ -38,31 +189,16 @@
     operation: Operation,
-    forward: NextLink,
+    nextLink: NextLink,
   ): Observable<FetchResult> {
-    const key = operation.toKey();
-    if (!this.counts[key]) this.counts[key] = 0;
+    const retryable = new RetryableOperation(
+      operation,
+      nextLink,
+      this.delayFor,
+      this.retryIf,
+    );
+    retryable.start();
+
     return new Observable(observer => {
-      const subscriber = {
-        next: data => {
-          this.counts[key] = 0;
-          observer.next(data);
-        },
-        error: error => {
-          this.counts[key]++;
-          if (this.counts[key] < this.max(operation)) {
-            this.timers[key] = setTimeout(() => {
-              const observable = forward(operation);
-              this.subscriptions[key] = observable.subscribe(subscriber);
-            }, this.interval(this.delay(operation), this.counts[key]));
-          } else {
-            observer.error(error);
-          }
-        },
-        complete: observer.complete.bind(observer),
-      };
-
-      this.subscriptions[key] = forward(operation).subscribe(subscriber);
-
+      retryable.subscribe(observer);
       return () => {
-        this.subscriptions[key].unsubscribe();
-        if (this.timers[key]) clearTimeout(this.timers[key]);
+        retryable.unsubscribe(observer);
       };
@@ -69,0 +205,0 @@ });

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Major refactor

Supply chain risk

Package has recently undergone a major refactor. It may be unstable or indicate significant internal changes. Use caution when updating to versions that include significant changes.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

54454

increased by134.79%

Number of package files

26

increased by62.5%

Lines of code

886

increased by192.41%

Number of lines in readme file

84

increased by75%

Worsened metrics

Dev dependency count

13

increased by8.33%

Number of medium supply chain risk alerts

1

increased byInfinity%

Dependency changes

• Updatedapollo-link@^1.0.6