Streaming pagination support (streamingRequest)

src/index.ts

@@ -13,6 +13,8 @@ import {
   ISerializer,
   IStreamingAPI,
   ITransport,
+  RequestMultipleCancel,
+  RequestMultipleHandler,
   SubscribeOptions,
   SubscriptionHandler,
 } from './interface';
@@ -266,6 +268,22 @@ class DCRFClient implements IStreamingAPI {
     }
   }
 
+  private sendRequest(payload: object, requestId: string, stream: string) {
+    payload = Object.assign({}, payload, {request_id: requestId});
+    if (this.options.preprocessPayload != null) {
+      // Note: this and the preprocessMessage handler below presume an object will be returned.
+      //       If you really want to return a 0, you're kinda SOL -- wrap it in an object :P
+      payload = this.options.preprocessPayload(stream, payload, requestId) || payload;
+    }
+
+    let message = this.buildMultiplexedMessage(stream, payload);
+    if (this.options.preprocessMessage != null) {
+      message = this.options.preprocessMessage(message) || message;
+    }
+
+    this.send(message);
+  }
+
   public request(stream: string, payload: object, requestId: string=UUID.generate()): Promise<any> {
     return new Promise((resolve, reject) => {
       const selector = this.buildRequestResponseSelector(stream, requestId);
@@ -281,21 +299,37 @@ class DCRFClient implements IStreamingAPI {
           reject(response);
         }
       });
+      this.sendRequest(payload, requestId, stream);
+    });
+  }
 
-      payload = Object.assign({}, payload, {request_id: requestId});
-      if (this.options.preprocessPayload != null) {
-        // Note: this and the preprocessMessage handler below presume an object will be returned.
-        //       If you really want to return a 0, you're kinda SOL -- wrap it in an object :P
-        payload = this.options.preprocessPayload(stream, payload, requestId) || payload;
-      }
+  public requestMultiple(stream: string, payload: object, callback: RequestMultipleHandler, requestId: string = UUID.generate()): RequestMultipleCancel {
+    const selector = this.buildRequestResponseSelector(stream, requestId);
 
-      let message = this.buildMultiplexedMessage(stream, payload);
-      if (this.options.preprocessMessage != null) {
-        message = this.options.preprocessMessage(message) || message;
-      }
+    let listenerId: number | null = this.dispatcher.listen(selector, (data: typeof selector & { payload: { response_status: number, data: any } }) => {
+      const {payload: response} = data;
+      const responseStatus = response.response_status;
 
-      this.send(message);
+      // 2xx is success
+      if (Math.floor(responseStatus / 100) === 2) {
+        callback(null, response.data);
+      } else {
+        if (listenerId) {
+          this.dispatcher.cancel(listenerId);
+          listenerId = null;
+        }
+        callback(response, null);
+      }
     });
+
+    this.sendRequest(payload, requestId, stream);
+
+    return () => {
+      if (listenerId) {
+        this.dispatcher.cancel(listenerId);
+        listenerId = null;
+      }
+    };
   }
 
   public send(object: object) {

src/interface.ts

@@ -11,6 +11,11 @@ type DispatchListener<T> = (response: T) => any;
 export
 type SubscriptionHandler = (payload: {[prop: string]: any}, action: string) => any;
 
+export
+type RequestMultipleHandler = (error: {response_status: number, data: any} | null, payload: {[prop: string]: any} | null) => any;
+
+export
+type RequestMultipleCancel = () => void;
 
 /**
  * Calls all handlers whose selectors match an incoming payload.
@@ -406,6 +411,8 @@ interface IStreamingAPI {
    *    On failure, the promise will be rejected with the entire API response.
    */
   request(stream: string, payload: object, requestId?: string): Promise<object>;
+
+  requestMultiple(stream: string, payload: object, callback:RequestMultipleHandler, requestId?: string): RequestMultipleCancel;
 }
 
 

test/test.ts

@@ -243,6 +243,114 @@ describe('DCRFClient', function() {
     });
   });
 
+  describe('requestMultiple', function() {
+    it('sends request and listen for responses until cancel', function () {
+      const responses: any[] = [];
+      const cancel = api.requestMultiple('test', {'key': 'unique'}, (error, response) => {
+        responses.push(response);
+      });
+
+      expect(transport.send).to.have.been.calledOnce;
+      const msg = transport.send.getCall(0).args[0];
+      const stream = msg.stream;
+      const requestId = msg.payload.request_id;
+
+      transport.emit('message', {
+        data: {
+          stream,
+          payload: {
+            request_id: requestId,
+            response_status: 200,
+            data: {response: 'unique'}
+          }
+        }
+      });
+
+      transport.emit('message', {
+        data: {
+          stream,
+          payload: {
+            request_id: requestId,
+            response_status: 200,
+            data: {response: 'unique2'}
+          }
+        }
+      });
+
+      cancel();
+
+      transport.emit('message', {
+        data: {
+          stream,
+          payload: {
+            request_id: requestId,
+            response_status: 200,
+            data: {response: 'unique3'}
+          }
+        }
+      });
+
+      expect(responses).to.deep.equal([{'response': 'unique'}, {'response': 'unique2'}]);
+    });
+
+    it('cancels when receiving an error.', function () {
+      const responses: any[] = [];
+      const errors: any[] = [];
+      const cancel = api.requestMultiple('test', {'key': 'unique'}, (error, response) => {
+        if (error) {
+          errors.push(error);
+        } else {
+          responses.push(response);
+        }
+      });
+
+      expect(transport.send).to.have.been.calledOnce;
+      const msg = transport.send.getCall(0).args[0];
+      const stream = msg.stream;
+      const requestId = msg.payload.request_id;
+
+      transport.emit('message', {
+        data: {
+          stream,
+          payload: {
+            request_id: requestId,
+            response_status: 200,
+            data: {response: 'unique'}
+          }
+        }
+      });
+
+      transport.emit('message', {
+        data: {
+          stream,
+          payload: {
+            request_id: requestId,
+            response_status: 400,
+            data: {response: 'unique2'}
+          }
+        }
+      });
+
+      transport.emit('message', {
+        data: {
+          stream,
+          payload: {
+            request_id: requestId,
+            response_status: 200,
+            data: {response: 'unique3'}
+          }
+        }
+      });
+
+      expect(responses).to.deep.equal([{'response': 'unique'}]);
+      expect(errors).to.deep.equal([{
+        request_id: requestId,
+        response_status: 400,
+        data: {response: 'unique2'}
+      }]);
+      expect(cancel).to.not.throw();
+    });
+  });
 
   describe('subscribe', function() {
     it('invokes callback on every update', function() {