apollo-link-batch - npm Package Compare versions

Comparing version 1.0.5 to 1.1.0

CHANGELOG.md

@@ -0,4 +1,9 @@
+# Change log
 
 ### vNext
 
+### 1.1.0
+- add input validation logic [PR#364](https://github.com/apollographql/apollo-link/pull/364)
+- ensure batch observables can handle multiple subscribers [PR#364](https://github.com/apollographql/apollo-link/pull/364)
+
 ### 1.0.4
@@ -5,0 +10,0 @@ - ApolloLink upgrade

lib/batching.d.ts

@@ -1,4 +0,3 @@
-/// <reference types="zen-observable" />
 import { Observable, Operation, NextLink, FetchResult } from 'apollo-link';
-export declare type BatchHandler = (operations: Operation[], forward: (NextLink | undefined)[]) => Observable<FetchResult[]> | null;
+export declare type BatchHandler = (operations: Operation[], forward?: (NextLink | undefined)[]) => Observable<FetchResult[]> | null;
 export interface BatchableRequest {
@@ -8,19 +7,21 @@ operation: Operation;
     observable?: Observable<FetchResult>;
-    next?: (result: FetchResult) => void;
-    error?: (error: Error) => void;
-    complete?: () => void;
+    next?: Array<(result: FetchResult) => void>;
+    error?: Array<(error: Error) => void>;
+    complete?: Array<() => void>;
 }
 export declare class OperationBatcher {
-    queuedRequests: BatchableRequest[];
+    queuedRequests: Map<string, BatchableRequest[]>;
     private batchInterval;
     private batchMax;
     private batchHandler;
-    constructor({batchInterval, batchMax, batchHandler}: {
+    private batchKey;
+    constructor({batchInterval, batchMax, batchHandler, batchKey}: {
         batchInterval: number;
         batchMax?: number;
         batchHandler: BatchHandler;
+        batchKey?: (Operation) => string;
     });
     enqueueRequest(request: BatchableRequest): Observable<FetchResult>;
-    consumeQueue(): (Observable<FetchResult> | undefined)[] | undefined;
-    private scheduleQueueConsumption();
+    consumeQueue(key?: string): (Observable<FetchResult> | undefined)[] | undefined;
+    private scheduleQueueConsumption(key?);
 }

lib/batching.js

@@ -12,8 +12,8 @@ var __assign = (this && this.__assign) || Object.assign || function(t) {
     function OperationBatcher(_a) {
-        var batchInterval = _a.batchInterval, _b = _a.batchMax, batchMax = _b === void 0 ? 0 : _b, batchHandler = _a.batchHandler;
-        this.queuedRequests = [];
-        this.queuedRequests = [];
+        var batchInterval = _a.batchInterval, _b = _a.batchMax, batchMax = _b === void 0 ? 0 : _b, batchHandler = _a.batchHandler, _c = _a.batchKey, batchKey = _c === void 0 ? function () { return ''; } : _c;
+        this.queuedRequests = new Map();
         this.batchInterval = batchInterval;
         this.batchMax = batchMax;
         this.batchHandler = batchHandler;
+        this.batchKey = batchKey;
     }
@@ -23,22 +23,41 @@ OperationBatcher.prototype.enqueueRequest = function (request) {
         var requestCopy = __assign({}, request);
-        requestCopy.observable =
-            requestCopy.observable ||
-                new Observable(function (observer) {
-                    _this.queuedRequests.push(requestCopy);
-                    requestCopy.next = requestCopy.next || observer.next.bind(observer);
-                    requestCopy.error = requestCopy.error || observer.error.bind(observer);
-                    requestCopy.complete =
-                        requestCopy.complete || observer.complete.bind(observer);
-                    if (_this.queuedRequests.length === 1) {
-                        _this.scheduleQueueConsumption();
-                    }
-                    if (_this.queuedRequests.length === _this.batchMax) {
-                        _this.consumeQueue();
-                    }
-                });
+        var queued = false;
+        var key = this.batchKey(request.operation);
+        if (!requestCopy.observable) {
+            requestCopy.observable = new Observable(function (observer) {
+                if (!_this.queuedRequests.has(key)) {
+                    _this.queuedRequests.set(key, []);
+                }
+                if (!queued) {
+                    _this.queuedRequests.get(key).push(requestCopy);
+                    queued = true;
+                }
+                requestCopy.next = requestCopy.next || [];
+                if (observer.next)
+                    requestCopy.next.push(observer.next.bind(observer));
+                requestCopy.error = requestCopy.error || [];
+                if (observer.error)
+                    requestCopy.error.push(observer.error.bind(observer));
+                requestCopy.complete = requestCopy.complete || [];
+                if (observer.complete)
+                    requestCopy.complete.push(observer.complete.bind(observer));
+                if (_this.queuedRequests.get(key).length === 1) {
+                    _this.scheduleQueueConsumption(key);
+                }
+                if (_this.queuedRequests.get(key).length === _this.batchMax) {
+                    _this.consumeQueue(key);
+                }
+            });
+        }
         return requestCopy.observable;
     };
-    OperationBatcher.prototype.consumeQueue = function () {
-        var requests = this.queuedRequests.map(function (queuedRequest) { return queuedRequest.operation; });
-        var forwards = this.queuedRequests.map(function (queuedRequest) { return queuedRequest.forward; });
+    OperationBatcher.prototype.consumeQueue = function (key) {
+        if (key === void 0) { key = ''; }
+        var queuedRequests = this.queuedRequests.get(key);
+        if (!queuedRequests) {
+            return;
+        }
+        this.queuedRequests.delete(key);
+        var requests = queuedRequests.map(function (queuedRequest) { return queuedRequest.operation; });
+        var forwards = queuedRequests.map(function (queuedRequest) { return queuedRequest.forward; });
         var observables = [];
@@ -48,3 +67,3 @@ var nexts = [];
         var completes = [];
-        this.queuedRequests.forEach(function (batchableRequest, index) {
+        queuedRequests.forEach(function (batchableRequest, index) {
             observables.push(batchableRequest.observable);
@@ -55,23 +74,32 @@ nexts.push(batchableRequest.next);
         });
-        this.queuedRequests = [];
         var batchedObservable = this.batchHandler(requests, forwards) || Observable.of();
+        var onError = function (error) {
+            errors.forEach(function (rejecters) {
+                if (rejecters) {
+                    rejecters.forEach(function (e) { return e(error); });
+                }
+            });
+        };
         batchedObservable.subscribe({
             next: function (results) {
+                if (!Array.isArray(results)) {
+                    results = [results];
+                }
+                if (nexts.length !== results.length) {
+                    var error = new Error("server returned results with length " + results.length + ", expected length of " + nexts.length);
+                    error.result = results;
+                    return onError(error);
+                }
                 results.forEach(function (result, index) {
+                    requests[index].setContext({ response: result });
                     if (nexts[index]) {
-                        nexts[index](result);
+                        nexts[index].forEach(function (next) { return next(result); });
                     }
                 });
             },
-            error: function (error) {
-                errors.forEach(function (rejecter, index) {
-                    if (errors[index]) {
-                        errors[index](error);
-                    }
-                });
-            },
+            error: onError,
             complete: function () {
                 completes.forEach(function (complete) {
                     if (complete) {
-                        complete();
+                        complete.forEach(function (c) { return c(); });
                     }
@@ -83,7 +111,8 @@ });
     };
-    OperationBatcher.prototype.scheduleQueueConsumption = function () {
+    OperationBatcher.prototype.scheduleQueueConsumption = function (key) {
         var _this = this;
+        if (key === void 0) { key = ''; }
         setTimeout(function () {
-            if (_this.queuedRequests.length) {
-                _this.consumeQueue();
+            if (_this.queuedRequests.get(key) && _this.queuedRequests.get(key).length) {
+                _this.consumeQueue(key);
             }
@@ -90,0 +119,0 @@ }, this.batchInterval);

lib/batchLink.d.ts

@@ -1,2 +0,1 @@
-/// <reference types="zen-observable" />
 import { ApolloLink, Operation, FetchResult, Observable, NextLink } from 'apollo-link';
@@ -9,11 +8,10 @@ import { BatchHandler } from './batching';
         batchMax?: number;
-        batchHandler: BatchHandler;
+        batchHandler?: BatchHandler;
+        batchKey?: (Operation) => string;
     }
 }
 export declare class BatchLink extends ApolloLink {
-    private batchInterval;
-    private batchMax;
     private batcher;
-    constructor(fetchParams: BatchLink.Options);
+    constructor(fetchParams?: BatchLink.Options);
     request(operation: Operation, forward?: NextLink): Observable<FetchResult> | null;
 }

lib/batchLink.js

@@ -17,16 +17,14 @@ var __extends = (this && this.__extends) || (function () {
     function BatchLink(fetchParams) {
+        if (fetchParams === void 0) { fetchParams = {}; }
         var _this = _super.call(this) || this;
-        _this.batchInterval = (fetchParams && fetchParams.batchInterval) || 10;
-        _this.batchMax = (fetchParams && fetchParams.batchMax) || 0;
-        if (typeof _this.batchInterval !== 'number') {
-            throw new Error("batchInterval must be a number, got " + _this.batchInterval);
-        }
-        if (typeof _this.batchMax !== 'number') {
-            throw new Error("batchMax must be a number, got " + _this.batchMax);
-        }
+        var _a = fetchParams.batchInterval, batchInterval = _a === void 0 ? 10 : _a, _b = fetchParams.batchMax, batchMax = _b === void 0 ? 0 : _b, _c = fetchParams.batchHandler, batchHandler = _c === void 0 ? function () { return null; } : _c, _d = fetchParams.batchKey, batchKey = _d === void 0 ? function () { return ''; } : _d;
         _this.batcher = new OperationBatcher({
-            batchInterval: _this.batchInterval,
-            batchMax: _this.batchMax,
-            batchHandler: fetchParams.batchHandler,
+            batchInterval: batchInterval,
+            batchMax: batchMax,
+            batchHandler: batchHandler,
+            batchKey: batchKey,
         });
+        if (fetchParams.batchHandler.length <= 1) {
+            _this.request = function (operation) { return _this.batcher.enqueueRequest({ operation: operation }); };
+        }
         return _this;
@@ -33,0 +31,0 @@ }

lib/bundle.umd.js

@@ -17,8 +17,8 @@ (function (global, factory) {
     function OperationBatcher(_a) {
-        var batchInterval = _a.batchInterval, _b = _a.batchMax, batchMax = _b === void 0 ? 0 : _b, batchHandler = _a.batchHandler;
-        this.queuedRequests = [];
-        this.queuedRequests = [];
+        var batchInterval = _a.batchInterval, _b = _a.batchMax, batchMax = _b === void 0 ? 0 : _b, batchHandler = _a.batchHandler, _c = _a.batchKey, batchKey = _c === void 0 ? function () { return ''; } : _c;
+        this.queuedRequests = new Map();
         this.batchInterval = batchInterval;
         this.batchMax = batchMax;
         this.batchHandler = batchHandler;
+        this.batchKey = batchKey;
     }
@@ -28,22 +28,41 @@ OperationBatcher.prototype.enqueueRequest = function (request) {
         var requestCopy = __assign({}, request);
-        requestCopy.observable =
-            requestCopy.observable ||
-                new apolloLink.Observable(function (observer) {
-                    _this.queuedRequests.push(requestCopy);
-                    requestCopy.next = requestCopy.next || observer.next.bind(observer);
-                    requestCopy.error = requestCopy.error || observer.error.bind(observer);
-                    requestCopy.complete =
-                        requestCopy.complete || observer.complete.bind(observer);
-                    if (_this.queuedRequests.length === 1) {
-                        _this.scheduleQueueConsumption();
-                    }
-                    if (_this.queuedRequests.length === _this.batchMax) {
-                        _this.consumeQueue();
-                    }
-                });
+        var queued = false;
+        var key = this.batchKey(request.operation);
+        if (!requestCopy.observable) {
+            requestCopy.observable = new apolloLink.Observable(function (observer) {
+                if (!_this.queuedRequests.has(key)) {
+                    _this.queuedRequests.set(key, []);
+                }
+                if (!queued) {
+                    _this.queuedRequests.get(key).push(requestCopy);
+                    queued = true;
+                }
+                requestCopy.next = requestCopy.next || [];
+                if (observer.next)
+                    requestCopy.next.push(observer.next.bind(observer));
+                requestCopy.error = requestCopy.error || [];
+                if (observer.error)
+                    requestCopy.error.push(observer.error.bind(observer));
+                requestCopy.complete = requestCopy.complete || [];
+                if (observer.complete)
+                    requestCopy.complete.push(observer.complete.bind(observer));
+                if (_this.queuedRequests.get(key).length === 1) {
+                    _this.scheduleQueueConsumption(key);
+                }
+                if (_this.queuedRequests.get(key).length === _this.batchMax) {
+                    _this.consumeQueue(key);
+                }
+            });
+        }
         return requestCopy.observable;
     };
-    OperationBatcher.prototype.consumeQueue = function () {
-        var requests = this.queuedRequests.map(function (queuedRequest) { return queuedRequest.operation; });
-        var forwards = this.queuedRequests.map(function (queuedRequest) { return queuedRequest.forward; });
+    OperationBatcher.prototype.consumeQueue = function (key) {
+        if (key === void 0) { key = ''; }
+        var queuedRequests = this.queuedRequests.get(key);
+        if (!queuedRequests) {
+            return;
+        }
+        this.queuedRequests.delete(key);
+        var requests = queuedRequests.map(function (queuedRequest) { return queuedRequest.operation; });
+        var forwards = queuedRequests.map(function (queuedRequest) { return queuedRequest.forward; });
         var observables = [];
@@ -53,3 +72,3 @@ var nexts = [];
         var completes = [];
-        this.queuedRequests.forEach(function (batchableRequest, index) {
+        queuedRequests.forEach(function (batchableRequest, index) {
             observables.push(batchableRequest.observable);
@@ -60,23 +79,32 @@ nexts.push(batchableRequest.next);
         });
-        this.queuedRequests = [];
         var batchedObservable = this.batchHandler(requests, forwards) || apolloLink.Observable.of();
+        var onError = function (error) {
+            errors.forEach(function (rejecters) {
+                if (rejecters) {
+                    rejecters.forEach(function (e) { return e(error); });
+                }
+            });
+        };
         batchedObservable.subscribe({
             next: function (results) {
+                if (!Array.isArray(results)) {
+                    results = [results];
+                }
+                if (nexts.length !== results.length) {
+                    var error = new Error("server returned results with length " + results.length + ", expected length of " + nexts.length);
+                    error.result = results;
+                    return onError(error);
+                }
                 results.forEach(function (result, index) {
+                    requests[index].setContext({ response: result });
                     if (nexts[index]) {
-                        nexts[index](result);
+                        nexts[index].forEach(function (next) { return next(result); });
                     }
                 });
             },
-            error: function (error) {
-                errors.forEach(function (rejecter, index) {
-                    if (errors[index]) {
-                        errors[index](error);
-                    }
-                });
-            },
+            error: onError,
             complete: function () {
                 completes.forEach(function (complete) {
                     if (complete) {
-                        complete();
+                        complete.forEach(function (c) { return c(); });
                     }
@@ -88,7 +116,8 @@ });
     };
-    OperationBatcher.prototype.scheduleQueueConsumption = function () {
+    OperationBatcher.prototype.scheduleQueueConsumption = function (key) {
         var _this = this;
+        if (key === void 0) { key = ''; }
         setTimeout(function () {
-            if (_this.queuedRequests.length) {
-                _this.consumeQueue();
+            if (_this.queuedRequests.get(key) && _this.queuedRequests.get(key).length) {
+                _this.consumeQueue(key);
             }
@@ -113,16 +142,14 @@ }, this.batchInterval);
     function BatchLink(fetchParams) {
+        if (fetchParams === void 0) { fetchParams = {}; }
         var _this = _super.call(this) || this;
-        _this.batchInterval = (fetchParams && fetchParams.batchInterval) || 10;
-        _this.batchMax = (fetchParams && fetchParams.batchMax) || 0;
-        if (typeof _this.batchInterval !== 'number') {
-            throw new Error("batchInterval must be a number, got " + _this.batchInterval);
-        }
-        if (typeof _this.batchMax !== 'number') {
-            throw new Error("batchMax must be a number, got " + _this.batchMax);
-        }
+        var _a = fetchParams.batchInterval, batchInterval = _a === void 0 ? 10 : _a, _b = fetchParams.batchMax, batchMax = _b === void 0 ? 0 : _b, _c = fetchParams.batchHandler, batchHandler = _c === void 0 ? function () { return null; } : _c, _d = fetchParams.batchKey, batchKey = _d === void 0 ? function () { return ''; } : _d;
         _this.batcher = new OperationBatcher({
-            batchInterval: _this.batchInterval,
-            batchMax: _this.batchMax,
-            batchHandler: fetchParams.batchHandler,
+            batchInterval: batchInterval,
+            batchMax: batchMax,
+            batchHandler: batchHandler,
+            batchKey: batchKey,
         });
+        if (fetchParams.batchHandler.length <= 1) {
+            _this.request = function (operation) { return _this.batcher.enqueueRequest({ operation: operation }); };
+        }
         return _this;
@@ -129,0 +156,0 @@ }

package.json

 {
   "name": "apollo-link-batch",
-  "version": "1.0.5",
-  "description":
-    "Apollo Link that performs batching and operation on batched Operations",
+  "version": "1.1.0",
+  "description": "Apollo Link that performs batching and operation on batched Operations",
   "author": "Evans Hauser <[e-mail]>",
@@ -27,4 +26,3 @@ "contributors": [
   "scripts": {
-    "build:browser":
-      "browserify ./lib/bundle.umd.js -o=./lib/bundle.js --i apollo-link && npm run minify:browser",
+    "build:browser": "browserify ./lib/bundle.umd.js -o=./lib/bundle.js --i apollo-link && npm run minify:browser",
     "build": "tsc -p .",
@@ -35,6 +33,4 @@ "bundle": "rollup -c",
     "filesize": "npm run build && npm run build:browser",
-    "lint":
-      "tslint --type-check -p tsconfig.json -c ../../tslint.json src/*.ts",
-    "minify:browser":
-      "uglifyjs -c -m -o ./lib/bundle.min.js -- ./lib/bundle.js",
+    "lint": "tslint --type-check -p tsconfig.json -c ../../tslint.json src/*.ts",
+    "minify:browser": "uglifyjs -c -m -o ./lib/bundle.min.js -- ./lib/bundle.js",
     "postbuild": "npm run bundle",
@@ -47,10 +43,10 @@ "prebuild": "npm run clean",
   "dependencies": {
-    "apollo-link": "^1.1.0"
+    "apollo-link": "^1.2.0"
   },
   "devDependencies": {
-    "@types/graphql": "0.12.3",
+    "@types/graphql": "0.12.4",
     "@types/jest": "21.1.10",
-    "browserify": "14.5.0",
-    "fetch-mock": "5.13.1",
-    "graphql": "0.13.0",
+    "browserify": "16.1.0",
+    "fetch-mock": "6.0.0",
+    "graphql": "0.13.1",
     "graphql-tag": "2.7.3",
@@ -60,7 +56,7 @@ "jest": "21.2.1",
     "rimraf": "2.6.1",
-    "rollup": "0.55.3",
+    "rollup": "0.56.2",
     "ts-jest": "21.2.4",
     "tslint": "5.9.1",
-    "typescript": "2.7.1",
-    "uglify-js": "3.3.9"
+    "typescript": "2.7.2",
+    "uglify-js": "3.3.11"
   },
@@ -72,4 +68,10 @@ "jest": {
     "testRegex": "(/__tests__/.*|\\.(test|spec))\\.(ts|tsx|js)$",
-    "moduleFileExtensions": ["ts", "tsx", "js", "json"]
+    "moduleFileExtensions": [
+      "ts",
+      "tsx",
+      "js",
+      "json"
+    ],
+    "mapCoverage": true
   }
 }

README.md

@@ -20,3 +20,3 @@ # Batch Link
 ## Options
-Batch Link takes an object with three options on it to customize the behavoir of the link. The only required option is the batchHandler function
+Batch Link takes an object with three options on it to customize the behavior of the link. The only required option is the batchHandler function
 
@@ -23,0 +23,0 @@ |name|value|default|required|

src/__tests__/batchLink.ts

@@ -7,2 +7,4 @@ import {
   FetchResult,
+  createOperation,
+  GraphQLRequest,
 } from 'apollo-link';
@@ -20,3 +22,3 @@ import gql from 'graphql-tag';
 interface MockedResponse {
-  request: Operation;
+  request: GraphQLRequest;
   result?: FetchResult;
@@ -27,3 +29,14 @@ error?: Error;
 
-function requestToKey(request: Operation): string {
+const terminatingCheck = (done, body) => {
+  return (...args) => {
+    try {
+      body(...args);
+      done();
+    } catch (error) {
+      done.fail(error);
+    }
+  };
+};
+
+function requestToKey(request: GraphQLRequest): string {
   const queryString =
@@ -97,3 +110,3 @@ typeof request.query === 'string' ? request.query : print(request.query);
       });
-      querySched.consumeQueue();
+      querySched.consumeQueue('');
     }).not.toThrow();
@@ -108,7 +121,10 @@ });
       },
+      batchKey: () => 'yo',
     });
 
-    expect(batcher.queuedRequests.length).toBe(0);
+    expect(batcher.queuedRequests.get('')).toBeUndefined();
+    expect(batcher.queuedRequests.get('yo')).toBeUndefined();
     batcher.consumeQueue();
-    expect(batcher.queuedRequests.length).toBe(0);
+    expect(batcher.queuedRequests.get('')).toBeUndefined();
+    expect(batcher.queuedRequests.get('yo')).toBeUndefined();
   });
@@ -134,10 +150,10 @@
     const request: BatchableRequest = {
-      operation: { query },
+      operation: createOperation({}, { query }),
     };
 
-    expect(batcher.queuedRequests.length).toBe(0);
+    expect(batcher.queuedRequests.get('')).toBeUndefined();
     batcher.enqueueRequest(request).subscribe({});
-    expect(batcher.queuedRequests.length).toBe(1);
+    expect(batcher.queuedRequests.get('').length).toBe(1);
     batcher.enqueueRequest(request).subscribe({});
-    expect(batcher.queuedRequests.length).toBe(2);
+    expect(batcher.queuedRequests.get('').length).toBe(2);
   });
@@ -170,5 +186,8 @@
     );
-    const operation: Operation = {
-      query,
-    };
+    const operation: Operation = createOperation(
+      {},
+      {
+        query,
+      },
+    );
 
@@ -181,7 +200,8 @@ it('should be able to consume from a queue containing a single query', done => {
 
-      myBatcher.enqueueRequest({ operation }).subscribe(resultObj => {
-        expect(myBatcher.queuedRequests.length).toBe(0);
-        expect(resultObj).toEqual({ data });
-        done();
-      });
+      myBatcher.enqueueRequest({ operation }).subscribe(
+        terminatingCheck(done, resultObj => {
+          expect(myBatcher.queuedRequests.get('')).toBeUndefined();
+          expect(resultObj).toEqual({ data });
+        }),
+      );
       const observables: (
@@ -191,9 +211,17 @@ | Observable<FetchResult>
 
-      expect(observables.length).toBe(1);
+      try {
+        expect(observables.length).toBe(1);
+      } catch (e) {
+        done.fail(e);
+      }
     });
 
     it('should be able to consume from a queue containing multiple queries', done => {
-      const request2: Operation = {
-        query,
-      };
+      const request2: Operation = createOperation(
+        {},
+        {
+          query,
+        },
+      );
+
       const BH = createMockBatchHandler(
@@ -219,3 +247,7 @@ {
       observable1.subscribe(resultObj1 => {
-        expect(resultObj1).toEqual({ data });
+        try {
+          expect(resultObj1).toEqual({ data });
+        } catch (e) {
+          done.fail(e);
+        }
 
@@ -230,3 +262,7 @@ if (notify) {
       observable2.subscribe(resultObj2 => {
-        expect(resultObj2).toEqual({ data });
+        try {
+          expect(resultObj2).toEqual({ data });
+        } catch (e) {
+          done.fail(e);
+        }
 
@@ -240,8 +276,12 @@ if (notify) {
 
-      expect(myBatcher.queuedRequests.length).toBe(2);
-      const observables: (
-        | Observable<FetchResult>
-        | undefined)[] = myBatcher.consumeQueue()!;
-      expect(myBatcher.queuedRequests.length).toBe(0);
-      expect(observables.length).toBe(2);
+      try {
+        expect(myBatcher.queuedRequests.get('').length).toBe(2);
+        const observables: (
+          | Observable<FetchResult>
+          | undefined)[] = myBatcher.consumeQueue()!;
+        expect(myBatcher.queuedRequests.get('')).toBeUndefined();
+        expect(observables.length).toBe(2);
+      } catch (e) {
+        done.fail(e);
+      }
     });
@@ -259,6 +299,7 @@
       const observable = myBatcher.enqueueRequest({ operation });
-      observable.subscribe(result => {
-        expect(result).toEqual({ data });
-        done();
-      });
+      observable.subscribe(
+        terminatingCheck(done, result => {
+          expect(result).toEqual({ data });
+        }),
+      );
       myBatcher.consumeQueue();
@@ -269,2 +310,6 @@ });
   it('should work when single query', done => {
+    const data = {
+      lastName: 'Ever',
+      firstName: 'Greatest',
+    };
     const batcher = new OperationBatcher({
@@ -274,2 +319,3 @@ batchInterval: 10,
         new Observable(observer => {
+          observer.next([{ data }]);
           setTimeout(observer.complete.bind(observer));
@@ -286,17 +332,36 @@ }),
     `;
-    const operation: Operation = { query };
+    const operation: Operation = createOperation({}, { query });
 
     batcher.enqueueRequest({ operation }).subscribe({});
-    expect(batcher.queuedRequests.length).toBe(1);
+    try {
+      expect(batcher.queuedRequests.get('').length).toBe(1);
+    } catch (e) {
+      done.fail(e);
+    }
 
-    setTimeout(() => {
-      expect(batcher.queuedRequests.length).toBe(0);
-      done();
-    }, 20);
+    setTimeout(
+      terminatingCheck(done, () => {
+        expect(batcher.queuedRequests.get('')).toBeUndefined();
+        expect(operation.getContext()).toEqual({ response: { data } });
+      }),
+      20,
+    );
   });
 
   it('should correctly batch multiple queries', done => {
+    const data = {
+      lastName: 'Ever',
+      firstName: 'Greatest',
+    };
+    const data2 = {
+      lastName: 'Hauser',
+      firstName: 'Evans',
+    };
     const batcher = new OperationBatcher({
       batchInterval: 10,
-      batchHandler: () => null,
+      batchHandler: () =>
+        new Observable(observer => {
+          observer.next([{ data }, { data: data2 }, { data }]);
+          setTimeout(observer.complete.bind(observer));
+        }),
     });
@@ -311,19 +376,34 @@ const query = gql`
     `;
-    const operation: Operation = { query };
+    const operation: Operation = createOperation({}, { query });
+    const operation2: Operation = createOperation({}, { query });
+    const operation3: Operation = createOperation({}, { query });
 
     batcher.enqueueRequest({ operation }).subscribe({});
-    batcher.enqueueRequest({ operation }).subscribe({});
-    expect(batcher.queuedRequests.length).toBe(2);
+    batcher.enqueueRequest({ operation: operation2 }).subscribe({});
+    try {
+      expect(batcher.queuedRequests.get('').length).toBe(2);
+    } catch (e) {
+      done.fail(e);
+    }
 
     setTimeout(() => {
       // The batch shouldn't be fired yet, so we can add one more request.
-      batcher.enqueueRequest({ operation }).subscribe({});
-      expect(batcher.queuedRequests.length).toBe(3);
+      batcher.enqueueRequest({ operation: operation3 }).subscribe({});
+      try {
+        expect(batcher.queuedRequests.get('').length).toBe(3);
+      } catch (e) {
+        done.fail(e);
+      }
     }, 5);
 
-    setTimeout(() => {
-      // The batch should've been fired by now.
-      expect(batcher.queuedRequests.length).toBe(0);
-      done();
-    }, 20);
+    setTimeout(
+      terminatingCheck(done, () => {
+        // The batch should've been fired by now.
+        expect(operation.getContext()).toEqual({ response: { data } });
+        expect(operation2.getContext()).toEqual({ response: { data: data2 } });
+        expect(operation3.getContext()).toEqual({ response: { data } });
+        expect(batcher.queuedRequests.get('')).toBeUndefined();
+      }),
+      20,
+    );
   });
@@ -340,5 +420,3 @@
     `;
-    const operation: Operation = {
-      query: query,
-    };
+    const operation: Operation = createOperation({}, { query });
     const error = new Error('Network error');
@@ -356,6 +434,5 @@ const BH = createMockBatchHandler({
     observable.subscribe({
-      error: (resError: Error) => {
+      error: terminatingCheck(done, (resError: Error) => {
         expect(resError.message).toBe('Network error');
-        done();
-      },
+      }),
     });
@@ -367,2 +444,8 @@ batcher.consumeQueue();
 describe('BatchLink', () => {
+  const query = gql`
+    {
+      id
+    }
+  `;
+
   it('does not need any constructor arguments', () => {
@@ -374,14 +457,293 @@ expect(
 
-  it('passes forward on', () => {
+  it('passes forward on', done => {
     const link = ApolloLink.from([
-      new BatchLink({ batchHandler: () => Observable.of() }),
+      new BatchLink({
+        batchInterval: 0,
+        batchMax: 1,
+        batchHandler: (operation, forward) => {
+          try {
+            expect(forward.length).toBe(1);
+            expect(operation.length).toBe(1);
+          } catch (e) {
+            done.fail(e);
+          }
+          return forward[0](operation[0]).map(result => [result]);
+        },
+      }),
+      new ApolloLink(operation => {
+        terminatingCheck(done, () => {
+          expect(operation.query).toEqual(query);
+        })();
+        return null;
+      }),
     ]);
-    execute(link, {
-      query: gql`
+
+    execute(
+      link,
+      createOperation(
+        {},
         {
-          id
+          query,
+        },
+      ),
+    ).subscribe(result => done.fail());
+  });
+
+  it('raises warning if terminating', () => {
+    let calls = 0;
+    const link_full = new BatchLink({
+      batchHandler: (operation, forward) =>
+        forward[0](operation[0]).map(r => [r]),
+    });
+    const link_one_op = new BatchLink({
+      batchHandler: operation => Observable.of(),
+    });
+    const link_no_op = new BatchLink({ batchHandler: () => Observable.of() });
+    const _warn = console.warn;
+    console.warn = warning => {
+      calls++;
+      expect(warning.message).toBeDefined();
+    };
+    expect(
+      link_one_op.concat((operation, forward) => forward(operation)),
+    ).toEqual(link_one_op);
+    expect(
+      link_no_op.concat((operation, forward) => forward(operation)),
+    ).toEqual(link_no_op);
+    console.warn = warning => {
+      throw Error('non-terminating link should not throw');
+    };
+    expect(
+      link_full.concat((operation, forward) => forward(operation)),
+    ).not.toEqual(link_full);
+    console.warn = _warn;
+    expect(calls).toBe(2);
+  });
+
+  it('correctly uses batch size', done => {
+    const sizes = [1, 2, 3];
+    const terminating = new ApolloLink(operation => {
+      try {
+        expect(operation.query).toEqual(query);
+      } catch (e) {
+        done.fail(e);
+      }
+      return Observable.of(operation.variables.count);
+    });
+
+    let runBatchSize = () => {
+      const size = sizes.pop();
+      if (!size) done();
+
+      const batchHandler = jest.fn((operation, forward) => {
+        try {
+          expect(operation.length).toBe(size);
+          expect(forward.length).toBe(size);
+        } catch (e) {
+          done.fail(e);
         }
-      `,
+        const observables = forward.map((f, i) => f(operation[i]));
+        return new Observable(observer => {
+          const data = [];
+          observables.forEach(obs =>
+            obs.subscribe(d => {
+              data.push(d);
+              if (data.length === observables.length) {
+                observer.next(data);
+                observer.complete();
+              }
+            }),
+          );
+        });
+      });
+
+      const link = ApolloLink.from([
+        new BatchLink({
+          batchInterval: 1000,
+          batchMax: size,
+          batchHandler,
+        }),
+        terminating,
+      ]);
+
+      Array.from(new Array(size)).forEach((_, i) => {
+        execute(link, {
+          query,
+          variables: { count: i },
+        }).subscribe({
+          next: data => {
+            expect(data).toBe(i);
+          },
+          complete: () => {
+            try {
+              expect(batchHandler.mock.calls.length).toBe(1);
+            } catch (e) {
+              done.fail(e);
+            }
+            runBatchSize();
+          },
+        });
+      });
+    };
+
+    runBatchSize();
+  });
+
+  it('correctly follows batch interval', done => {
+    const intervals = [10, 20, 30];
+
+    const runBatchInterval = () => {
+      const mock = jest.fn();
+
+      const batchInterval = intervals.pop();
+      if (!batchInterval) return done();
+
+      const batchHandler = jest.fn((operation, forward) => {
+        try {
+          expect(operation.length).toBe(1);
+          expect(forward.length).toBe(1);
+        } catch (e) {
+          done.fail(e);
+        }
+
+        return forward[0](operation[0]).map(d => [d]);
+      });
+
+      const link = ApolloLink.from([
+        new BatchLink({
+          batchInterval,
+          batchMax: 0,
+          batchHandler,
+        }),
+        () => Observable.of(42),
+      ]);
+
+      execute(
+        link,
+        createOperation(
+          {},
+          {
+            query,
+          },
+        ),
+      ).subscribe({
+        next: data => {
+          try {
+            expect(data).toBe(42);
+          } catch (e) {
+            done.fail(e);
+          }
+        },
+        complete: () => {
+          mock(batchHandler.mock.calls.length);
+        },
+      });
+
+      setTimeout(() => {
+        const checkCalls = mock.mock.calls.slice(0, -1);
+        try {
+          expect(checkCalls.length).toBe(2);
+          checkCalls.forEach(args => expect(args[0]).toBe(0));
+          expect(mock).lastCalledWith(1);
+          expect(batchHandler.mock.calls.length).toBe(1);
+        } catch (e) {
+          done.fail(e);
+        }
+
+        runBatchInterval();
+      }, batchInterval + 5);
+
+      setTimeout(() => mock(batchHandler.mock.calls.length), batchInterval - 5);
+      setTimeout(() => mock(batchHandler.mock.calls.length), batchInterval / 2);
+    };
+    runBatchInterval();
+  });
+
+  it('throws an error when more requests than results', done => {
+    const result = [{ data: {} }];
+    const batchHandler = jest.fn(op => Observable.of(result));
+
+    const link = ApolloLink.from([
+      new BatchLink({
+        batchInterval: 10,
+        batchMax: 2,
+        batchHandler,
+      }),
+    ]);
+
+    [1, 2].forEach(x => {
+      execute(link, {
+        query,
+      }).subscribe({
+        next: data => {
+          done.fail('next should not be called');
+        },
+        error: terminatingCheck(done, error => {
+          expect(error).toBeDefined();
+          expect(error.result).toEqual(result);
+        }),
+        complete: () => {
+          done.fail('complete should not be called');
+        },
+      });
     });
   });
+
+  describe('batchKey', () => {
+    it('should allow different batches to be created separately', done => {
+      const data = { data: {} };
+      const result = [data, data];
+
+      const batchHandler = jest.fn(op => {
+        try {
+          expect(op.length).toBe(2);
+        } catch (e) {
+          done.fail(e);
+        }
+        return Observable.of(result);
+      });
+      let key = true;
+      const batchKey = () => {
+        key = !key;
+        return '' + !key;
+      };
+
+      const link = ApolloLink.from([
+        new BatchLink({
+          batchInterval: 1,
+          //if batchKey does not work, then the batch size would be 3
+          batchMax: 3,
+          batchHandler,
+          batchKey,
+        }),
+      ]);
+
+      let count = 0;
+      [1, 2, 3, 4].forEach(x => {
+        execute(link, {
+          query,
+        }).subscribe({
+          next: d => {
+            try {
+              expect(d).toEqual(data);
+            } catch (e) {
+              done.fail(e);
+            }
+          },
+          error: done.fail,
+          complete: () => {
+            count++;
+            if (count === 4) {
+              try {
+                expect(batchHandler.mock.calls.length).toBe(2);
+                done();
+              } catch (e) {
+                done.fail(e);
+              }
+            }
+          },
+        });
+      });
+    });
+  });
 });

src/batching.ts

@@ -5,3 +5,3 @@ import { Observable, Operation, NextLink, FetchResult } from 'apollo-link';
   operations: Operation[],
-  forward: (NextLink | undefined)[],
+  forward?: (NextLink | undefined)[],
 ) => Observable<FetchResult[]> | null;
@@ -17,5 +17,5 @@
   observable?: Observable<FetchResult>;
-  next?: (result: FetchResult) => void;
-  error?: (error: Error) => void;
-  complete?: () => void;
+  next?: Array<(result: FetchResult) => void>;
+  error?: Array<(error: Error) => void>;
+  complete?: Array<() => void>;
 }
@@ -28,3 +28,4 @@
   // Queue on which the QueryBatcher will operate on a per-tick basis.
-  public queuedRequests: BatchableRequest[] = [];
+  // Public only for testing
+  public queuedRequests: Map<string, BatchableRequest[]>;
 
@@ -36,2 +37,3 @@ private batchInterval: number;
   private batchHandler: BatchHandler;
+  private batchKey: (Operation) => string;
 
@@ -42,2 +44,3 @@ constructor({
     batchHandler,
+    batchKey = () => '',
   }: {
@@ -47,32 +50,53 @@ batchInterval: number;
     batchHandler: BatchHandler;
+    batchKey?: (Operation) => string;
   }) {
-    this.queuedRequests = [];
+    this.queuedRequests = new Map();
     this.batchInterval = batchInterval;
     this.batchMax = batchMax;
     this.batchHandler = batchHandler;
+    this.batchKey = batchKey;
   }
 
   public enqueueRequest(request: BatchableRequest): Observable<FetchResult> {
-    const requestCopy = { ...request };
+    const requestCopy = {
+      ...request,
+    };
+    let queued = false;
 
-    requestCopy.observable =
-      requestCopy.observable ||
-      new Observable<FetchResult>(observer => {
-        this.queuedRequests.push(requestCopy);
+    const key = this.batchKey(request.operation);
 
-        requestCopy.next = requestCopy.next || observer.next.bind(observer);
-        requestCopy.error = requestCopy.error || observer.error.bind(observer);
-        requestCopy.complete =
-          requestCopy.complete || observer.complete.bind(observer);
+    if (!requestCopy.observable) {
+      requestCopy.observable = new Observable<FetchResult>(observer => {
+        if (!this.queuedRequests.has(key)) {
+          this.queuedRequests.set(key, []);
+        }
 
+        if (!queued) {
+          this.queuedRequests.get(key).push(requestCopy);
+          queued = true;
+        }
+
+        //called for each subscriber, so need to save all listeners(next, error, complete)
+        requestCopy.next = requestCopy.next || [];
+        if (observer.next) requestCopy.next.push(observer.next.bind(observer));
+
+        requestCopy.error = requestCopy.error || [];
+        if (observer.error)
+          requestCopy.error.push(observer.error.bind(observer));
+
+        requestCopy.complete = requestCopy.complete || [];
+        if (observer.complete)
+          requestCopy.complete.push(observer.complete.bind(observer));
+
         // The first enqueued request triggers the queue consumption after `batchInterval` milliseconds.
-        if (this.queuedRequests.length === 1) {
-          this.scheduleQueueConsumption();
+        if (this.queuedRequests.get(key).length === 1) {
+          this.scheduleQueueConsumption(key);
         }
 
         // When amount of requests reaches `batchMax`, trigger the queue consumption without waiting on the `batchInterval`.
-        if (this.queuedRequests.length === this.batchMax) {
-          this.consumeQueue();
+        if (this.queuedRequests.get(key).length === this.batchMax) {
+          this.consumeQueue(key);
         }
       });
+    }
 
@@ -84,8 +108,18 @@ return requestCopy.observable;
   // Returns a list of promises (one for each query).
-  public consumeQueue(): (Observable<FetchResult> | undefined)[] | undefined {
-    const requests: Operation[] = this.queuedRequests.map(
+  public consumeQueue(
+    key: string = '',
+  ): (Observable<FetchResult> | undefined)[] | undefined {
+    const queuedRequests = this.queuedRequests.get(key);
+
+    if (!queuedRequests) {
+      return;
+    }
+
+    this.queuedRequests.delete(key);
+
+    const requests: Operation[] = queuedRequests.map(
       queuedRequest => queuedRequest.operation,
     );
 
-    const forwards: NextLink[] = this.queuedRequests.map(
+    const forwards: NextLink[] = queuedRequests.map(
       queuedRequest => queuedRequest.forward,
@@ -98,3 +132,3 @@ );
     const completes: any[] = [];
-    this.queuedRequests.forEach((batchableRequest, index) => {
+    queuedRequests.forEach((batchableRequest, index) => {
       observables.push(batchableRequest.observable);
@@ -106,26 +140,46 @@ nexts.push(batchableRequest.next);
 
-    this.queuedRequests = [];
-
     const batchedObservable =
       this.batchHandler(requests, forwards) || Observable.of();
 
+    const onError = error => {
+      //each callback list in batch
+      errors.forEach(rejecters => {
+        if (rejecters) {
+          //each subscriber to request
+          rejecters.forEach(e => e(error));
+        }
+      });
+    };
+
     batchedObservable.subscribe({
       next: results => {
+        if (!Array.isArray(results)) {
+          results = [results];
+        }
+
+        if (nexts.length !== results.length) {
+          const error = new Error(
+            `server returned results with length ${
+              results.length
+            }, expected length of ${nexts.length}`,
+          );
+          (error as any).result = results;
+
+          return onError(error);
+        }
+
         results.forEach((result, index) => {
+          // attach the raw response to the context for usage
+          requests[index].setContext({ response: result });
           if (nexts[index]) {
-            nexts[index](result);
+            nexts[index].forEach(next => next(result));
           }
         });
       },
-      error: error => {
-        errors.forEach((rejecter, index) => {
-          if (errors[index]) {
-            errors[index](error);
-          }
-        });
-      },
+      error: onError,
       complete: () => {
         completes.forEach(complete => {
           if (complete) {
-            complete();
+            //each subscriber to request
+            complete.forEach(c => c());
           }
@@ -139,6 +193,6 @@ });
 
-  private scheduleQueueConsumption(): void {
+  private scheduleQueueConsumption(key: string = ''): void {
     setTimeout(() => {
-      if (this.queuedRequests.length) {
-        this.consumeQueue();
+      if (this.queuedRequests.get(key) && this.queuedRequests.get(key).length) {
+        this.consumeQueue(key);
       }
@@ -145,0 +199,0 @@ }, this.batchInterval);

src/batchLink.ts

@@ -31,3 +31,8 @@ import {
      */
-    batchHandler: BatchHandler;
+    batchHandler?: BatchHandler;
+
+    /**
+     * creates the key for a batch
+     */
+    batchKey?: (Operation) => string;
   }
@@ -37,27 +42,25 @@ }
 export class BatchLink extends ApolloLink {
-  private batchInterval: number;
-  private batchMax: number;
   private batcher: OperationBatcher;
 
-  constructor(fetchParams: BatchLink.Options) {
+  constructor(fetchParams: BatchLink.Options = {}) {
     super();
 
-    this.batchInterval = (fetchParams && fetchParams.batchInterval) || 10;
-    this.batchMax = (fetchParams && fetchParams.batchMax) || 0;
+    const {
+      batchInterval = 10,
+      batchMax = 0,
+      batchHandler = () => null,
+      batchKey = () => '',
+    } = fetchParams;
 
-    if (typeof this.batchInterval !== 'number') {
-      throw new Error(
-        `batchInterval must be a number, got ${this.batchInterval}`,
-      );
-    }
+    this.batcher = new OperationBatcher({
+      batchInterval,
+      batchMax,
+      batchHandler,
+      batchKey,
+    });
 
-    if (typeof this.batchMax !== 'number') {
-      throw new Error(`batchMax must be a number, got ${this.batchMax}`);
+    //make this link terminating
+    if (fetchParams.batchHandler.length <= 1) {
+      this.request = operation => this.batcher.enqueueRequest({ operation });
     }
-
-    this.batcher = new OperationBatcher({
-      batchInterval: this.batchInterval,
-      batchMax: this.batchMax,
-      batchHandler: fetchParams.batchHandler,
-    });
   }
@@ -64,0 +67,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

101370

increased by119.53%

Number of package files

34

increased by70%

Lines of code

1743

increased by112.3%

Worsened metrics

Number of medium supply chain risk alerts

1

increased byInfinity%

Dependency changes

• Updatedapollo-link@^1.2.0