@open-rpc/client-js - npm Package Compare versions

Comparing version 1.1.1 to 1.2.0

build/__mocks__/isomorphic-fetch.js

 "use strict";
+var __importStar = (this && this.__importStar) || function (mod) {
+    if (mod && mod.__esModule) return mod;
+    var result = {};
+    if (mod != null) for (var k in mod) if (Object.hasOwnProperty.call(mod, k)) result[k] = mod[k];
+    result["default"] = mod;
+    return result;
+};
 Object.defineProperty(exports, "__esModule", { value: true });
+var req = __importStar(require("./requestData"));
 var Fetch = function (url, options) {
+    if (url.match(/crash/)) {
+        throw new Error("Random Segfault that crashes fetch");
+    }
     var resultPromise = {
         text: function () {
-            return Promise.resolve(options.body);
+            return Promise.resolve(req.generateMockResponseData(url, options.body));
         },
@@ -8,0 +19,0 @@ };

build/__mocks__/isomorphic-ws.d.ts

 declare class WebSocket {
     private callbacks;
-    constructor(uri: string, props: any);
+    private url;
+    constructor(url: string, props: any);
     addEventListener(eventName: string, callback: any): void;
     removeEventListener(eventName: string, callback: any): void;
-    send(data: any): void;
+    send(data: any, callback: (err?: Error) => void): void;
     close(): void;
 }
 export default WebSocket;

build/__mocks__/isomorphic-ws.js

 "use strict";
+var __importStar = (this && this.__importStar) || function (mod) {
+    if (mod && mod.__esModule) return mod;
+    var result = {};
+    if (mod != null) for (var k in mod) if (Object.hasOwnProperty.call(mod, k)) result[k] = mod[k];
+    result["default"] = mod;
+    return result;
+};
 Object.defineProperty(exports, "__esModule", { value: true });
+var req = __importStar(require("./requestData"));
 var WebSocket = /** @class */ (function () {
-    function WebSocket(uri, props) {
+    function WebSocket(url, props) {
         this.callbacks = {};
+        this.url = url;
     }
@@ -18,7 +27,17 @@ WebSocket.prototype.addEventListener = function (eventName, callback) {
     };
-    WebSocket.prototype.send = function (data) {
+    WebSocket.prototype.send = function (data, callback) {
+        var _this = this;
+        if (this.url.match(/crash-null/)) {
+            callback();
+            return;
+        }
+        if (this.url.match(/crash/)) {
+            callback(new Error("Random Segfault that crashes fetch"));
+            return;
+        }
         Object.entries(this.callbacks).forEach(function (_a) {
-            var eventName = _a[0], callback = _a[1];
+            var eventName = _a[0], cb = _a[1];
             if (eventName === "message") {
-                callback({ data: data });
+                cb({ data: req.generateMockResponseData(_this.url, data) });
+                callback();
             }
@@ -25,0 +44,0 @@ });

build/index.d.ts

@@ -5,2 +5,3 @@ import RequestManager from "./RequestManager";
 import WebSocketTransport from "./transports/WebSocketTransport";
+import { JSONRPCError } from "./Error";
 interface IClient {
@@ -61,5 +62,8 @@ request(method: string, params: any): Promise<any>;
      */
-    request(method: string, params: any): Promise<any>;
+    request(method: string, params: any, timeout?: number): Promise<any>;
+    notify(method: string, params: any): Promise<any>;
+    onNotification(callback: (data: any) => void): void;
+    onError(callback: (data: JSONRPCError) => void): void;
 }
 export default Client;
 export { Client, RequestManager, HTTPTransport, EventEmitterTransport, WebSocketTransport, };

build/index.js

 "use strict";
 var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
+    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
     return new (P || (P = Promise))(function (resolve, reject) {
         function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
         function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
-        function step(result) { result.done ? resolve(result.value) : new P(function (resolve) { resolve(result.value); }).then(fulfilled, rejected); }
+        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
         step((generator = generator.apply(thisArg, _arguments || [])).next());
@@ -106,3 +107,3 @@ });
      */
-    Client.prototype.request = function (method, params) {
+    Client.prototype.request = function (method, params, timeout) {
         return __awaiter(this, void 0, void 0, function () {
@@ -114,3 +115,3 @@ return __generator(this, function (_a) {
                         _a.sent();
-                        return [2 /*return*/, this.requestManager.request(method, params)];
+                        return [2 /*return*/, this.requestManager.request(method, params, false, timeout)];
                 }
@@ -120,2 +121,20 @@ });
     };
+    Client.prototype.notify = function (method, params) {
+        return __awaiter(this, void 0, void 0, function () {
+            return __generator(this, function (_a) {
+                switch (_a.label) {
+                    case 0: return [4 /*yield*/, this.requestManager.connectPromise];
+                    case 1:
+                        _a.sent();
+                        return [2 /*return*/, this.requestManager.request(method, params, true)];
+                }
+            });
+        });
+    };
+    Client.prototype.onNotification = function (callback) {
+        this.requestManager.requestChannel.addListener("notification", callback);
+    };
+    Client.prototype.onError = function (callback) {
+        this.requestManager.requestChannel.addListener("error", callback);
+    };
     return Client;
@@ -122,0 +141,0 @@ }());

build/index.test.js

@@ -10,4 +10,2 @@ "use strict";
 var events_1 = require("events");
-jest.mock("./RequestManager");
-var mockedRequestManager = RequestManager_1.default;
 describe("client-js", function () {
@@ -25,19 +23,32 @@ it("can be constructed", function () {
     });
+    it("has a notify method that returns a promise", function () {
+        var emitter = new events_1.EventEmitter();
+        var c = new _1.default(new RequestManager_1.default([new EventEmitterTransport_1.default(emitter, "from1", "to1")]));
+        expect(typeof c.request).toEqual("function");
+        expect(typeof c.notify("my_method", null).then).toEqual("function");
+    });
+    it("can register error and subscription handlers", function () {
+        var emitter = new events_1.EventEmitter();
+        var c = new _1.default(new RequestManager_1.default([new EventEmitterTransport_1.default(emitter, "from1", "to1")]));
+        // tslint:disable-next-line:no-empty
+        c.onError(function (err) { });
+        // tslint:disable-next-line:no-empty
+        c.onNotification(function (data) { });
+    });
     describe("startBatch", function () {
-        it("calls the requestManager.startBatch", function () {
+        it("calls startBatch", function () {
             var emitter = new events_1.EventEmitter();
-            var rm = new mockedRequestManager([new EventEmitterTransport_1.default(emitter, "from1", "to1")]);
+            var rm = new RequestManager_1.default([new EventEmitterTransport_1.default(emitter, "from1", "to1")]);
             var c = new _1.default(rm);
             c.startBatch();
-            expect(mockedRequestManager.mock.instances[0].startBatch).toHaveBeenCalled();
+            //      expect(mockedRequestManager.mock.instances[0].startBatch).toHaveBeenCalled();
         });
     });
-    describe("stopBatch", function () {
+    describe("can call stopBatch", function () {
         var emitter = new events_1.EventEmitter();
-        var rm = new mockedRequestManager([new EventEmitterTransport_1.default(emitter, "from1", "to1")]);
+        var rm = new RequestManager_1.default([new EventEmitterTransport_1.default(emitter, "from1", "to1")]);
         var c = new _1.default(rm);
         c.startBatch();
         c.stopBatch();
-        expect(mockedRequestManager.mock.instances[0].startBatch).toHaveBeenCalled();
     });
 });

build/RequestManager.d.ts

@@ -1,18 +0,24 @@
-import ITransport from "./transports/Transport";
-interface IJSONRPCRequest {
-    jsonrpc: "2.0";
-    id: string | number;
-    method: string;
-    params: any[] | object;
+/// <reference types="node" />
+import { Transport } from "./transports/Transport";
+import { IBatchRequest } from "./Request";
+import { JSONRPCError } from "./Error";
+import StrictEventEmitter from "strict-event-emitter-types";
+import { EventEmitter } from "events";
+export declare type RequestChannel = StrictEventEmitter<EventEmitter, IRequestEvents>;
+export interface IRequestEvents {
+    "error": (err: JSONRPCError) => void;
+    "notification": (data: any) => void;
 }
 declare class RequestManager {
-    transports: ITransport[];
+    transports: Transport[];
     connectPromise: Promise<any>;
-    batch: IJSONRPCRequest[];
+    batch: IBatchRequest[];
+    requestChannel: RequestChannel;
     private requests;
     private batchStarted;
     private lastId;
-    constructor(transports: ITransport[]);
+    constructor(transports: Transport[]);
     connect(): Promise<any>;
-    request(method: string, params: any): Promise<any>;
+    getPrimaryTransport(): Transport;
+    request(method: string, params: any[], notification?: boolean, timeout?: number): Promise<any>;
     close(): void;
@@ -27,4 +33,6 @@ /**
     stopBatch(): void;
-    private onData;
+    private makeRequest;
+    private handleError;
+    private handleNotification;
 }
 export default RequestManager;

build/RequestManager.js

 "use strict";
 var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
+    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
     return new (P || (P = Promise))(function (resolve, reject) {
         function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
         function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
-        function step(result) { result.done ? resolve(result.value) : new P(function (resolve) { resolve(result.value); }).then(fulfilled, rejected); }
+        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
         step((generator = generator.apply(thisArg, _arguments || [])).next());
@@ -38,2 +39,3 @@ });
 Object.defineProperty(exports, "__esModule", { value: true });
+var events_1 = require("events");
 /*
@@ -52,2 +54,3 @@ ** Naive Request Manager, only use 1st transport.
         this.connectPromise = this.connect();
+        this.requestChannel = new events_1.EventEmitter();
     }
@@ -60,3 +63,4 @@ RequestManager.prototype.connect = function () {
                     case 0:
-                        transport.onData(this.onData.bind(this));
+                        transport.subscribe("error", this.handleError.bind(this));
+                        transport.subscribe("notification", this.handleNotification.bind(this));
                         return [4 /*yield*/, transport.connect()];
@@ -70,24 +74,21 @@ case 1:
     };
-    RequestManager.prototype.request = function (method, params) {
+    RequestManager.prototype.getPrimaryTransport = function () {
+        return this.transports[0];
+    };
+    RequestManager.prototype.request = function (method, params, notification, timeout) {
+        if (notification === void 0) { notification = false; }
         return __awaiter(this, void 0, void 0, function () {
-            var i, transport, payload;
+            var internalID, id, payload, result;
             var _this = this;
             return __generator(this, function (_a) {
-                i = (++this.lastId).toString();
-                transport = this.transports[0];
-                payload = {
-                    jsonrpc: "2.0",
-                    id: i,
-                    method: method,
-                    params: params,
-                };
-                return [2 /*return*/, new Promise(function (resolve, reject) {
-                        _this.requests[i] = { resolve: resolve, reject: reject };
-                        if (_this.batchStarted) {
-                            _this.batch.push(payload);
-                        }
-                        else {
-                            transport.sendData(JSON.stringify(payload));
-                        }
-                    }).finally(function () { return _this.requests[i] = undefined; })];
+                internalID = (++this.lastId).toString();
+                id = notification ? null : internalID;
+                payload = { request: this.makeRequest(method, params, id), internalID: internalID };
+                if (this.batchStarted) {
+                    result = new Promise(function (resolve, reject) {
+                        _this.batch.push({ resolve: resolve, reject: reject, request: payload });
+                    });
+                    return [2 /*return*/, result];
+                }
+                return [2 /*return*/, this.getPrimaryTransport().sendData(payload, timeout)];
             });
@@ -108,5 +109,2 @@ });
     RequestManager.prototype.startBatch = function () {
-        if (this.batchStarted) {
-            return;
-        }
         this.batchStarted = true;
@@ -119,31 +117,23 @@ };
         if (this.batch.length === 0) {
+            this.batchStarted = false;
             return;
         }
-        var batch = JSON.stringify(this.batch);
+        this.getPrimaryTransport().sendData(this.batch);
         this.batch = [];
-        this.transports[0].sendData(batch);
+        this.batchStarted = false;
     };
-    RequestManager.prototype.onData = function (data) {
-        var _this = this;
-        var parsedData = JSON.parse(data);
-        var results = parsedData instanceof Array ? parsedData : [parsedData];
-        results.forEach(function (response) {
-            var id = typeof response.id === "string" ? response.id : response.id.toString();
-            var promiseForResult = _this.requests[id];
-            if (promiseForResult === undefined) {
-                throw new Error("Received an unrecognized response id: " + response.id + ". Valid ids are: " + Object.keys(_this.requests));
-            }
-            if (response.error) {
-                promiseForResult.reject(response.error);
-            }
-            else if (response.result) {
-                promiseForResult.resolve(response.result);
-            }
-            else {
-                promiseForResult.reject(new Error("Malformed JSON-RPC response object: " + JSON.stringify(response)));
-            }
-        });
+    RequestManager.prototype.makeRequest = function (method, params, id) {
+        if (id) {
+            return { jsonrpc: "2.0", id: id, method: method, params: params };
+        }
+        return { jsonrpc: "2.0", method: method, params: params };
     };
+    RequestManager.prototype.handleError = function (data) {
+        this.requestChannel.emit("error", data);
+    };
+    RequestManager.prototype.handleNotification = function (data) {
+        this.requestChannel.emit(data);
+    };
     return RequestManager;
 }());
 exports.default = RequestManager;

build/RequestManager.test.js

 "use strict";
 var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
+    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
     return new (P || (P = Promise))(function (resolve, reject) {
         function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
         function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
-        function step(result) { result.done ? resolve(result.value) : new P(function (resolve) { resolve(result.value); }).then(fulfilled, rejected); }
+        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
         step((generator = generator.apply(thisArg, _arguments || [])).next());
@@ -40,3 +41,2 @@ });
 };
-var _this = this;
 Object.defineProperty(exports, "__esModule", { value: true });
@@ -46,4 +46,5 @@ var RequestManager_1 = __importDefault(require("./RequestManager"));
 var events_1 = require("events");
+var eventEmitter_1 = require("./__mocks__/eventEmitter");
 describe("client-js", function () {
-    it("can be constructed", function () {
+    it("can be constructed and connect", function () {
         var emitter = new events_1.EventEmitter();
@@ -54,26 +55,28 @@ var transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
     });
-    it("has a request method that returns a promise", function () {
+    it("can close", function () {
         var emitter = new events_1.EventEmitter();
         var transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
         var c = new RequestManager_1.default([transport]);
-        expect(typeof c.request).toEqual("function");
-        expect(typeof c.request("my_method", null).then).toEqual("function");
+        c.close();
     });
-    it("can connect", function () { return __awaiter(_this, void 0, void 0, function () {
-        var emitter, transport, c;
+    it("can send a request", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, transport, c, result;
         return __generator(this, function (_a) {
-            emitter = new events_1.EventEmitter();
-            transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
-            c = new RequestManager_1.default([transport]);
-            return [2 /*return*/, c.connect()];
+            switch (_a.label) {
+                case 0:
+                    emitter = new events_1.EventEmitter();
+                    eventEmitter_1.addMockServerTransport(emitter, "to1://local/rpc-request", "from1");
+                    transport = new EventEmitterTransport_1.default(emitter, "from1", "to1://local/rpc-request");
+                    c = new RequestManager_1.default([transport]);
+                    return [4 /*yield*/, c.request("foo", ["bar"])];
+                case 1:
+                    result = _a.sent();
+                    expect(result.method).toEqual("foo");
+                    expect(result.params).toEqual(["bar"]);
+                    return [2 /*return*/];
+            }
         });
     }); });
-    it("can close", function () {
-        var emitter = new events_1.EventEmitter();
-        var transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
-        var c = new RequestManager_1.default([transport]);
-        c.close();
-    });
-    it("can send a request", function () { return __awaiter(_this, void 0, void 0, function () {
-        var emitter, transport, serverTransport, c, reqPromise;
+    it("can error on error response", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, transport, c;
         return __generator(this, function (_a) {
@@ -83,13 +86,8 @@ switch (_a.label) {
                     emitter = new events_1.EventEmitter();
-                    transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
-                    serverTransport = new EventEmitterTransport_1.default(emitter, "to1", "from1");
+                    eventEmitter_1.addMockServerTransport(emitter, "to1://local/rpc-error", "from1");
+                    transport = new EventEmitterTransport_1.default(emitter, "from1", "to1://local/rpc-error");
                     c = new RequestManager_1.default([transport]);
-                    return [4 /*yield*/, c.connect()];
+                    return [4 /*yield*/, expect(c.request("foo", ["bar"])).rejects.toThrowError("Error message")];
                 case 1:
                     _a.sent();
-                    reqPromise = c.request("foo", []);
-                    serverTransport.sendData(JSON.stringify({ id: 0, result: { foo: "foofoo" } }));
-                    return [4 /*yield*/, expect(reqPromise).resolves.toEqual({ foo: "foofoo" })];
-                case 2:
-                    _a.sent();
                     return [2 /*return*/];
@@ -99,17 +97,4 @@ }
     }); });
-    it("can error on malformed response", function (done) {
-        var emitter = new events_1.EventEmitter();
-        var transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
-        var serverTransport = new EventEmitterTransport_1.default(emitter, "to1", "from1");
-        var c = new RequestManager_1.default([transport]);
-        c.connect().then(function () {
-            c.request("foo", []).catch(function (e) {
-                expect(e.message).toContain("Malformed");
-                done();
-            });
-            serverTransport.sendData(JSON.stringify({ id: 0, foo: "bar" }));
-        });
-    });
-    it("can error on batchng a request", function () { return __awaiter(_this, void 0, void 0, function () {
-        var emitter, transport, c;
+    it("can error on malformed response and recieve error", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, transport, c, unknownError, formatError;
         return __generator(this, function (_a) {
@@ -119,8 +104,18 @@ switch (_a.label) {
                     emitter = new events_1.EventEmitter();
-                    transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
+                    eventEmitter_1.addMockServerTransport(emitter, "to1://local/rpc-garbage", "from1");
+                    transport = new EventEmitterTransport_1.default(emitter, "from1", "to1://local/rpc-garbage");
                     c = new RequestManager_1.default([transport]);
-                    return [4 /*yield*/, c.connect()];
+                    unknownError = new Promise(function (resolve) {
+                        c.requestChannel.on("error", function (d) {
+                            resolve(d);
+                        });
+                    });
+                    return [4 /*yield*/, expect(c.request("foo", ["bar"], false, 1000))
+                            .rejects.toThrowError("Request timeout request took longer than 1000 ms to resolve")];
                 case 1:
                     _a.sent();
-                    expect(function () { return c.stopBatch(); }).toThrow();
+                    return [4 /*yield*/, unknownError];
+                case 2:
+                    formatError = _a.sent();
+                    expect(formatError.message).toContain("Bad response format");
                     return [2 /*return*/];
@@ -130,104 +125,14 @@ }
     }); });
-    it("can return errors on batch requests", function (done) {
-        var emitter = new events_1.EventEmitter();
-        var transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
-        var serverTransport = new EventEmitterTransport_1.default(emitter, "to1", "from1");
-        var c = new RequestManager_1.default([transport]);
-        c.connect().then(function () {
-            c.startBatch();
-            var requests = [
-                c.request("foo", []),
-                c.request("foo", []),
-            ];
-            Promise.all(requests).catch(function (e) {
-                expect(e).toEqual({
-                    code: 509,
-                    message: "too much 509",
-                    data: {
-                        test: "data",
-                    },
-                });
-                c.close();
-                done();
-            });
-            c.stopBatch();
-            serverTransport.sendData(JSON.stringify([
-                {
-                    jsonrpc: "2.0",
-                    id: "0",
-                    error: {
-                        code: 509,
-                        message: "too much 509",
-                        data: {
-                            test: "data",
-                        },
-                    },
-                },
-                {
-                    jsonrpc: "2.0",
-                    id: "1",
-                    result: "bar",
-                },
-            ]));
+    it("can error on batchng a request", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, transport, c;
+        return __generator(this, function (_a) {
+            emitter = new events_1.EventEmitter();
+            transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
+            c = new RequestManager_1.default([transport]);
+            expect(function () { return c.stopBatch(); }).toThrow();
+            return [2 /*return*/];
         });
-    });
-    it("can batch a request", function (done) {
-        var emitter = new events_1.EventEmitter();
-        var transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
-        var serverTransport = new EventEmitterTransport_1.default(emitter, "to1", "from1");
-        var c = new RequestManager_1.default([transport]);
-        c.connect().then(function () {
-            c.startBatch();
-            var requests = [
-                c.request("foo", []),
-                c.request("foo", []),
-            ];
-            c.stopBatch();
-            Promise.all(requests).then(function (_a) {
-                var a = _a[0], b = _a[1];
-                expect(a).toEqual("foo");
-                expect(b).toEqual("bar");
-                c.close();
-                done();
-            });
-            serverTransport.sendData(JSON.stringify([
-                {
-                    jsonrpc: "2.0",
-                    id: 0,
-                    result: "foo",
-                },
-                {
-                    jsonrpc: "2.0",
-                    id: 1,
-                    result: "bar",
-                },
-            ]));
-        });
-    });
-    it("can send a request and error", function (done) {
-        var emitter = new events_1.EventEmitter();
-        var transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
-        var serverTransport = new EventEmitterTransport_1.default(emitter, "to1", "from1");
-        var c = new RequestManager_1.default([transport]);
-        c.connect().then(function () {
-            c.request("foo", [])
-                .catch(function (e) {
-                expect(e.message).toEqual("out of order");
-                done();
-            });
-            serverTransport.sendData(JSON.stringify({
-                jsonrpc: "2.0",
-                id: 0,
-                error: {
-                    code: 0,
-                    message: "out of order",
-                    data: {
-                        foo: "bar",
-                    },
-                },
-            }));
-        });
-    });
-    it("onData throws if the ID is not found", function () { return __awaiter(_this, void 0, void 0, function () {
-        var emitter, transport, serverTransport, c;
+    }); });
+    it("can return errors on batch requests", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, transport, c, requests;
         return __generator(this, function (_a) {
@@ -237,13 +142,14 @@ switch (_a.label) {
                     emitter = new events_1.EventEmitter();
-                    transport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
-                    serverTransport = new EventEmitterTransport_1.default(emitter, "to1", "from1");
+                    eventEmitter_1.addMockServerTransport(emitter, "to1://local/rpc-error", "from1");
+                    transport = new EventEmitterTransport_1.default(emitter, "from1", "to1://local/rpc-error");
                     c = new RequestManager_1.default([transport]);
-                    return [4 /*yield*/, c.connect()];
+                    c.startBatch();
+                    requests = [
+                        c.request("foo", ["bar"]),
+                        c.request("foo", ["bar"]),
+                    ];
+                    c.stopBatch();
+                    return [4 /*yield*/, expect(Promise.all(requests)).rejects.toThrowError("Error message")];
                 case 1:
                     _a.sent();
-                    expect(function () { return serverTransport.sendData(JSON.stringify({
-                        jsonrpc: "2.0",
-                        id: 10,
-                        result: 123,
-                    })); }).toThrow("Received an unrecognized response id: 10. Valid ids are: ");
                     return [2 /*return*/];
@@ -253,2 +159,50 @@ }
     }); });
+    it("can batch a request", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, transport, c, requests, _a, a, b;
+        return __generator(this, function (_b) {
+            switch (_b.label) {
+                case 0:
+                    emitter = new events_1.EventEmitter();
+                    eventEmitter_1.addMockServerTransport(emitter, "to1://local/rpc-request", "from1");
+                    transport = new EventEmitterTransport_1.default(emitter, "from1", "to1://local/rpc-request");
+                    c = new RequestManager_1.default([transport]);
+                    c.startBatch();
+                    requests = [
+                        c.request("foo", []),
+                        c.request("foo", ["bar"]),
+                    ];
+                    c.stopBatch();
+                    return [4 /*yield*/, Promise.all(requests)];
+                case 1:
+                    _a = _b.sent(), a = _a[0], b = _a[1];
+                    expect(a.method).toEqual("foo");
+                    expect(b.method).toEqual("foo");
+                    expect(a.params).toEqual([]);
+                    expect(b.params).toEqual(["bar"]);
+                    return [2 /*return*/];
+            }
+        });
+    }); });
+    it("can batch a notifications", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, transport, c, requests, _a, a, b;
+        return __generator(this, function (_b) {
+            switch (_b.label) {
+                case 0:
+                    emitter = new events_1.EventEmitter();
+                    eventEmitter_1.addMockServerTransport(emitter, "to1://local/rpc-request", "from1");
+                    transport = new EventEmitterTransport_1.default(emitter, "from1", "to1://local/rpc-request");
+                    c = new RequestManager_1.default([transport]);
+                    c.startBatch();
+                    requests = [
+                        c.request("foo", [], true),
+                        c.request("foo", ["bar"], true),
+                    ];
+                    c.stopBatch();
+                    return [4 /*yield*/, Promise.all(requests)];
+                case 1:
+                    _a = _b.sent(), a = _a[0], b = _a[1];
+                    return [2 /*return*/];
+            }
+        });
+    }); });
     describe("stopBatch", function () {
@@ -266,3 +220,3 @@ it("does nothing if the batch is empty", function () {
     describe("startBatch", function () {
-        it("it does nothing if a batch is already started", function () { return __awaiter(_this, void 0, void 0, function () {
+        it("it does nothing if a batch is already started", function () { return __awaiter(void 0, void 0, void 0, function () {
             var emitter, transport, c;
@@ -269,0 +223,0 @@ return __generator(this, function (_a) {

build/transports/EventEmitterTransport.d.ts

 /// <reference types="node" />
 import { EventEmitter } from "events";
-import ITransport from "./Transport";
-declare class EventEmitterTransport implements ITransport {
+import { Transport } from "./Transport";
+import { JSONRPCRequestData } from "../Request";
+declare class EventEmitterTransport extends Transport {
     connection: EventEmitter;
     private reqUri;
     private resUri;
-    private onDataCallbacks;
-    constructor(emitter: EventEmitter, reqUri: string, resUri: string);
+    constructor(destEmitter: EventEmitter, reqUri: string, resUri: string);
     connect(): Promise<any>;
-    onData(callback: (data: string) => void): void;
-    sendData(data: string): void;
+    sendData(data: JSONRPCRequestData, timeout?: number): Promise<any>;
     close(): void;
 }
 export default EventEmitterTransport;

build/transports/EventEmitterTransport.js

 "use strict";
+var __extends = (this && this.__extends) || (function () {
+    var extendStatics = function (d, b) {
+        extendStatics = Object.setPrototypeOf ||
+            ({ __proto__: [] } instanceof Array && function (d, b) { d.__proto__ = b; }) ||
+            function (d, b) { for (var p in b) if (b.hasOwnProperty(p)) d[p] = b[p]; };
+        return extendStatics(d, b);
+    };
+    return function (d, b) {
+        extendStatics(d, b);
+        function __() { this.constructor = d; }
+        d.prototype = b === null ? Object.create(b) : (__.prototype = b.prototype, new __());
+    };
+})();
 Object.defineProperty(exports, "__esModule", { value: true });
-var EventEmitterTransport = /** @class */ (function () {
-    function EventEmitterTransport(emitter, reqUri, resUri) {
-        this.onDataCallbacks = [];
-        this.connection = emitter;
-        this.reqUri = reqUri;
-        this.resUri = resUri;
+var Transport_1 = require("./Transport");
+var Request_1 = require("../Request");
+var Error_1 = require("../Error");
+var EventEmitterTransport = /** @class */ (function (_super) {
+    __extends(EventEmitterTransport, _super);
+    function EventEmitterTransport(destEmitter, reqUri, resUri) {
+        var _this = _super.call(this) || this;
+        _this.connection = destEmitter;
+        _this.reqUri = reqUri;
+        _this.resUri = resUri;
+        return _this;
     }
@@ -13,14 +31,21 @@ EventEmitterTransport.prototype.connect = function () {
         this.connection.on(this.reqUri, function (data) {
-            _this.onDataCallbacks.map(function (callback) {
-                callback(data);
-            });
+            _this.transportRequestManager.resolveResponse(data);
         });
         return Promise.resolve();
     };
-    EventEmitterTransport.prototype.onData = function (callback) {
-        this.onDataCallbacks.push(callback);
+    EventEmitterTransport.prototype.sendData = function (data, timeout) {
+        var prom = this.transportRequestManager.addRequest(data, timeout);
+        var notifications = Request_1.getNotifications(data);
+        var parsedData = this.parseData(data);
+        try {
+            this.connection.emit(this.resUri, parsedData);
+            this.transportRequestManager.settlePendingRequest(notifications);
+            return prom;
+        }
+        catch (e) {
+            var responseErr = new Error_1.JSONRPCError(e.message, Error_1.ERR_UNKNOWN, e);
+            this.transportRequestManager.settlePendingRequest(notifications, responseErr);
+            return Promise.reject(responseErr);
+        }
     };
-    EventEmitterTransport.prototype.sendData = function (data) {
-        this.connection.emit(this.resUri, data);
-    };
     EventEmitterTransport.prototype.close = function () {
@@ -30,3 +55,3 @@ this.connection.removeAllListeners();
     return EventEmitterTransport;
-}());
+}(Transport_1.Transport));
 exports.default = EventEmitterTransport;

build/transports/EventEmitterTransport.test.js

 "use strict";
+var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
+    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
+    return new (P || (P = Promise))(function (resolve, reject) {
+        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
+        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
+        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
+        step((generator = generator.apply(thisArg, _arguments || [])).next());
+    });
+};
+var __generator = (this && this.__generator) || function (thisArg, body) {
+    var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
+    return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
+    function verb(n) { return function (v) { return step([n, v]); }; }
+    function step(op) {
+        if (f) throw new TypeError("Generator is already executing.");
+        while (_) try {
+            if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1])).done) return t;
+            if (y = 0, t) op = [op[0] & 2, t.value];
+            switch (op[0]) {
+                case 0: case 1: t = op; break;
+                case 4: _.label++; return { value: op[1], done: false };
+                case 5: _.label++; y = op[1]; op = [0]; continue;
+                case 7: op = _.ops.pop(); _.trys.pop(); continue;
+                default:
+                    if (!(t = _.trys, t = t.length > 0 && t[t.length - 1]) && (op[0] === 6 || op[0] === 2)) { _ = 0; continue; }
+                    if (op[0] === 3 && (!t || (op[1] > t[0] && op[1] < t[3]))) { _.label = op[1]; break; }
+                    if (op[0] === 6 && _.label < t[1]) { _.label = t[1]; t = op; break; }
+                    if (t && _.label < t[2]) { _.label = t[2]; _.ops.push(op); break; }
+                    if (t[2]) _.ops.pop();
+                    _.trys.pop(); continue;
+            }
+            op = body.call(thisArg, _);
+        } catch (e) { op = [6, e]; y = 0; } finally { f = t = 0; }
+        if (op[0] & 5) throw op[1]; return { value: op[0] ? op[1] : void 0, done: true };
+    }
+};
 var __importDefault = (this && this.__importDefault) || function (mod) {
@@ -8,8 +44,19 @@ return (mod && mod.__esModule) ? mod : { "default": mod };
 var events_1 = require("events");
+var requestData_1 = require("../__mocks__/requestData");
+var eventEmitter_1 = require("../__mocks__/eventEmitter");
 describe("EventEmitterTransport", function () {
-    it("can connect", function () {
-        var emitter = new events_1.EventEmitter();
-        var eventEmitterTransport = new EventEmitterTransport_1.default(emitter, "foo://in", "foo://out");
-        eventEmitterTransport.connect();
-    });
+    it("can connect", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, eventEmitterTransport;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    emitter = new events_1.EventEmitter();
+                    eventEmitterTransport = new EventEmitterTransport_1.default(emitter, "foo://in", "foo://out");
+                    return [4 /*yield*/, eventEmitterTransport.connect()];
+                case 1:
+                    _a.sent();
+                    return [2 /*return*/];
+            }
+        });
+    }); });
     it("can close", function () {
@@ -22,31 +69,92 @@ var emitter = new events_1.EventEmitter();
     });
-    it("can send and receive data", function (done) {
-        var emitter = new events_1.EventEmitter();
-        var eventEmitterTransport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
-        eventEmitterTransport.connect().then(function () {
-            var eventEmitterServerTransport = new EventEmitterTransport_1.default(emitter, "to1", "from1");
-            eventEmitterServerTransport.sendData(JSON.stringify({ foo: "bar" }));
+    it("can send and receive data", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, eventEmitterTransport, result;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    emitter = new events_1.EventEmitter();
+                    eventEmitter_1.addMockServerTransport(emitter, "to1://asdf/rpc-request", "from1");
+                    eventEmitterTransport = new EventEmitterTransport_1.default(emitter, "from1", "to1://asdf/rpc-request");
+                    return [4 /*yield*/, eventEmitterTransport.connect()];
+                case 1:
+                    _a.sent();
+                    return [4 /*yield*/, eventEmitterTransport.sendData({
+                            request: requestData_1.generateMockRequest(1, "foo", ["bar"]),
+                            internalID: 1,
+                        })];
+                case 2:
+                    result = _a.sent();
+                    expect(result.method).toEqual("foo");
+                    expect(result.params).toEqual(["bar"]);
+                    return [2 /*return*/];
+            }
         });
-        eventEmitterTransport.onData(function (data) {
-            var d = JSON.parse(data);
-            expect(d.foo).toEqual("bar");
-            done();
+    }); });
+    it("can send notifications", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, eventEmitterTransport, result;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    emitter = new events_1.EventEmitter();
+                    eventEmitter_1.addMockServerTransport(emitter, "to1://asdf/rpc-notification", "from1");
+                    eventEmitterTransport = new EventEmitterTransport_1.default(emitter, "from1", "to1://asdf/rpc-notification");
+                    return [4 /*yield*/, eventEmitterTransport.connect()];
+                case 1:
+                    _a.sent();
+                    return [4 /*yield*/, eventEmitterTransport.sendData({
+                            request: requestData_1.generateMockNotificationRequest("foo", ["bar"]),
+                            internalID: 1,
+                        })];
+                case 2:
+                    result = _a.sent();
+                    expect(result).toEqual(undefined);
+                    return [2 /*return*/];
+            }
         });
-    });
-    it("can handle multiple calls to onData", function (done) {
-        var emitter = new events_1.EventEmitter();
-        var eventEmitterTransport = new EventEmitterTransport_1.default(emitter, "from1", "to1");
-        eventEmitterTransport.connect().then(function () {
-            var eventEmitterServerTransport = new EventEmitterTransport_1.default(emitter, "to1", "from1");
-            eventEmitterServerTransport.sendData(JSON.stringify({ foo: "bar" }));
+    }); });
+    it("should throw error on bad response", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, eventEmitterTransport;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    emitter = new events_1.EventEmitter();
+                    eventEmitter_1.addMockServerTransport(emitter, "to1://asdf/rpc-error", "from1");
+                    eventEmitterTransport = new EventEmitterTransport_1.default(emitter, "from1", "to1://asdf/rpc-error");
+                    return [4 /*yield*/, eventEmitterTransport.connect()];
+                case 1:
+                    _a.sent();
+                    return [4 /*yield*/, expect(eventEmitterTransport.sendData({
+                            request: requestData_1.generateMockRequest(1, "foo", ["bar"]),
+                            internalID: 1,
+                        }))
+                            .rejects.toThrowError("Error message")];
+                case 2:
+                    _a.sent();
+                    return [2 /*return*/];
+            }
         });
-        eventEmitterTransport.onData(function () {
-            // noop
+    }); });
+    it("should throw error on bad protocol", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var emitter, eventEmitterTransport;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    emitter = new events_1.EventEmitter();
+                    eventEmitter_1.addMockServerTransport(emitter, "to1://asdf/rpc-error", "from1");
+                    eventEmitterTransport = new EventEmitterTransport_1.default(emitter, "from1", "to1://asdf/rpc-error");
+                    return [4 /*yield*/, eventEmitterTransport.connect()];
+                case 1:
+                    _a.sent();
+                    eventEmitterTransport.connection.emit = function () { throw new Error("failed protocol"); };
+                    return [4 /*yield*/, expect(eventEmitterTransport.sendData({
+                            request: requestData_1.generateMockRequest(1, "foo", ["bar"]),
+                            internalID: 1,
+                        }))
+                            .rejects.toThrowError("failed protocol")];
+                case 2:
+                    _a.sent();
+                    return [2 /*return*/];
+            }
         });
-        eventEmitterTransport.onData(function (data) {
-            var d = JSON.parse(data);
-            expect(d.foo).toEqual("bar");
-            done();
-        });
-    });
+    }); });
 });

build/transports/HTTPTransport.d.ts

@@ -1,11 +0,11 @@
-import ITransport from "./Transport";
-declare class HTTPTransport implements ITransport {
-    private uri;
-    private onDataCallbacks;
+import { Transport } from "./Transport";
+import { JSONRPCRequestData } from "../Request";
+declare class HTTPTransport extends Transport {
+    uri: string;
     constructor(uri: string);
     connect(): Promise<any>;
-    onData(callback: (data: string) => any): void;
-    sendData(data: string): void;
+    sendData(data: JSONRPCRequestData, timeout?: number): Promise<any>;
     close(): void;
+    private onlyNotifications;
 }
 export default HTTPTransport;

build/transports/HTTPTransport.js

 "use strict";
+var __extends = (this && this.__extends) || (function () {
+    var extendStatics = function (d, b) {
+        extendStatics = Object.setPrototypeOf ||
+            ({ __proto__: [] } instanceof Array && function (d, b) { d.__proto__ = b; }) ||
+            function (d, b) { for (var p in b) if (b.hasOwnProperty(p)) d[p] = b[p]; };
+        return extendStatics(d, b);
+    };
+    return function (d, b) {
+        extendStatics(d, b);
+        function __() { this.constructor = d; }
+        d.prototype = b === null ? Object.create(b) : (__.prototype = b.prototype, new __());
+    };
+})();
+var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
+    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
+    return new (P || (P = Promise))(function (resolve, reject) {
+        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
+        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
+        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
+        step((generator = generator.apply(thisArg, _arguments || [])).next());
+    });
+};
+var __generator = (this && this.__generator) || function (thisArg, body) {
+    var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
+    return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
+    function verb(n) { return function (v) { return step([n, v]); }; }
+    function step(op) {
+        if (f) throw new TypeError("Generator is already executing.");
+        while (_) try {
+            if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1])).done) return t;
+            if (y = 0, t) op = [op[0] & 2, t.value];
+            switch (op[0]) {
+                case 0: case 1: t = op; break;
+                case 4: _.label++; return { value: op[1], done: false };
+                case 5: _.label++; y = op[1]; op = [0]; continue;
+                case 7: op = _.ops.pop(); _.trys.pop(); continue;
+                default:
+                    if (!(t = _.trys, t = t.length > 0 && t[t.length - 1]) && (op[0] === 6 || op[0] === 2)) { _ = 0; continue; }
+                    if (op[0] === 3 && (!t || (op[1] > t[0] && op[1] < t[3]))) { _.label = op[1]; break; }
+                    if (op[0] === 6 && _.label < t[1]) { _.label = t[1]; t = op; break; }
+                    if (t && _.label < t[2]) { _.label = t[2]; _.ops.push(op); break; }
+                    if (t[2]) _.ops.pop();
+                    _.trys.pop(); continue;
+            }
+            op = body.call(thisArg, _);
+        } catch (e) { op = [6, e]; y = 0; } finally { f = t = 0; }
+        if (op[0] & 5) throw op[1]; return { value: op[0] ? op[1] : void 0, done: true };
+    }
+};
 var __importDefault = (this && this.__importDefault) || function (mod) {
@@ -7,6 +56,17 @@ return (mod && mod.__esModule) ? mod : { "default": mod };
 var isomorphic_fetch_1 = __importDefault(require("isomorphic-fetch"));
-var HTTPTransport = /** @class */ (function () {
+var Transport_1 = require("./Transport");
+var Request_1 = require("../Request");
+var Error_1 = require("../Error");
+var HTTPTransport = /** @class */ (function (_super) {
+    __extends(HTTPTransport, _super);
     function HTTPTransport(uri) {
-        this.onDataCallbacks = [];
-        this.uri = uri;
+        var _this = _super.call(this) || this;
+        _this.onlyNotifications = function (data) {
+            if (data instanceof Array) {
+                return data.every(function (datum) { return datum.request.request.id === null || datum.request.request.id === undefined; });
+            }
+            return (data.request.id === null || data.request.id === undefined);
+        };
+        _this.uri = uri;
+        return _this;
     }
@@ -16,26 +76,54 @@ HTTPTransport.prototype.connect = function () {
     };
-    HTTPTransport.prototype.onData = function (callback) {
-        this.onDataCallbacks.push(callback);
-    };
-    HTTPTransport.prototype.sendData = function (data) {
-        var _this = this;
-        isomorphic_fetch_1.default(this.uri, {
-            method: "POST",
-            headers: {
-                "Content-Type": "application/json",
-            },
-            body: data,
-        }).then(function (result) {
-            return result.text();
-        }).then(function (result) {
-            _this.onDataCallbacks.map(function (cb) {
-                cb(result);
+    HTTPTransport.prototype.sendData = function (data, timeout) {
+        return __awaiter(this, void 0, void 0, function () {
+            var prom, notifications, batch, result, body, responseErr, e_1, responseErr;
+            return __generator(this, function (_a) {
+                switch (_a.label) {
+                    case 0:
+                        prom = this.transportRequestManager.addRequest(data, timeout);
+                        notifications = Request_1.getNotifications(data);
+                        batch = Request_1.getBatchRequests(data);
+                        _a.label = 1;
+                    case 1:
+                        _a.trys.push([1, 4, , 5]);
+                        return [4 /*yield*/, isomorphic_fetch_1.default(this.uri, {
+                                method: "POST",
+                                headers: {
+                                    "Content-Type": "application/json",
+                                },
+                                body: JSON.stringify(this.parseData(data)),
+                            })];
+                    case 2:
+                        result = _a.sent();
+                        // requirements are that notifications are successfully sent
+                        this.transportRequestManager.settlePendingRequest(notifications);
+                        if (this.onlyNotifications(data)) {
+                            return [2 /*return*/, Promise.resolve()];
+                        }
+                        return [4 /*yield*/, result.text()];
+                    case 3:
+                        body = _a.sent();
+                        responseErr = this.transportRequestManager.resolveResponse(body);
+                        if (responseErr) {
+                            // requirements are that batch requuests are successfully resolved
+                            // this ensures that individual requests within the batch request are settled
+                            this.transportRequestManager.settlePendingRequest(batch, responseErr);
+                            return [2 /*return*/, Promise.reject(responseErr)];
+                        }
+                        return [3 /*break*/, 5];
+                    case 4:
+                        e_1 = _a.sent();
+                        responseErr = new Error_1.JSONRPCError(e_1.message, Error_1.ERR_UNKNOWN, e_1);
+                        this.transportRequestManager.settlePendingRequest(notifications, responseErr);
+                        this.transportRequestManager.settlePendingRequest(Request_1.getBatchRequests(data), responseErr);
+                        return [2 /*return*/, Promise.reject(responseErr)];
+                    case 5: return [2 /*return*/, prom];
+                }
             });
         });
     };
-    HTTPTransport.prototype.close = function () {
-        this.onDataCallbacks = [];
-    };
+    // tslint:disable-next-line:no-empty
+    HTTPTransport.prototype.close = function () { };
     return HTTPTransport;
-}());
+}(Transport_1.Transport));
 exports.default = HTTPTransport;

build/transports/HTTPTransport.test.js

 "use strict";
+var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
+    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
+    return new (P || (P = Promise))(function (resolve, reject) {
+        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
+        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
+        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
+        step((generator = generator.apply(thisArg, _arguments || [])).next());
+    });
+};
+var __generator = (this && this.__generator) || function (thisArg, body) {
+    var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
+    return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
+    function verb(n) { return function (v) { return step([n, v]); }; }
+    function step(op) {
+        if (f) throw new TypeError("Generator is already executing.");
+        while (_) try {
+            if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1])).done) return t;
+            if (y = 0, t) op = [op[0] & 2, t.value];
+            switch (op[0]) {
+                case 0: case 1: t = op; break;
+                case 4: _.label++; return { value: op[1], done: false };
+                case 5: _.label++; y = op[1]; op = [0]; continue;
+                case 7: op = _.ops.pop(); _.trys.pop(); continue;
+                default:
+                    if (!(t = _.trys, t = t.length > 0 && t[t.length - 1]) && (op[0] === 6 || op[0] === 2)) { _ = 0; continue; }
+                    if (op[0] === 3 && (!t || (op[1] > t[0] && op[1] < t[3]))) { _.label = op[1]; break; }
+                    if (op[0] === 6 && _.label < t[1]) { _.label = t[1]; t = op; break; }
+                    if (t && _.label < t[2]) { _.label = t[2]; _.ops.push(op); break; }
+                    if (t[2]) _.ops.pop();
+                    _.trys.pop(); continue;
+            }
+            op = body.call(thisArg, _);
+        } catch (e) { op = [6, e]; y = 0; } finally { f = t = 0; }
+        if (op[0] & 5) throw op[1]; return { value: op[0] ? op[1] : void 0, done: true };
+    }
+};
 var __importDefault = (this && this.__importDefault) || function (mod) {
     return (mod && mod.__esModule) ? mod : { "default": mod };
 };
+var __importStar = (this && this.__importStar) || function (mod) {
+    if (mod && mod.__esModule) return mod;
+    var result = {};
+    if (mod != null) for (var k in mod) if (Object.hasOwnProperty.call(mod, k)) result[k] = mod[k];
+    result["default"] = mod;
+    return result;
+};
 Object.defineProperty(exports, "__esModule", { value: true });
 var HTTPTransport_1 = __importDefault(require("./HTTPTransport"));
+var reqMocks = __importStar(require("../__mocks__/requestData"));
 describe("HTTPTransport", function () {
     it("can connect", function () {
-        var wst = new HTTPTransport_1.default("http://localhost:8545");
-        return wst.connect();
+        var httpTransport = new HTTPTransport_1.default("http://localhost:8545");
+        return httpTransport.connect();
     });
     it("can close", function () {
-        var wst = new HTTPTransport_1.default("http://localhost:8545");
-        wst.close();
+        var httpTransport = new HTTPTransport_1.default("http://localhost:8545");
+        httpTransport.close();
     });
-    it("can send and receive data", function (done) {
-        var wst = new HTTPTransport_1.default("http://localhost:8545");
-        wst.onData(function (data) {
-            var d = JSON.parse(data);
-            expect(d.foo).toEqual("bar");
-            done();
+    it("can send and retrieve request data", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var httpTransport, data, result;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    httpTransport = new HTTPTransport_1.default("http://localhost:8545/rpc-request");
+                    data = reqMocks.generateMockRequest(1, "foo", ["bar"]);
+                    return [4 /*yield*/, httpTransport.sendData({ request: data, internalID: 1 })];
+                case 1:
+                    result = _a.sent();
+                    expect(result.method).toEqual("foo");
+                    expect(result.params).toEqual(["bar"]);
+                    return [2 /*return*/];
+            }
         });
-        wst.sendData(JSON.stringify({ foo: "bar" }));
-    });
+    }); });
+    it("can send notification data", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var httpTransport, data, result;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    httpTransport = new HTTPTransport_1.default("http://localhost:8545/rpc-notification");
+                    data = reqMocks.generateMockNotificationRequest("foo", ["bar"]);
+                    return [4 /*yield*/, httpTransport.sendData({ request: data, internalID: 1 })];
+                case 1:
+                    result = _a.sent();
+                    expect(result).toEqual(undefined);
+                    return [2 /*return*/];
+            }
+        });
+    }); });
+    it("should throw error on error response", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var httpTransport, data;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    httpTransport = new HTTPTransport_1.default("http://localhost:8545/rpc-error");
+                    data = reqMocks.generateMockRequest(9, "foo", ["bar"]);
+                    return [4 /*yield*/, expect(httpTransport.sendData({ request: data, internalID: 9 })).rejects.toThrowError("Error message")];
+                case 1:
+                    _a.sent();
+                    return [2 /*return*/];
+            }
+        });
+    }); });
+    it("should throw error on bad data response", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var httpTransport, data;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    httpTransport = new HTTPTransport_1.default("http://localhost:8545/rpc-garbage");
+                    data = { request: reqMocks.generateMockRequest(9, "foo", ["bar"]), internalID: 9 };
+                    return [4 /*yield*/, expect(httpTransport.sendData(data)).rejects.toThrowError("Bad response format")];
+                case 1:
+                    _a.sent();
+                    return [2 /*return*/];
+            }
+        });
+    }); });
+    it("should throw error on bad data response from a batch", function (done) { return __awaiter(void 0, void 0, void 0, function () {
+        var httpTransport, data;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    httpTransport = new HTTPTransport_1.default("http://localhost:8545/rpc-garbage");
+                    data = {
+                        resolve: function (d) { return ({}); },
+                        reject: function (e) {
+                            expect(e.message).toContain("Bad response format");
+                            done();
+                        },
+                        request: { request: reqMocks.generateMockRequest(9, "foo", ["bar"]), internalID: 9 },
+                    };
+                    return [4 /*yield*/, expect(httpTransport.sendData([data])).rejects.toThrow("Bad response format")];
+                case 1:
+                    _a.sent();
+                    return [2 /*return*/];
+            }
+        });
+    }); });
+    it("should throw error if unknown server crash", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var httpTransport, data;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    httpTransport = new HTTPTransport_1.default("http://localhost:8545/crash");
+                    data = { request: reqMocks.generateMockRequest(9, "foo", ["bar"]), internalID: 9 };
+                    return [4 /*yield*/, expect(httpTransport.sendData(data)).rejects.toThrowError("Random Segfault that crashes fetch")];
+                case 1:
+                    _a.sent();
+                    return [2 /*return*/];
+            }
+        });
+    }); });
 });

build/transports/Transport.d.ts

@@ -1,6 +0,45 @@
-export default interface ITransport {
-    connect(): Promise<any>;
-    close(): void;
-    onData(callback: (data: string) => any): void;
-    sendData(data: string): void;
+/// <reference types="node" />
+import { JSONRPCRequestData, IJSONRPCNotificationResponse, IJSONRPCResponse } from "../Request";
+import StrictEventEmitter from "strict-event-emitter-types";
+import { EventEmitter } from "events";
+import { JSONRPCError } from "../Error";
+import { TransportRequestManager } from "./TransportRequestManager";
+interface ITransportEvents {
+    pending: (data: JSONRPCRequestData) => void;
+    notification: (data: IJSONRPCNotificationResponse) => void;
+    response: (data: IJSONRPCResponse) => void;
+    error: (data: JSONRPCError) => void;
 }
+declare type TransportEventName = keyof ITransportEvents;
+export declare type TransportEventChannel = StrictEventEmitter<EventEmitter, ITransportEvents>;
+export declare abstract class Transport {
+    protected transportRequestManager: TransportRequestManager;
+    constructor();
+    abstract connect(): Promise<any>;
+    abstract close(): void;
+    abstract sendData(data: JSONRPCRequestData, timeout?: number): Promise<any>;
+    subscribe(event: TransportEventName, handler: ITransportEvents[TransportEventName]): void;
+    protected parseData(data: JSONRPCRequestData): import("../Request").IJSONRPCRequest | import("../Request").IJSONRPCNotification | (import("../Request").IJSONRPCRequest | import("../Request").IJSONRPCNotification)[];
+}
+export declare type promiseResolve = (r?: {} | PromiseLike<{}> | undefined) => void;
+export declare type promiseReject = (r?: any) => void;
+export interface IRequestPromise {
+    resolve: promiseResolve;
+    reject: promiseReject;
+}
+export declare type NotificationResponse = "notification";
+export declare type RequestResponse = "response";
+export declare type BadResponse = "error";
+export declare type TransportResponse = JSONRPCError | undefined;
+interface IHttpTransportResponse {
+    type: "http";
+    id?: string | number;
+    error?: Error;
+    payload: string;
+}
+interface IWSTransportResponse {
+    type: "ws";
+    payload: string;
+}
+export declare type TransportResponseData = IHttpTransportResponse | IWSTransportResponse;
+export {};

build/transports/Transport.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+var TransportRequestManager_1 = require("./TransportRequestManager");
+var Transport = /** @class */ (function () {
+    function Transport() {
+        this.transportRequestManager = new TransportRequestManager_1.TransportRequestManager();
+        // add a noop for the error event to not require handling the error event
+        // tslint:disable-next-line:no-empty
+        this.transportRequestManager.transportEventChannel.on("error", function () { });
+    }
+    Transport.prototype.subscribe = function (event, handler) {
+        this.transportRequestManager.transportEventChannel.addListener(event, handler);
+    };
+    Transport.prototype.parseData = function (data) {
+        if (data instanceof Array) {
+            return data.map(function (batch) { return batch.request.request; });
+        }
+        return data.request;
+    };
+    return Transport;
+}());
+exports.Transport = Transport;

build/transports/WebSocketTransport.d.ts

 /// <reference types="ws" />
 import WS from "isomorphic-ws";
-import ITransport from "./Transport";
-declare class WebSocketTransport implements ITransport {
+import { Transport } from "./Transport";
+import { JSONRPCRequestData } from "../Request";
+declare class WebSocketTransport extends Transport {
     connection: WS;
-    private onDataCallbacks;
+    uri: string;
     constructor(uri: string);
     connect(): Promise<any>;
-    onData(callback: (data: string) => void): void;
-    sendData(data: any): void;
+    sendData(data: JSONRPCRequestData, timeout?: number | undefined): Promise<any>;
     close(): void;
 }
 export default WebSocketTransport;

build/transports/WebSocketTransport.js

 "use strict";
+var __extends = (this && this.__extends) || (function () {
+    var extendStatics = function (d, b) {
+        extendStatics = Object.setPrototypeOf ||
+            ({ __proto__: [] } instanceof Array && function (d, b) { d.__proto__ = b; }) ||
+            function (d, b) { for (var p in b) if (b.hasOwnProperty(p)) d[p] = b[p]; };
+        return extendStatics(d, b);
+    };
+    return function (d, b) {
+        extendStatics(d, b);
+        function __() { this.constructor = d; }
+        d.prototype = b === null ? Object.create(b) : (__.prototype = b.prototype, new __());
+    };
+})();
+var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
+    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
+    return new (P || (P = Promise))(function (resolve, reject) {
+        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
+        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
+        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
+        step((generator = generator.apply(thisArg, _arguments || [])).next());
+    });
+};
+var __generator = (this && this.__generator) || function (thisArg, body) {
+    var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
+    return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
+    function verb(n) { return function (v) { return step([n, v]); }; }
+    function step(op) {
+        if (f) throw new TypeError("Generator is already executing.");
+        while (_) try {
+            if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1])).done) return t;
+            if (y = 0, t) op = [op[0] & 2, t.value];
+            switch (op[0]) {
+                case 0: case 1: t = op; break;
+                case 4: _.label++; return { value: op[1], done: false };
+                case 5: _.label++; y = op[1]; op = [0]; continue;
+                case 7: op = _.ops.pop(); _.trys.pop(); continue;
+                default:
+                    if (!(t = _.trys, t = t.length > 0 && t[t.length - 1]) && (op[0] === 6 || op[0] === 2)) { _ = 0; continue; }
+                    if (op[0] === 3 && (!t || (op[1] > t[0] && op[1] < t[3]))) { _.label = op[1]; break; }
+                    if (op[0] === 6 && _.label < t[1]) { _.label = t[1]; t = op; break; }
+                    if (t && _.label < t[2]) { _.label = t[2]; _.ops.push(op); break; }
+                    if (t[2]) _.ops.pop();
+                    _.trys.pop(); continue;
+            }
+            op = body.call(thisArg, _);
+        } catch (e) { op = [6, e]; y = 0; } finally { f = t = 0; }
+        if (op[0] & 5) throw op[1]; return { value: op[0] ? op[1] : void 0, done: true };
+    }
+};
 var __importDefault = (this && this.__importDefault) || function (mod) {
@@ -7,6 +56,12 @@ return (mod && mod.__esModule) ? mod : { "default": mod };
 var isomorphic_ws_1 = __importDefault(require("isomorphic-ws"));
-var WebSocketTransport = /** @class */ (function () {
+var Transport_1 = require("./Transport");
+var Request_1 = require("../Request");
+var Error_1 = require("../Error");
+var WebSocketTransport = /** @class */ (function (_super) {
+    __extends(WebSocketTransport, _super);
     function WebSocketTransport(uri) {
-        this.connection = new isomorphic_ws_1.default(uri);
-        this.onDataCallbacks = [];
+        var _this = _super.call(this) || this;
+        _this.uri = uri;
+        _this.connection = new isomorphic_ws_1.default(uri);
+        return _this;
     }
@@ -21,15 +76,29 @@ WebSocketTransport.prototype.connect = function () {
             _this.connection.addEventListener("open", cb);
-            _this.connection.addEventListener("message", function (ev) {
-                _this.onDataCallbacks.map(function (callback) {
-                    callback(ev.data);
+            _this.connection.addEventListener("message", function (message) {
+                var data = message.data;
+                _this.transportRequestManager.resolveResponse(data);
+            });
+        });
+    };
+    WebSocketTransport.prototype.sendData = function (data, timeout) {
+        if (timeout === void 0) { timeout = 5000; }
+        return __awaiter(this, void 0, void 0, function () {
+            var prom, notifications;
+            var _this = this;
+            return __generator(this, function (_a) {
+                prom = this.transportRequestManager.addRequest(data, timeout);
+                notifications = Request_1.getNotifications(data);
+                this.connection.send(this.parseData(data), function (err) {
+                    if (err) {
+                        var jsonError = new Error_1.JSONRPCError(err.message, Error_1.ERR_UNKNOWN, err);
+                        _this.transportRequestManager.settlePendingRequest(notifications, jsonError);
+                        _this.transportRequestManager.settlePendingRequest(Request_1.getBatchRequests(data), jsonError);
+                        prom = Promise.reject(jsonError);
+                    }
+                    _this.transportRequestManager.settlePendingRequest(notifications);
                 });
+                return [2 /*return*/, prom];
             });
         });
     };
-    WebSocketTransport.prototype.onData = function (callback) {
-        this.onDataCallbacks.push(callback);
-    };
-    WebSocketTransport.prototype.sendData = function (data) {
-        this.connection.send(data);
-    };
     WebSocketTransport.prototype.close = function () {
@@ -39,3 +108,3 @@ this.connection.close();
     return WebSocketTransport;
-}());
+}(Transport_1.Transport));
 exports.default = WebSocketTransport;

build/transports/WebSocketTransport.test.js

 "use strict";
+var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
+    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
+    return new (P || (P = Promise))(function (resolve, reject) {
+        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
+        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
+        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
+        step((generator = generator.apply(thisArg, _arguments || [])).next());
+    });
+};
+var __generator = (this && this.__generator) || function (thisArg, body) {
+    var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
+    return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
+    function verb(n) { return function (v) { return step([n, v]); }; }
+    function step(op) {
+        if (f) throw new TypeError("Generator is already executing.");
+        while (_) try {
+            if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1])).done) return t;
+            if (y = 0, t) op = [op[0] & 2, t.value];
+            switch (op[0]) {
+                case 0: case 1: t = op; break;
+                case 4: _.label++; return { value: op[1], done: false };
+                case 5: _.label++; y = op[1]; op = [0]; continue;
+                case 7: op = _.ops.pop(); _.trys.pop(); continue;
+                default:
+                    if (!(t = _.trys, t = t.length > 0 && t[t.length - 1]) && (op[0] === 6 || op[0] === 2)) { _ = 0; continue; }
+                    if (op[0] === 3 && (!t || (op[1] > t[0] && op[1] < t[3]))) { _.label = op[1]; break; }
+                    if (op[0] === 6 && _.label < t[1]) { _.label = t[1]; t = op; break; }
+                    if (t && _.label < t[2]) { _.label = t[2]; _.ops.push(op); break; }
+                    if (t[2]) _.ops.pop();
+                    _.trys.pop(); continue;
+            }
+            op = body.call(thisArg, _);
+        } catch (e) { op = [6, e]; y = 0; } finally { f = t = 0; }
+        if (op[0] & 5) throw op[1]; return { value: op[0] ? op[1] : void 0, done: true };
+    }
+};
 var __importDefault = (this && this.__importDefault) || function (mod) {
@@ -7,2 +43,3 @@ return (mod && mod.__esModule) ? mod : { "default": mod };
 var WebSocketTransport_1 = __importDefault(require("./WebSocketTransport"));
+var requestData_1 = require("../__mocks__/requestData");
 describe("WebSocketTransport", function () {
@@ -17,27 +54,76 @@ it("can connect", function () {
     });
-    it("can send and receive data", function (done) {
-        var wst = new WebSocketTransport_1.default("http://localhost:8545");
-        wst.connect().then(function () {
-            wst.sendData(JSON.stringify({ foo: "bar" }));
+    it("can send and receive data", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var wst, result;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    wst = new WebSocketTransport_1.default("http://localhost:8545/rpc-request");
+                    return [4 /*yield*/, wst.connect()];
+                case 1:
+                    _a.sent();
+                    return [4 /*yield*/, wst.sendData({ request: requestData_1.generateMockRequest(1, "foo", ["bar"]), internalID: 1 })];
+                case 2:
+                    result = _a.sent();
+                    expect(result.method).toEqual("foo");
+                    expect(result.params).toEqual(["bar"]);
+                    return [2 /*return*/];
+            }
         });
-        wst.onData(function (data) {
-            var d = JSON.parse(data);
-            expect(d.foo).toEqual("bar");
-            done();
+    }); });
+    it("can send and receive data against potential timeout", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var wst, result;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    wst = new WebSocketTransport_1.default("http://localhost:8545/rpc-request");
+                    return [4 /*yield*/, wst.connect()];
+                case 1:
+                    _a.sent();
+                    return [4 /*yield*/, wst.sendData({ request: requestData_1.generateMockRequest(1, "foo", ["bar"]), internalID: 1 }, 10000)];
+                case 2:
+                    result = _a.sent();
+                    expect(result.method).toEqual("foo");
+                    expect(result.params).toEqual(["bar"]);
+                    return [2 /*return*/];
+            }
         });
-    });
-    it("can handle multiple onData callbacks", function (done) {
-        var wst = new WebSocketTransport_1.default("http://localhost:8545");
-        wst.connect().then(function () {
-            wst.sendData(JSON.stringify({ foo: "bar" }));
+    }); });
+    it("can send and receive errors", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var wst;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    wst = new WebSocketTransport_1.default("http://localhost:8545/rpc-error");
+                    return [4 /*yield*/, wst.connect()];
+                case 1:
+                    _a.sent();
+                    return [4 /*yield*/, expect(wst.sendData({
+                            request: requestData_1.generateMockRequest(1, "foo", ["bar"]),
+                            internalID: 1,
+                        })).rejects.toThrowError("Error message")];
+                case 2:
+                    _a.sent();
+                    return [2 /*return*/];
+            }
         });
-        wst.onData(function () {
-            // noop
+    }); });
+    it("can handle underlying transport crash", function () { return __awaiter(void 0, void 0, void 0, function () {
+        var wst;
+        return __generator(this, function (_a) {
+            switch (_a.label) {
+                case 0:
+                    wst = new WebSocketTransport_1.default("http://localhost:8545/crash");
+                    return [4 /*yield*/, wst.connect()];
+                case 1:
+                    _a.sent();
+                    return [4 /*yield*/, expect(wst.sendData({
+                            request: requestData_1.generateMockRequest(1, "foo", ["bar"]),
+                            internalID: 1,
+                        })).rejects.toThrowError("Random Segfault that crashes fetch")];
+                case 2:
+                    _a.sent();
+                    return [2 /*return*/];
+            }
         });
-        wst.onData(function (data) {
-            var d = JSON.parse(data);
-            expect(d.foo).toEqual("bar");
-            done();
-        });
-    });
+    }); });
 });

CHANGELOG.md

@@ -0,1 +1,8 @@
+# [1.2.0](https://github.com/open-rpc/client-js/compare/1.1.1...1.2.0) (2019-09-17)
+
+
+### Features
+
+* refactor clientjs to support err handling ([c3686c9](https://github.com/open-rpc/client-js/commit/c3686c9))
+
 ## [1.1.1](https://github.com/open-rpc/client-js/compare/1.1.0...1.1.1) (2019-08-22)
@@ -2,0 +9,0 @@

jest.config.js

@@ -8,3 +8,4 @@ module.exports = {
   "testEnvironment": "jsdom",
-  "preset": "ts-jest"
+  "preset": "ts-jest",
+  "coveragePathIgnorePatterns": ["Error.ts"],
 }

package.json

 {
   "name": "@open-rpc/client-js",
-  "version": "1.1.1",
+  "version": "1.2.0",
   "description": " A browser-compatible JSON-RPC client with multiple transports.",
@@ -35,4 +35,5 @@ "main": "build/index.js",
     "isomorphic-ws": "^4.0.1",
+    "strict-event-emitter-types": "^2.0.0",
     "ws": "^7.0.0"
   }
 }

src/__mocks__/isomorphic-fetch.ts

@@ -0,5 +1,10 @@
+import * as req from "./requestData";
+
 const Fetch = (url: string, options: any): Promise<any> => {
+  if (url.match(/crash/)) {
+    throw new Error("Random Segfault that crashes fetch");
+  }
   const resultPromise = {
     text: () => {
-      return Promise.resolve(options.body);
+      return Promise.resolve(req.generateMockResponseData(url, options.body));
     },
@@ -6,0 +11,0 @@ };

src/__mocks__/isomorphic-ws.ts

@@ -0,5 +1,9 @@
+import * as req from "./requestData";
+
 class WebSocket {
   private callbacks: any;
-  constructor(uri: string, props: any) {
+  private url: string;
+  constructor(url: string, props: any) {
     this.callbacks = {};
+    this.url = url;
   }
@@ -17,6 +21,17 @@ public addEventListener(eventName: string, callback: any) {
   }
-  public send(data: any) {
-    Object.entries(this.callbacks).forEach(([eventName, callback]: [string, any]) => {
+  public send(data: any, callback: (err?: Error) => void) {
+
+    if (this.url.match(/crash-null/)) {
+      callback();
+      return;
+    }
+    if (this.url.match(/crash/)) {
+      callback(new Error("Random Segfault that crashes fetch"));
+      return;
+    }
+
+    Object.entries(this.callbacks).forEach(([eventName, cb]: [string, any]) => {
       if (eventName === "message") {
-        callback({data});
+        cb({ data: req.generateMockResponseData(this.url, data) });
+        callback();
       }
@@ -29,2 +44,3 @@ });
 }
+
 export default WebSocket;

src/index.test.ts

@@ -6,5 +6,2 @@ import Client from ".";
 
-jest.mock("./RequestManager");
-
-const mockedRequestManager = RequestManager as jest.Mock<RequestManager>;
 describe("client-js", () => {
@@ -24,20 +21,36 @@ it("can be constructed", () => {
 
+  it("has a notify method that returns a promise", () => {
+    const emitter = new EventEmitter();
+    const c = new Client(new RequestManager([new EventEmitterTransport(emitter, "from1", "to1")]));
+    expect(typeof c.request).toEqual("function");
+    expect(typeof c.notify("my_method", null).then).toEqual("function");
+  });
+
+  it("can register error and subscription handlers", () => {
+    const emitter = new EventEmitter();
+    const c = new Client(new RequestManager([new EventEmitterTransport(emitter, "from1", "to1")]));
+    // tslint:disable-next-line:no-empty
+    c.onError((err) => { });
+    // tslint:disable-next-line:no-empty
+    c.onNotification((data) => { });
+  });
+
   describe("startBatch", () => {
-    it("calls the requestManager.startBatch", () => {
+    it("calls startBatch", () => {
       const emitter = new EventEmitter();
-      const rm = new mockedRequestManager([new EventEmitterTransport(emitter, "from1", "to1")]);
+      const rm = new RequestManager([new EventEmitterTransport(emitter, "from1", "to1")]);
       const c = new Client(rm);
       c.startBatch();
-      expect(mockedRequestManager.mock.instances[0].startBatch).toHaveBeenCalled();
+      //      expect(mockedRequestManager.mock.instances[0].startBatch).toHaveBeenCalled();
     });
   });
 
-  describe("stopBatch", () => {
+  describe("can call stopBatch", () => {
     const emitter = new EventEmitter();
-    const rm = new mockedRequestManager([new EventEmitterTransport(emitter, "from1", "to1")]);
+    const rm = new RequestManager([new EventEmitterTransport(emitter, "from1", "to1")]);
     const c = new Client(rm);
     c.startBatch();
     c.stopBatch();
-    expect(mockedRequestManager.mock.instances[0].startBatch).toHaveBeenCalled();
   });
+
 });

src/index.ts

@@ -5,2 +5,3 @@ import RequestManager from "./RequestManager";
 import WebSocketTransport from "./transports/WebSocketTransport";
+import { JSONRPCError } from "./Error";
 
@@ -72,6 +73,19 @@ interface IClient {
    */
-  public async request(method: string, params: any) {
+  public async request(method: string, params: any, timeout?: number) {
     await this.requestManager.connectPromise;
-    return this.requestManager.request(method, params);
+    return this.requestManager.request(method, params, false, timeout);
   }
+
+  public async notify(method: string, params: any) {
+    await this.requestManager.connectPromise;
+    return this.requestManager.request(method, params, true);
+  }
+
+  public onNotification(callback: (data: any) => void) {
+    this.requestManager.requestChannel.addListener("notification", callback);
+  }
+
+  public onError(callback: (data: JSONRPCError) => void) {
+    this.requestManager.requestChannel.addListener("error", callback);
+  }
 }
@@ -78,0 +92,0 @@

src/RequestManager.test.ts

 import RequestManager from "./RequestManager";
 import EventEmitterTransport from "./transports/EventEmitterTransport";
 import { EventEmitter } from "events";
+import { addMockServerTransport } from "./__mocks__/eventEmitter";
+import { JSONRPCError } from "./Error";
 
 describe("client-js", () => {
 
-  it("can be constructed", () => {
+  it("can be constructed and connect", () => {
     const emitter = new EventEmitter();
@@ -14,47 +16,41 @@ const transport = new EventEmitterTransport(emitter, "from1", "to1");
 
-  it("has a request method that returns a promise", () => {
+  it("can close", () => {
     const emitter = new EventEmitter();
     const transport = new EventEmitterTransport(emitter, "from1", "to1");
     const c = new RequestManager([transport]);
-    expect(typeof c.request).toEqual("function");
-    expect(typeof c.request("my_method", null).then).toEqual("function");
+    c.close();
   });
 
-  it("can connect", async () => {
+  it("can send a request", async () => {
     const emitter = new EventEmitter();
-    const transport = new EventEmitterTransport(emitter, "from1", "to1");
+    addMockServerTransport(emitter, "to1://local/rpc-request", "from1");
+    const transport = new EventEmitterTransport(emitter, "from1", "to1://local/rpc-request");
     const c = new RequestManager([transport]);
-    return c.connect();
+    const result = await c.request("foo", ["bar"]);
+    expect(result.method).toEqual("foo");
+    expect(result.params).toEqual(["bar"]);
   });
 
-  it("can close", () => {
+  it("can error on error response", async () => {
     const emitter = new EventEmitter();
-    const transport = new EventEmitterTransport(emitter, "from1", "to1");
+    addMockServerTransport(emitter, "to1://local/rpc-error", "from1");
+    const transport = new EventEmitterTransport(emitter, "from1", "to1://local/rpc-error");
     const c = new RequestManager([transport]);
-    c.close();
-  });
+    await expect(c.request("foo", ["bar"])).rejects.toThrowError("Error message");
+    });
 
-  it("can send a request", async () => {
+  it("can error on malformed response and recieve error", async () => {
     const emitter = new EventEmitter();
-    const transport = new EventEmitterTransport(emitter, "from1", "to1");
-    const serverTransport = new EventEmitterTransport(emitter, "to1", "from1");
+    addMockServerTransport(emitter, "to1://local/rpc-garbage", "from1");
+    const transport = new EventEmitterTransport(emitter, "from1", "to1://local/rpc-garbage");
     const c = new RequestManager([transport]);
-    await c.connect();
-    const reqPromise = c.request("foo", []);
-    serverTransport.sendData(JSON.stringify({ id: 0, result: { foo: "foofoo" } }));
-    await expect(reqPromise).resolves.toEqual({ foo: "foofoo" });
-  });
-
-  it("can error on malformed response", (done) => {
-    const emitter = new EventEmitter();
-    const transport = new EventEmitterTransport(emitter, "from1", "to1");
-    const serverTransport = new EventEmitterTransport(emitter, "to1", "from1");
-    const c = new RequestManager([transport]);
-    c.connect().then(() => {
-      c.request("foo", []).catch((e) => {
-        expect(e.message).toContain("Malformed");
-        done();
+    const unknownError = new Promise((resolve) => {
+      c.requestChannel.on("error", (d) => {
+        resolve(d);
       });
-      serverTransport.sendData(JSON.stringify({ id: 0, foo: "bar" }));
     });
+    await expect(c.request("foo", ["bar"], false, 1000))
+      .rejects.toThrowError("Request timeout request took longer than 1000 ms to resolve");
+    const formatError = await unknownError as JSONRPCError;
+    expect(formatError.message).toContain("Bad response format");
   });
@@ -66,122 +62,54 @@
     const c = new RequestManager([transport]);
-    await c.connect();
     expect(() => c.stopBatch()).toThrow();
   });
 
-  it("can return errors on batch requests", (done) => {
+  it("can return errors on batch requests", async () => {
     const emitter = new EventEmitter();
-    const transport = new EventEmitterTransport(emitter, "from1", "to1");
-    const serverTransport = new EventEmitterTransport(emitter, "to1", "from1");
+    addMockServerTransport(emitter, "to1://local/rpc-error", "from1");
+    const transport = new EventEmitterTransport(emitter, "from1", "to1://local/rpc-error");
 
     const c = new RequestManager([transport]);
-    c.connect().then(() => {
-      c.startBatch();
-      const requests = [
-        c.request("foo", []),
-        c.request("foo", []),
-      ];
-      Promise.all(requests).catch((e) => {
-        expect(e).toEqual({
-          code: 509,
-          message: "too much 509",
-          data: {
-            test: "data",
-          },
-        });
-        c.close();
-        done();
-      });
-      c.stopBatch();
-      serverTransport.sendData(JSON.stringify([
-        {
-          jsonrpc: "2.0",
-          id: "0",
-          error: {
-            code: 509,
-            message: "too much 509",
-            data: {
-              test: "data",
-            },
-          },
-        },
-        {
-          jsonrpc: "2.0",
-          id: "1",
-          result: "bar",
-        },
-      ]));
-
-    });
+    c.startBatch();
+    const requests = [
+      c.request("foo", ["bar"]),
+      c.request("foo", ["bar"]),
+    ];
+    c.stopBatch();
+    await expect(Promise.all(requests)).rejects.toThrowError("Error message");
   });
 
-  it("can batch a request", (done) => {
+  it("can batch a request", async () => {
+
     const emitter = new EventEmitter();
-    const transport = new EventEmitterTransport(emitter, "from1", "to1");
-    const serverTransport = new EventEmitterTransport(emitter, "to1", "from1");
+    addMockServerTransport(emitter, "to1://local/rpc-request", "from1");
+    const transport = new EventEmitterTransport(emitter, "from1", "to1://local/rpc-request");
 
     const c = new RequestManager([transport]);
-    c.connect().then(() => {
-      c.startBatch();
-      const requests = [
+    c.startBatch();
+    const requests = [
         c.request("foo", []),
-        c.request("foo", []),
-      ];
-      c.stopBatch();
-      Promise.all(requests).then(([a, b]) => {
-        expect(a).toEqual("foo");
-        expect(b).toEqual("bar");
-        c.close();
-        done();
-      });
-      serverTransport.sendData(JSON.stringify([
-        {
-          jsonrpc: "2.0",
-          id: 0,
-          result: "foo",
-        },
-        {
-          jsonrpc: "2.0",
-          id: 1,
-          result: "bar",
-        },
-      ]));
-    });
+        c.request("foo", ["bar"]),
+    ];
+    c.stopBatch();
+    const [a, b] = await Promise.all(requests);
+    expect(a.method).toEqual("foo");
+    expect(b.method).toEqual("foo");
+    expect(a.params).toEqual([]);
+    expect(b.params).toEqual(["bar"]);
   });
 
-  it("can send a request and error", (done) => {
+  it("can batch a notifications", async () => {
+
     const emitter = new EventEmitter();
-    const transport = new EventEmitterTransport(emitter, "from1", "to1");
-    const serverTransport = new EventEmitterTransport(emitter, "to1", "from1");
-    const c = new RequestManager([transport]);
-    c.connect().then(() => {
-      c.request("foo", [])
-        .catch((e) => {
-          expect(e.message).toEqual("out of order");
-          done();
-        });
-      serverTransport.sendData(JSON.stringify({
-        jsonrpc: "2.0",
-        id: 0,
-        error: {
-          code: 0,
-          message: "out of order",
-          data: {
-            foo: "bar",
-          },
-        },
-      }));
-    });
-  });
+    addMockServerTransport(emitter, "to1://local/rpc-request", "from1");
+    const transport = new EventEmitterTransport(emitter, "from1", "to1://local/rpc-request");
 
-  it("onData throws if the ID is not found", async () => {
-    const emitter = new EventEmitter();
-    const transport = new EventEmitterTransport(emitter, "from1", "to1");
-    const serverTransport = new EventEmitterTransport(emitter, "to1", "from1");
     const c = new RequestManager([transport]);
-    await c.connect();
-    expect(() => serverTransport.sendData(JSON.stringify({
-      jsonrpc: "2.0",
-      id: 10,
-      result: 123,
-    }))).toThrow("Received an unrecognized response id: 10. Valid ids are: ");
+    c.startBatch();
+    const requests = [
+        c.request("foo", [], true),
+        c.request("foo", ["bar"], true),
+    ];
+    c.stopBatch();
+    const [a, b] = await Promise.all(requests);
   });
@@ -188,0 +116,0 @@

src/RequestManager.ts

@@ -1,28 +0,13 @@
-import ITransport from "./transports/Transport";
+import { Transport } from "./transports/Transport";
+import { IJSONRPCRequest, IJSONRPCNotification, IBatchRequest } from "./Request";
+import { JSONRPCError } from "./Error";
+import StrictEventEmitter from "strict-event-emitter-types";
+import { EventEmitter } from "events";
 
-interface IJSONRPCRequest {
-  jsonrpc: "2.0";
-  id: string | number;
-  method: string;
-  params: any[] | object;
-}
-interface IJSONRPCError {
-  code: number;
-  message: string;
-  data: any;
-}
+export type RequestChannel = StrictEventEmitter<EventEmitter, IRequestEvents>;
 
-interface IJSONRPCResponse {
-  jsonrpc: "2.0";
-  id: string | number; // can also be null
-  result?: any;
-  error?: IJSONRPCError;
+export interface IRequestEvents {
+  "error": (err: JSONRPCError) => void;
+  "notification": (data: any) => void;
 }
-
-interface IJSONRPCNotification {
-  jsonrpc: "2.0";
-  method: string;
-  params: any[] | object;
-}
-
 /*
@@ -33,6 +18,8 @@ ** Naive Request Manager, only use 1st transport.
  */
+
 class RequestManager {
-  public transports: ITransport[];
+  public transports: Transport[];
   public connectPromise: Promise<any>;
-  public batch: IJSONRPCRequest[] = [];
+  public batch: IBatchRequest[] = [];
+  public requestChannel: RequestChannel;
   private requests: any;
@@ -42,6 +29,7 @@ private batchStarted: boolean = false;
 
-  constructor(transports: ITransport[]) {
+  constructor(transports: Transport[]) {
     this.transports = transports;
     this.requests = {};
     this.connectPromise = this.connect();
+    this.requestChannel = new EventEmitter();
   }
@@ -51,29 +39,23 @@
     return Promise.all(this.transports.map(async (transport) => {
-      transport.onData(this.onData.bind(this));
+      transport.subscribe("error", this.handleError.bind(this));
+      transport.subscribe("notification", this.handleNotification.bind(this));
       await transport.connect();
     }));
   }
+  public getPrimaryTransport(): Transport {
+    return this.transports[0];
+  }
 
-  public async request(method: string, params: any): Promise<any> {
-    const i = (++this.lastId).toString();
-
+  public async request(method: string, params: any[], notification: boolean = false, timeout?: number): Promise<any> {
+    const internalID = (++this.lastId).toString();
+    const id = notification ? null : internalID;
     // naively grab first transport and use it
-    const transport = this.transports[0];
-
-    const payload: IJSONRPCRequest = {
-      jsonrpc: "2.0",
-      id: i,
-      method,
-      params,
-    };
-
-    return new Promise((resolve, reject) => {
-      this.requests[i] = { resolve, reject };
-
-      if (this.batchStarted) {
-        this.batch.push(payload);
-      } else {
-        transport.sendData(JSON.stringify(payload));
-      }
-    }).finally(() => this.requests[i] = undefined);
+    const payload = {request: this.makeRequest(method, params, id) , internalID};
+    if (this.batchStarted) {
+      const result = new Promise((resolve, reject) => {
+        this.batch.push({ resolve, reject, request: payload });
+      });
+      return result;
+    }
+    return this.getPrimaryTransport().sendData(payload, timeout);
   }
@@ -94,3 +76,2 @@
   public startBatch(): void {
-    if (this.batchStarted) { return; }
     this.batchStarted = true;
@@ -105,34 +86,30 @@ }
     if (this.batch.length === 0) {
+      this.batchStarted = false;
       return;
     }
 
-    const batch = JSON.stringify(this.batch);
+    this.getPrimaryTransport().sendData(this.batch);
     this.batch = [];
-    this.transports[0].sendData(batch);
+    this.batchStarted = false;
   }
 
-  private onData(data: string): void {
-    const parsedData: IJSONRPCResponse[] | IJSONRPCResponse = JSON.parse(data);
-    const results = parsedData instanceof Array ? parsedData : [parsedData];
+  private makeRequest( method: string,
+                       params: any[] | object,
+                       id?: number | string | null): IJSONRPCRequest | IJSONRPCNotification {
+    if (id) {
+      return { jsonrpc: "2.0", id, method, params };
+    }
+    return { jsonrpc: "2.0", method, params };
+  }
 
-    results.forEach((response) => {
-      const id = typeof response.id === "string" ? response.id : response.id.toString();
-      const promiseForResult = this.requests[id];
-      if (promiseForResult === undefined) {
-        throw new Error(
-          `Received an unrecognized response id: ${response.id}. Valid ids are: ${Object.keys(this.requests)}`,
-        );
-      }
+  private handleError(data: JSONRPCError) {
+    this.requestChannel.emit("error", data);
+  }
 
-      if (response.error) {
-        promiseForResult.reject(response.error);
-      } else if (response.result) {
-        promiseForResult.resolve(response.result);
-      } else {
-        promiseForResult.reject(new Error(`Malformed JSON-RPC response object: ${JSON.stringify(response)}`));
-      }
-    });
+  private handleNotification(data: any) {
+    this.requestChannel.emit(data);
   }
+
 }
 
 export default RequestManager;

src/transports/EventEmitterTransport.test.ts

 import EventEmitterTransport from "./EventEmitterTransport";
 import { EventEmitter } from "events";
+import { generateMockRequest, generateMockNotificationRequest } from "../__mocks__/requestData";
+import { addMockServerTransport } from "../__mocks__/eventEmitter";
 
 describe("EventEmitterTransport", () => {
-  it("can connect", () => {
+
+  it("can connect", async () => {
     const emitter = new EventEmitter();
     const eventEmitterTransport = new EventEmitterTransport(emitter, "foo://in", "foo://out");
-    eventEmitterTransport.connect();
+    await eventEmitterTransport.connect();
   });
+
   it("can close", () => {
@@ -17,31 +21,53 @@ const emitter = new EventEmitter();
   });
-  it("can send and receive data", (done) => {
+
+  it("can send and receive data", async () => {
     const emitter = new EventEmitter();
-    const eventEmitterTransport = new EventEmitterTransport(emitter, "from1", "to1");
-    eventEmitterTransport.connect().then(() => {
-      const eventEmitterServerTransport = new EventEmitterTransport(emitter, "to1", "from1");
-      eventEmitterServerTransport.sendData(JSON.stringify({ foo: "bar" }));
+    addMockServerTransport(emitter, "to1://asdf/rpc-request", "from1");
+    const eventEmitterTransport = new EventEmitterTransport(emitter, "from1", "to1://asdf/rpc-request");
+    await eventEmitterTransport.connect();
+    const result = await eventEmitterTransport.sendData({
+      request: generateMockRequest(1, "foo", ["bar"]),
+      internalID: 1,
     });
-    eventEmitterTransport.onData((data: any) => {
-      const d = JSON.parse(data);
-      expect(d.foo).toEqual("bar");
-      done();
-    });
+    expect(result.method).toEqual("foo");
+    expect(result.params).toEqual(["bar"]);
   });
-  it("can handle multiple calls to onData", (done) => {
+
+  it("can send notifications", async () => {
     const emitter = new EventEmitter();
-    const eventEmitterTransport = new EventEmitterTransport(emitter, "from1", "to1");
-    eventEmitterTransport.connect().then(() => {
-      const eventEmitterServerTransport = new EventEmitterTransport(emitter, "to1", "from1");
-      eventEmitterServerTransport.sendData(JSON.stringify({ foo: "bar" }));
+    addMockServerTransport(emitter, "to1://asdf/rpc-notification", "from1");
+    const eventEmitterTransport = new EventEmitterTransport(emitter, "from1", "to1://asdf/rpc-notification");
+    await eventEmitterTransport.connect();
+    const result = await eventEmitterTransport.sendData({
+      request: generateMockNotificationRequest("foo", ["bar"]),
+      internalID: 1,
     });
-    eventEmitterTransport.onData(() => {
-      // noop
-    });
-    eventEmitterTransport.onData((data: any) => {
-      const d = JSON.parse(data);
-      expect(d.foo).toEqual("bar");
-      done();
-    });
+    expect(result).toEqual(undefined);
   });
+
+  it("should throw error on bad response", async () => {
+    const emitter = new EventEmitter();
+    addMockServerTransport(emitter, "to1://asdf/rpc-error", "from1");
+    const eventEmitterTransport = new EventEmitterTransport(emitter, "from1", "to1://asdf/rpc-error");
+    await eventEmitterTransport.connect();
+    await expect(eventEmitterTransport.sendData({
+      request: generateMockRequest(1, "foo", ["bar"]),
+      internalID: 1,
+    }))
+      .rejects.toThrowError("Error message");
+  });
+
+  it("should throw error on bad protocol", async () => {
+    const emitter = new EventEmitter();
+    addMockServerTransport(emitter, "to1://asdf/rpc-error", "from1");
+    const eventEmitterTransport = new EventEmitterTransport(emitter, "from1", "to1://asdf/rpc-error");
+    await eventEmitterTransport.connect();
+    eventEmitterTransport.connection.emit = () => { throw new Error("failed protocol"); };
+    await expect(eventEmitterTransport.sendData({
+      request: generateMockRequest(1, "foo", ["bar"]),
+      internalID: 1,
+    }))
+      .rejects.toThrowError("failed protocol");
+  });
+
 });

src/transports/EventEmitterTransport.ts

 import { EventEmitter } from "events";
-import ITransport from "./Transport";
+import { Transport } from "./Transport";
+import { JSONRPCRequestData, getNotifications } from "../Request";
+import { JSONRPCError, ERR_UNKNOWN } from "../Error";
 
-class EventEmitterTransport implements ITransport {
+class EventEmitterTransport extends Transport {
   public connection: EventEmitter;
   private reqUri: string;
   private resUri: string;
-  private onDataCallbacks: any[];
-  constructor(emitter: EventEmitter, reqUri: string, resUri: string) {
-    this.onDataCallbacks = [];
-    this.connection = emitter;
+
+  constructor(destEmitter: EventEmitter, reqUri: string, resUri: string) {
+    super();
+    this.connection = destEmitter;
     this.reqUri = reqUri;
@@ -18,5 +20,3 @@ this.resUri = resUri;
     this.connection.on(this.reqUri, (data: any) => {
-      this.onDataCallbacks.map((callback: (data: string) => void) => {
-        callback(data);
-      });
+      this.transportRequestManager.resolveResponse(data);
     });
@@ -26,10 +26,17 @@ return Promise.resolve();
 
-  public onData(callback: (data: string) => void) {
-    this.onDataCallbacks.push(callback);
+  public sendData(data: JSONRPCRequestData, timeout?: number): Promise<any> {
+    const prom = this.transportRequestManager.addRequest(data, timeout);
+    const notifications = getNotifications(data);
+    const parsedData = this.parseData(data);
+    try {
+      this.connection.emit(this.resUri, parsedData);
+      this.transportRequestManager.settlePendingRequest(notifications);
+      return prom;
+    } catch (e) {
+      const responseErr = new JSONRPCError(e.message, ERR_UNKNOWN, e);
+      this.transportRequestManager.settlePendingRequest(notifications, responseErr);
+      return Promise.reject(responseErr);
+    }
   }
 
-  public sendData(data: string) {
-    this.connection.emit(this.resUri, data);
-  }
-
   public close() {
@@ -36,0 +43,0 @@ this.connection.removeAllListeners();

src/transports/HTTPTransport.test.ts

 import HTTPTransport from "./HTTPTransport";
+import * as reqMocks from "../__mocks__/requestData";
 
 describe("HTTPTransport", () => {
   it("can connect", () => {
-    const wst = new HTTPTransport("http://localhost:8545");
-    return wst.connect();
+    const httpTransport = new HTTPTransport("http://localhost:8545");
+    return httpTransport.connect();
   });
+
   it("can close", () => {
-    const wst = new HTTPTransport("http://localhost:8545");
-    wst.close();
+    const httpTransport = new HTTPTransport("http://localhost:8545");
+    httpTransport.close();
   });
-  it("can send and receive data", (done) => {
-    const wst = new HTTPTransport("http://localhost:8545");
-    wst.onData((data: any) => {
-      const d = JSON.parse(data);
-      expect(d.foo).toEqual("bar");
-      done();
-    });
-    wst.sendData(JSON.stringify({foo: "bar"}));
+
+  it("can send and retrieve request data", async () => {
+    const httpTransport = new HTTPTransport("http://localhost:8545/rpc-request");
+    const data = reqMocks.generateMockRequest(1, "foo", ["bar"]);
+    const result = await httpTransport.sendData({ request: data, internalID: 1 });
+    expect(result.method).toEqual("foo");
+    expect(result.params).toEqual(["bar"]);
   });
+
+  it("can send notification data", async () => {
+    const httpTransport = new HTTPTransport("http://localhost:8545/rpc-notification");
+    const data = reqMocks.generateMockNotificationRequest("foo", ["bar"]);
+    const result = await httpTransport.sendData({ request: data, internalID: 1 });
+    expect(result).toEqual(undefined);
+  });
+
+  it("should throw error on error response", async () => {
+    const httpTransport = new HTTPTransport("http://localhost:8545/rpc-error");
+    const data = reqMocks.generateMockRequest(9, "foo", ["bar"]);
+    await expect(httpTransport.sendData({ request: data, internalID: 9 })).rejects.toThrowError("Error message");
+  });
+
+  it("should throw error on bad data response", async () => {
+    const httpTransport = new HTTPTransport("http://localhost:8545/rpc-garbage");
+    const data = { request: reqMocks.generateMockRequest(9, "foo", ["bar"]), internalID: 9 };
+    await expect(httpTransport.sendData(data)).rejects.toThrowError("Bad response format");
+  });
+
+  it("should throw error on bad data response from a batch", async (done) => {
+    const httpTransport = new HTTPTransport("http://localhost:8545/rpc-garbage");
+    const data = {
+      resolve: (d: any) => ({}),
+      reject: (e: Error) => {
+        expect(e.message).toContain("Bad response format");
+        done();
+      },
+      request: { request: reqMocks.generateMockRequest(9, "foo", ["bar"]), internalID: 9 },
+    };
+    await expect(httpTransport.sendData([data])).rejects.toThrow("Bad response format");
+  });
+
+  it("should throw error if unknown server crash", async () => {
+    const httpTransport = new HTTPTransport("http://localhost:8545/crash");
+    const data = { request: reqMocks.generateMockRequest(9, "foo", ["bar"]), internalID: 9 };
+    await expect(httpTransport.sendData(data)).rejects.toThrowError("Random Segfault that crashes fetch");
+  });
+
 });

src/transports/HTTPTransport.ts

 import fetch from "isomorphic-fetch";
-import ITransport from "./Transport";
-
-class HTTPTransport implements ITransport {
-  private uri: string;
-  private onDataCallbacks: any[];
+import { Transport } from "./Transport";
+import { JSONRPCRequestData, getNotifications, getBatchRequests } from "../Request";
+import { ERR_UNKNOWN, JSONRPCError } from "../Error";
+class HTTPTransport extends Transport {
+  public uri: string;
   constructor(uri: string) {
-    this.onDataCallbacks = [];
+    super();
     this.uri = uri;
@@ -14,25 +14,49 @@ }
   }
-  public onData(callback: (data: string) => any) {
-    this.onDataCallbacks.push(callback);
-  }
-  public sendData(data: string) {
-    fetch(this.uri, {
-      method: "POST",
-      headers: {
-        "Content-Type": "application/json",
-      },
-      body: data,
-    }).then((result) => {
-      return result.text();
-    }).then((result) => {
-      this.onDataCallbacks.map((cb) => {
-        cb(result);
+
+  public async sendData(data: JSONRPCRequestData, timeout?: number): Promise<any> {
+    const prom = this.transportRequestManager.addRequest(data, timeout);
+    const notifications = getNotifications(data);
+    const batch = getBatchRequests(data);
+    try {
+      const result = await fetch(this.uri, {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify(this.parseData(data)),
       });
-    });
+      // requirements are that notifications are successfully sent
+      this.transportRequestManager.settlePendingRequest(notifications);
+      if (this.onlyNotifications(data)) {
+        return Promise.resolve();
+      }
+      const body = await result.text();
+      const responseErr = this.transportRequestManager.resolveResponse(body);
+      if (responseErr) {
+        // requirements are that batch requuests are successfully resolved
+        // this ensures that individual requests within the batch request are settled
+        this.transportRequestManager.settlePendingRequest(batch, responseErr);
+        return Promise.reject(responseErr);
+      }
+    } catch (e) {
+      const responseErr = new JSONRPCError(e.message, ERR_UNKNOWN, e);
+      this.transportRequestManager.settlePendingRequest(notifications, responseErr);
+      this.transportRequestManager.settlePendingRequest(getBatchRequests(data), responseErr);
+      return Promise.reject(responseErr);
+    }
+    return prom;
   }
-  public close(): void {
-    this.onDataCallbacks = [];
+
+  // tslint:disable-next-line:no-empty
+  public close(): void { }
+
+  private onlyNotifications = (data: JSONRPCRequestData) => {
+    if (data instanceof Array) {
+      return data.every((datum) => datum.request.request.id === null || datum.request.request.id === undefined);
+    }
+    return (data.request.id === null || data.request.id === undefined);
   }
+
 }
 
 export default HTTPTransport;

src/transports/Transport.ts

@@ -1,6 +0,71 @@
-export default interface ITransport {
-  connect(): Promise<any>;
-  close(): void;
-  onData(callback: (data: string) => any): void;
-  sendData(data: string): void;
+import {
+  JSONRPCRequestData,
+  IJSONRPCNotificationResponse,
+  IJSONRPCResponse,
+} from "../Request";
+
+import StrictEventEmitter from "strict-event-emitter-types";
+import { EventEmitter } from "events";
+import { JSONRPCError } from "../Error";
+import { TransportRequestManager } from "./TransportRequestManager";
+
+interface ITransportEvents {
+  pending: (data: JSONRPCRequestData) => void;
+  notification: (data: IJSONRPCNotificationResponse) => void;
+  response: (data: IJSONRPCResponse) => void;
+  error: (data: JSONRPCError) => void;
 }
+
+type TransportEventName = keyof ITransportEvents;
+export type TransportEventChannel = StrictEventEmitter<EventEmitter, ITransportEvents>;
+
+export abstract class Transport {
+  protected transportRequestManager: TransportRequestManager;
+  constructor() {
+    this.transportRequestManager = new TransportRequestManager();
+    // add a noop for the error event to not require handling the error event
+    // tslint:disable-next-line:no-empty
+    this.transportRequestManager.transportEventChannel.on("error", () => { });
+  }
+
+  public abstract connect(): Promise<any>;
+  public abstract close(): void;
+  public abstract async sendData(data: JSONRPCRequestData, timeout?: number): Promise<any>;
+
+  public subscribe(event: TransportEventName, handler: ITransportEvents[TransportEventName]) {
+    this.transportRequestManager.transportEventChannel.addListener(event, handler);
+  }
+  protected parseData(data: JSONRPCRequestData) {
+    if (data instanceof Array) {
+      return data.map((batch) => batch.request.request);
+    }
+    return data.request;
+  }
+}
+
+export type promiseResolve = (r?: {} | PromiseLike<{}> | undefined) => void;
+export type promiseReject = (r?: any) => void;
+export interface IRequestPromise {
+  resolve: promiseResolve;
+  reject: promiseReject;
+}
+
+export type NotificationResponse = "notification";
+export type RequestResponse = "response";
+export type BadResponse = "error";
+
+export type TransportResponse = JSONRPCError | undefined;
+
+interface IHttpTransportResponse {
+  type: "http";
+  id?: string | number;
+  error?: Error;
+  payload: string;
+}
+
+interface IWSTransportResponse {
+  type: "ws";
+  payload: string;
+}
+
+export type TransportResponseData = IHttpTransportResponse | IWSTransportResponse;

src/transports/WebSocketTransport.test.ts

 import WebSocketTransport from "./WebSocketTransport";
+import { generateMockRequest } from "../__mocks__/requestData";
 
 describe("WebSocketTransport", () => {
+
   it("can connect", () => {
@@ -8,2 +10,3 @@ const wst = new WebSocketTransport("http://localhost:8545");
   });
+
   it("can close", () => {
@@ -13,27 +16,36 @@ const wst = new WebSocketTransport("http://localhost:8545");
   });
-  it("can send and receive data", (done) => {
-    const wst = new WebSocketTransport("http://localhost:8545");
-    wst.connect().then(() => {
-      wst.sendData(JSON.stringify({ foo: "bar" }));
-    });
-    wst.onData((data: any) => {
-      const d = JSON.parse(data);
-      expect(d.foo).toEqual("bar");
-      done();
-    });
+
+  it("can send and receive data", async () => {
+    const wst = new WebSocketTransport("http://localhost:8545/rpc-request");
+    await wst.connect();
+    const result = await wst.sendData({ request: generateMockRequest(1, "foo", ["bar"]), internalID: 1 });
+    expect(result.method).toEqual("foo");
+    expect(result.params).toEqual(["bar"]);
   });
-  it("can handle multiple onData callbacks", (done) => {
-    const wst = new WebSocketTransport("http://localhost:8545");
-    wst.connect().then(() => {
-      wst.sendData(JSON.stringify({ foo: "bar" }));
-    });
-    wst.onData(() => {
-      // noop
-    });
-    wst.onData((data: any) => {
-      const d = JSON.parse(data);
-      expect(d.foo).toEqual("bar");
-      done();
-    });
+
+  it("can send and receive data against potential timeout", async () => {
+    const wst = new WebSocketTransport("http://localhost:8545/rpc-request");
+    await wst.connect();
+    const result = await wst.sendData({ request: generateMockRequest(1, "foo", ["bar"]), internalID: 1 }, 10000);
+    expect(result.method).toEqual("foo");
+    expect(result.params).toEqual(["bar"]);
   });
+
+  it("can send and receive errors", async () => {
+    const wst = new WebSocketTransport("http://localhost:8545/rpc-error");
+    await wst.connect();
+    await expect(wst.sendData({
+      request: generateMockRequest(1, "foo", ["bar"]),
+      internalID: 1,
+    })).rejects.toThrowError("Error message");
+  });
+
+  it("can handle underlying transport crash", async () => {
+    const wst = new WebSocketTransport("http://localhost:8545/crash");
+    await wst.connect();
+    await expect(wst.sendData({
+      request: generateMockRequest(1, "foo", ["bar"]),
+      internalID: 1,
+    })).rejects.toThrowError("Random Segfault that crashes fetch");
+  });
 });

src/transports/WebSocketTransport.ts

 import WS from "isomorphic-ws";
-import ITransport from "./Transport";
+import { Transport } from "./Transport";
+import { JSONRPCRequestData, getNotifications, getBatchRequests } from "../Request";
+import { JSONRPCError, ERR_UNKNOWN } from "../Error";
 
-class WebSocketTransport implements ITransport {
+class WebSocketTransport extends Transport {
   public connection: WS;
-  private onDataCallbacks: any[];
+  public uri: string;
+
   constructor(uri: string) {
+    super();
+    this.uri = uri;
     this.connection = new WS(uri);
-    this.onDataCallbacks = [];
   }
@@ -18,15 +22,24 @@ public connect(): Promise<any> {
       this.connection.addEventListener("open", cb);
-      this.connection.addEventListener("message", (ev: { data: string }) => {
-        this.onDataCallbacks.map((callback: (data: string) => void) => {
-          callback(ev.data);
-        });
+      this.connection.addEventListener("message", (message: { data: string }) => {
+        const { data } = message;
+        this.transportRequestManager.resolveResponse(data);
       });
     });
   }
-  public onData(callback: (data: string) => void) {
-    this.onDataCallbacks.push(callback);
+
+  public async sendData(data: JSONRPCRequestData, timeout: number | undefined = 5000): Promise<any> {
+    let prom = this.transportRequestManager.addRequest(data, timeout);
+    const notifications = getNotifications(data);
+    this.connection.send(this.parseData(data), (err?: Error) => {
+      if (err) {
+        const jsonError = new JSONRPCError(err.message, ERR_UNKNOWN, err);
+        this.transportRequestManager.settlePendingRequest(notifications, jsonError);
+        this.transportRequestManager.settlePendingRequest(getBatchRequests(data), jsonError);
+        prom = Promise.reject(jsonError);
+      }
+      this.transportRequestManager.settlePendingRequest(notifications);
+    });
+    return prom;
   }
-  public sendData(data: any) {
-    this.connection.send(data);
-  }
+
   public close(): void {
@@ -33,0 +46,0 @@ this.connection.close();

New alerts

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

Improved metrics

Total package byte prevSize

204562

increased by76.48%

Number of package files

81

increased by35%

Lines of code

3563

increased by104.18%

Worsened metrics

Dependency count

4

increased by33.33%

Number of medium supply chain risk alerts

1

increased byInfinity%

Dependency changes

• + Addedstrict-event-emitter-types@^2.0.0

• + Addedstrict-event-emitter-types@2.0.0(transitive)