Add pubsub interface definition (#54)

* Add pubsub interface definition

* Functional JsonRpcCoder

* Functional HttpProvider

* Functional WsProvider

* Functional split completed

* Split actual method subscription

* Update flow

* Flatten no user-specified array in sub call

packages/api-provider/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@polkadot/api-provider",
-  "version": "0.7.1",
+  "version": "0.8.0",
   "description": "Transport providers for the API",
   "main": "index.js",
   "keywords": [
@@ -34,7 +34,7 @@
     "nock": "^9.1.0"
   },
   "dependencies": {
-    "@polkadot/util": "^0.13.1",
+    "@polkadot/util": "^0.17.2",
     "babel-runtime": "^6.26.0",
     "isomorphic-fetch": "^2.2.1",
     "websocket": "^1.0.25"

packages/api-provider/src/coder/json/decodeResponse.js

@@ -0,0 +1,29 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { JsonRpcResponse } from '../../types';
+import type { RpcCoderState } from './types';
+
+const assert = require('@polkadot/util/assert');
+const isNumber = require('@polkadot/util/is/number');
+const isUndefined = require('@polkadot/util/is/undefined');
+
+module.exports = function decodeResponse (self: RpcCoderState, response: JsonRpcResponse): mixed {
+  assert(response, 'Empty response object received');
+
+  assert(response.jsonrpc === '2.0', 'Invalid jsonrpc field in decoded object');
+
+  assert(isNumber(response.id), 'Invalid id field in decoded object');
+
+  if (response.error) {
+    const { code, message } = response.error;
+
+    throw new Error(`[${code}]: ${message}`);
+  }
+
+  assert(!isUndefined(response.result), 'No result found in JsonRpc response');
+
+  return response.result;
+};

packages/api-provider/src/coder/json/decodeResponse.spec.js

@@ -0,0 +1,49 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const createCoder = require('./index');
+
+describe('decodeResponse', () => {
+  let coder;
+
+  beforeEach(() => {
+    coder = createCoder();
+  });
+
+  it('expects a non-empty input object', () => {
+    expect(
+      () => coder.decodeResponse()
+    ).toThrow(/Empty response/);
+  });
+
+  it('expects a valid jsonrpc field', () => {
+    expect(
+      () => coder.decodeResponse({})
+    ).toThrow(/Invalid jsonrpc/);
+  });
+
+  it('expects a valid id field', () => {
+    expect(
+      () => coder.decodeResponse({ jsonrpc: '2.0' })
+    ).toThrow(/Invalid id/);
+  });
+
+  it('expects a valid result field', () => {
+    expect(
+      () => coder.decodeResponse({ id: 1, jsonrpc: '2.0' })
+    ).toThrow(/No result/);
+  });
+
+  it('throws any error found', () => {
+    expect(
+      () => coder.decodeResponse({ id: 1, jsonrpc: '2.0', error: { code: 123, message: 'test error' } })
+    ).toThrow(/\[123\]: test error/);
+  });
+
+  it('returns the result', () => {
+    expect(
+      coder.decodeResponse({ id: 1, jsonrpc: '2.0', result: 'some result' })
+    ).toEqual('some result');
+  });
+});

packages/api-provider/src/coder/json/encodeJson.js

@@ -0,0 +1,14 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { RpcCoderState } from './types';
+
+const encodeObject = require('./encodeObject');
+
+module.exports = function encodeJson (self: RpcCoderState, method: string, params: Array<mixed>): string {
+  return JSON.stringify(
+    encodeObject(self, method, params)
+  );
+};

packages/api-provider/src/coder/json/encodeJson.spec.js

@@ -0,0 +1,19 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const createCoder = require('./index');
+
+describe('encodeJson', () => {
+  let coder;
+
+  beforeEach(() => {
+    coder = createCoder();
+  });
+
+  it('encodes a valid JsonRPC JSON string', () => {
+    expect(
+      coder.encodeJson('method', 'params')
+    ).toEqual('{"id":1,"jsonrpc":"2.0","method":"method","params":"params"}');
+  });
+});

packages/api-provider/src/coder/json/encodeObject.js

@@ -0,0 +1,16 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { JsonRpcRequest } from '../../types';
+import type { RpcCoderState } from './types';
+
+module.exports = function encodeObject (self: RpcCoderState, method: string, params: Array<mixed>): JsonRpcRequest {
+  return {
+    id: ++self.id,
+    jsonrpc: '2.0',
+    method,
+    params
+  };
+};

packages/api-provider/src/coder/json/encodeObject.spec.js

@@ -0,0 +1,29 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const createCoder = require('./index');
+
+describe('encodeObject', () => {
+  let coder;
+
+  beforeEach(() => {
+    coder = createCoder();
+  });
+
+  it('starts with id === 0 (nothing sent)', () => {
+    expect(coder.getId()).toEqual(0);
+  });
+
+  it('encodes a valid JsonRPC object', () => {
+    expect(
+      coder.encodeObject('method', 'params')
+    ).toEqual({
+      id: 1,
+      jsonrpc: '2.0',
+      method: 'method',
+      params: 'params'
+    });
+    expect(coder.getId()).toEqual(1);
+  });
+});

packages/api-provider/src/coder/json/index.js

@@ -0,0 +1,28 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { JsonRpcRequest, JsonRpcResponse } from '../../types';
+import type { RpcCoder, RpcCoderState } from './types';
+
+const decodeResponse = require('./decodeResponse');
+const encodeJson = require('./encodeJson');
+const encodeObject = require('./encodeObject');
+
+module.exports = function rpcCoder (): RpcCoder {
+  const self: RpcCoderState = {
+    id: 0
+  };
+
+  return {
+    decodeResponse: (response: JsonRpcResponse): mixed =>
+      decodeResponse(self, response),
+    encodeJson: (method: string, params: Array<mixed>): string =>
+      encodeJson(self, method, params),
+    encodeObject: (method: string, params: Array<mixed>): JsonRpcRequest =>
+      encodeObject(self, method, params),
+    getId: (): number =>
+      self.id
+  };
+};

packages/api-provider/src/coder/json/types.js

@@ -0,0 +1,17 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { JsonRpcRequest, JsonRpcResponse } from '../../types';
+
+export type RpcCoder = {
+  decodeResponse: (response: JsonRpcResponse) => mixed,
+  encodeObject: (method: string, params: Array<mixed>) => JsonRpcRequest,
+  encodeJson: (method: string, params: Array<mixed>) => string,
+  getId: () => number
+};
+
+export type RpcCoderState = {
+  id: number
+};

packages/api-provider/src/http/index.js

@@ -3,45 +3,26 @@
 // of the ISC license. See the LICENSE file for details.
 // @flow
 
-import type { ProviderInterface } from '../types';
+import type { ProviderInterface, ProviderInterface$Callback } from '../types';
 
 require('./polyfill');
 
-const assert = require('@polkadot/util/assert');
-const JsonRpcCoder = require('../jsonRpcCoder');
+const send = require('./send');
+const subscribe = require('./subscribe');
+const state = require('./state');
+const unsubscribe = require('./unsubscribe');
 
-module.exports = class HttpProvider extends JsonRpcCoder implements ProviderInterface {
-  _endpoint: string;
+module.exports = function httpProvider (endpoint: string): ProviderInterface {
+  const self = state(endpoint);
 
-  constructor (endpoint: string) {
-    super();
-
-    assert(/^http:\/\//.test(endpoint), `Endpoint should start with 'http://', received '${endpoint}'`);
-
-    this._endpoint = endpoint;
-  }
-
-  // flowlint-next-line unsafe-getters-setters:off
-  get isConnected (): boolean {
-    return true;
-  }
-
-  async send (method: string, params: Array<mixed>): Promise<mixed> {
-    const body = this.encodeJson(method, params);
-    const response = await fetch(this._endpoint, {
-      body,
-      headers: {
-        'Accept': 'application/json',
-        'Content-Length': `${body.length}`,
-        'Content-Type': 'application/json'
-      },
-      method: 'POST'
-    });
-
-    assert(response.ok, `[${response.status}]: ${response.statusText}`);
-
-    const result = await response.json();
-
-    return this.decodeResponse(result);
-  }
+  return {
+    isConnected: (): boolean =>
+      true,
+    send: (method: string, params: Array<mixed>): Promise<mixed> =>
+      send(self, method, params),
+    subscribe: (method: string, params: Array<mixed>, cb: ProviderInterface$Callback): Promise<number> =>
+      subscribe(self, method, params, cb),
+    unsubscribe: (id: number): Promise<boolean> =>
+      unsubscribe(self, id)
+  };
 };

packages/api-provider/src/http/index.spec.js

@@ -2,111 +2,30 @@
 // This software may be modified and distributed under the terms
 // of the ISC license. See the LICENSE file for details.
 
-const { mockHttp, TEST_HTTP_URL } = require('../../test/mockHttp');
+const { TEST_HTTP_URL } = require('../../test/mockHttp');
 
-const Http = require('./index');
+const createHttp = require('./index');
 
 describe('Http', () => {
   let http;
-  let mock;
-  let encodeSpy;
-  let decodeSpy;
 
   beforeEach(() => {
-    http = new Http(TEST_HTTP_URL);
-
-    encodeSpy = jest.spyOn(http, 'encodeJson');
-    decodeSpy = jest.spyOn(http, 'decodeResponse');
+    http = createHttp(TEST_HTTP_URL);
   });
 
-  afterEach(() => {
-    encodeSpy.mockRestore();
-    decodeSpy.mockRestore();
-
-    if (mock) {
-      mock.done();
-      mock = null;
-    }
+  it('always returns isConnected true', () => {
+    expect(http.isConnected()).toEqual(true);
   });
 
-  it('requires an http:// prefixed endpoint', () => {
-    expect(
-      () => new Http('ws://')
-    ).toThrow(/with 'http/);
-  });
-
-  describe('isConnected', () => {
-    it('always returns true', () => {
-      expect(http.isConnected).toEqual(true);
+  it('does not (yet) support subscribe', () => {
+    return http.subscribe().catch((error) => {
+      expect(error.message).toMatch(/has not been implemented/);
     });
   });
 
-  describe('send', () => {
-    it('encodes requests', () => {
-      mock = mockHttp([{
-        method: 'test_encoding',
-        reply: {
-          result: 'ok'
-        }
-      }]);
-
-      return http
-        .send('test_encoding', ['param'])
-        .then((result) => {
-          expect(encodeSpy).toHaveBeenCalledWith('test_encoding', ['param']);
-        });
-    });
-
-    it('decodes responses', () => {
-      mock = mockHttp([{
-        method: 'test_encoding',
-        reply: {
-          result: 'ok'
-        }
-      }]);
-
-      return http
-        .send('test_encoding', ['param'])
-        .then((result) => {
-          expect(decodeSpy).toHaveBeenCalledWith({
-            id: 1,
-            jsonrpc: '2.0',
-            result: 'ok'
-          });
-        });
-    });
-
-    it('passes the body through correctly', () => {
-      mock = mockHttp([{
-        method: 'test_body',
-        reply: {
-          result: 'ok'
-        }
-      }]);
-
-      return http
-        .send('test_body', ['param'])
-        .then((result) => {
-          expect(mock.body['test_body']).toEqual({
-            id: 1,
-            jsonrpc: '2.0',
-            method: 'test_body',
-            params: ['param']
-          });
-        });
-    });
-
-    it('throws error when !response.ok', () => {
-      mock = mockHttp([{
-        code: 500,
-        method: 'test_error'
-      }]);
-
-      return http
-        .send('test_error', [])
-        .catch((error) => {
-          expect(error.message).toMatch(/\[500\]: Internal Server/);
-        });
+  it('does not (yet) support unsubscribe', () => {
+    return http.unsubscribe().catch((error) => {
+      expect(error.message).toMatch(/has not been implemented/);
     });
   });
 });

packages/api-provider/src/http/send.js

@@ -0,0 +1,27 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { HttpState } from './types';
+
+const assert = require('@polkadot/util/assert');
+
+module.exports = async function send ({ coder, endpoint }: HttpState, method: string, params: Array<mixed>): Promise<mixed> {
+  const body = coder.encodeJson(method, params);
+  const response = await fetch(endpoint, {
+    body,
+    headers: {
+      'Accept': 'application/json',
+      'Content-Length': `${body.length}`,
+      'Content-Type': 'application/json'
+    },
+    method: 'POST'
+  });
+
+  assert(response.ok, `[${response.status}]: ${response.statusText}`);
+
+  const result = await response.json();
+
+  return coder.decodeResponse(result);
+};

packages/api-provider/src/http/send.spec.js

@@ -0,0 +1,56 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const { mockHttp, TEST_HTTP_URL } = require('../../test/mockHttp');
+
+const createHttp = require('./index');
+
+describe('send', () => {
+  let http;
+  let mock;
+
+  beforeEach(() => {
+    http = createHttp(TEST_HTTP_URL);
+  });
+
+  afterEach(() => {
+    if (mock) {
+      mock.done();
+      mock = null;
+    }
+  });
+
+  it('passes the body through correctly', () => {
+    mock = mockHttp([{
+      method: 'test_body',
+      reply: {
+        result: 'ok'
+      }
+    }]);
+
+    return http
+      .send('test_body', ['param'])
+      .then((result) => {
+        expect(mock.body['test_body']).toEqual({
+          id: 1,
+          jsonrpc: '2.0',
+          method: 'test_body',
+          params: ['param']
+        });
+      });
+  });
+
+  it('throws error when !response.ok', () => {
+    mock = mockHttp([{
+      code: 500,
+      method: 'test_error'
+    }]);
+
+    return http
+      .send('test_error', [])
+      .catch((error) => {
+        expect(error.message).toMatch(/\[500\]: Internal Server/);
+      });
+  });
+});

packages/api-provider/src/http/state.js

@@ -0,0 +1,21 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { HttpState } from './types';
+
+const assert = require('@polkadot/util/assert');
+const l = require('@polkadot/util/logger')('api-http');
+
+const coder = require('../coder/json');
+
+module.exports = function state (endpoint: string): HttpState {
+  assert(/^http:\/\//.test(endpoint), `Endpoint should start with 'http://', received '${endpoint}'`);
+
+  return {
+    coder: coder(),
+    endpoint,
+    l
+  };
+};

packages/api-provider/src/http/state.spec.js

@@ -0,0 +1,13 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const state = require('./state');
+
+describe('state', () => {
+  it('requires an http:// prefixed endpoint', () => {
+    expect(
+      () => state('ws://')
+    ).toThrow(/with 'http/);
+  });
+});

packages/api-provider/src/http/subscribe.js

@@ -0,0 +1,11 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { ProviderInterface$Callback } from '../types';
+import type { HttpState } from './types';
+
+module.exports = async function subscribe (self: HttpState, method: string, params: Array<mixed>, cb: ProviderInterface$Callback): Promise<number> {
+  throw new Error('Subscriptions has not been implemented');
+};

packages/api-provider/src/http/types.js

@@ -0,0 +1,13 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { Logger } from '@polkadot/util/types';
+import type { RpcCoder } from '../coder/json/types';
+
+export type HttpState = {
+  coder: RpcCoder,
+  endpoint: string,
+  l: Logger
+};

packages/api-provider/src/http/unsubscribe.js

@@ -0,0 +1,10 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { HttpState } from './types';
+
+module.exports = async function unsubscribe (self: HttpState, id: number): Promise<boolean> {
+  throw new Error('Subscriptions has not been implemented');
+};

packages/api-provider/src/index.js

@@ -3,10 +3,10 @@
 // of the ISC license. See the LICENSE file for details.
 // @flow
 
-const Http = require('./http');
-const Ws = require('./ws');
+const http = require('./http');
+const ws = require('./ws');
 
 module.exports = {
-  Http,
-  Ws
+  http,
+  ws
 };

packages/api-provider/src/jsonRpcCoder.js

@@ -1,52 +0,0 @@
-// Copyright 2017-2018 Jaco Greeff
-// This software may be modified and distributed under the terms
-// of the ISC license. See the LICENSE file for details.
-// @flow
-
-import type { JsonRpcRequest, JsonRpcResponse } from './types';
-
-const assert = require('@polkadot/util/assert');
-const isNumber = require('@polkadot/util/is/number');
-const isUndefined = require('@polkadot/util/is/undefined');
-
-module.exports = class JsonRpcCoder {
-  _id: number = 0;
-
-  decodeResponse (response: JsonRpcResponse): mixed {
-    assert(response, 'Empty response object received');
-
-    assert(response.jsonrpc === '2.0', 'Invalid jsonrpc field in decoded object');
-
-    assert(isNumber(response.id), 'Invalid id field in decoded object');
-
-    if (response.error) {
-      const { code, message } = response.error;
-
-      throw new Error(`[${code}]: ${message}`);
-    }
-
-    assert(!isUndefined(response.result), 'No result found in JsonRpc response');
-
-    return response.result;
-  }
-
-  encodeObject (method: string, params: Array<mixed>): JsonRpcRequest {
-    return {
-      id: ++this._id,
-      jsonrpc: '2.0',
-      method,
-      params
-    };
-  }
-
-  encodeJson (method: string, params: Array<mixed>): string {
-    return JSON.stringify(
-      this.encodeObject(method, params)
-    );
-  }
-
-  // flowlint-next-line unsafe-getters-setters:off
-  get id (): number {
-    return this._id;
-  }
-};

packages/api-provider/src/jsonRpcCoder.spec.js

@@ -1,79 +0,0 @@
-// Copyright 2017-2018 Jaco Greeff
-// This software may be modified and distributed under the terms
-// of the ISC license. See the LICENSE file for details.
-
-const JsonRpcCoder = require('./jsonRpcCoder');
-
-describe('JsonRpcCoder', () => {
-  let coder;
-
-  beforeEach(() => {
-    coder = new JsonRpcCoder();
-  });
-
-  describe('id', () => {
-    it('starts with id === 0 (nothing sent)', () => {
-      expect(coder.id).toEqual(0);
-    });
-  });
-
-  describe('decodeResponse', () => {
-    it('expects a non-empty input object', () => {
-      expect(
-        () => coder.decodeResponse()
-      ).toThrow(/Empty response/);
-    });
-
-    it('expects a valid jsonrpc field', () => {
-      expect(
-        () => coder.decodeResponse({})
-      ).toThrow(/Invalid jsonrpc/);
-    });
-
-    it('expects a valid id field', () => {
-      expect(
-        () => coder.decodeResponse({ jsonrpc: '2.0' })
-      ).toThrow(/Invalid id/);
-    });
-
-    it('expects a valid result field', () => {
-      expect(
-        () => coder.decodeResponse({ id: 1, jsonrpc: '2.0' })
-      ).toThrow(/No result/);
-    });
-
-    it('throws any error found', () => {
-      expect(
-        () => coder.decodeResponse({ id: 1, jsonrpc: '2.0', error: { code: 123, message: 'test error' } })
-      ).toThrow(/\[123\]: test error/);
-    });
-
-    it('returns the result', () => {
-      expect(
-        coder.decodeResponse({ id: 1, jsonrpc: '2.0', result: 'some result' })
-      ).toEqual('some result');
-    });
-  });
-
-  describe('encodeObject', () => {
-    it('encodes a valid JsonRPC object', () => {
-      expect(
-        coder.encodeObject('method', 'params')
-      ).toEqual({
-        id: 1,
-        jsonrpc: '2.0',
-        method: 'method',
-        params: 'params'
-      });
-      expect(coder.id).toEqual(1);
-    });
-  });
-
-  describe('encodeJson', () => {
-    it('encodes a valid JsonRPC JSON string', () => {
-      expect(
-        coder.encodeJson('method', 'params')
-      ).toEqual('{"id":1,"jsonrpc":"2.0","method":"method","params":"params"}');
-    });
-  });
-});

packages/api-provider/src/types.js

@@ -23,7 +23,11 @@ export type JsonRpcResponseBase = {
 
 export type JsonRpcResponse = JsonRpcObject & JsonRpcResponseBase;
 
-export interface ProviderInterface {
-  +isConnected: boolean;
-  send (method: string, params: Array<mixed>): Promise<mixed>;
+export type ProviderInterface$Callback = (error: ?Error, result: mixed) => void;
+
+export type ProviderInterface = {
+  isConnected (): boolean,
+  send (method: string, params: Array<mixed>): Promise<mixed>,
+  subscribe (method: string, params: Array<mixed>, cb: ProviderInterface$Callback): Promise<number>,
+  unsubscribe (id: number): Promise<boolean>
 }

packages/api-provider/src/ws/connect.js

@@ -0,0 +1,20 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { WsState } from './types';
+
+const onClose = require('./onClose');
+const onError = require('./onError');
+const onMessage = require('./onMessage');
+const onOpen = require('./onOpen');
+
+module.exports = function connect (self: WsState): void {
+  self.websocket = new WebSocket(self.endpoint);
+
+  self.websocket.onclose = onClose(self);
+  self.websocket.onerror = onError(self);
+  self.websocket.onmessage = onMessage(self);
+  self.websocket.onopen = onOpen(self);
+};

packages/api-provider/src/ws/connect.spec.js

@@ -0,0 +1,34 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const { mockWs, TEST_WS_URL } = require('../../test/mockWs');
+
+const createState = require('./state');
+const connect = require('./connect');
+
+describe('onError', () => {
+  let mock;
+
+  beforeEach(() => {
+    mock = mockWs([]);
+  });
+
+  afterEach(() => {
+    if (mock) {
+      mock.done();
+      mock = null;
+    }
+  });
+
+  it('sets up the on* handlers', () => {
+    const state = createState(TEST_WS_URL, false);
+
+    connect(state);
+
+    expect(state.websocket.onclose[0]).toBeDefined();
+    expect(state.websocket.onerror[0]).toBeDefined();
+    expect(state.websocket.onmessage[0]).toBeDefined();
+    expect(state.websocket.onopen[0]).toBeDefined();
+  });
+});

packages/api-provider/src/ws/index.js

@@ -3,128 +3,33 @@
 // of the ISC license. See the LICENSE file for details.
 // @flow
 
-import type { JsonRpcResponse, ProviderInterface } from '../types';
-
-type AwaitingType = {
-  callback: (error: ?Error, result: mixed) => void
-};
+import type { ProviderInterface, ProviderInterface$Callback } from '../types';
 
 require('./polyfill');
 
-const assert = require('@polkadot/util/assert');
-const JsonRpcCoder = require('../jsonRpcCoder');
+const connect = require('./connect');
+const send = require('./send');
+const state = require('./state');
+const subscribe = require('./subscribe');
+const unsubscribe = require('./unsubscribe');
 
-module.exports = class WsProvider extends JsonRpcCoder implements ProviderInterface {
-  _autoConnect: boolean = true;
-  _endpoint: string;
-  _handlers: { [number]: AwaitingType } = {};
-  _isConnected: boolean = false;
-  _queued: { [number]: string } = {};
-  _websocket: WebSocket;
+module.exports = function wsProvider (endpoint: string, autoConnect: boolean = true): ProviderInterface {
+  const self = state(endpoint, autoConnect);
 
-  constructor (endpoint: string, autoConnect: boolean = true) {
-    super();
-
-    assert(/^ws:\/\//.test(endpoint), `Endpoint should start with 'ws://', received '${endpoint}'`);
-
-    this._endpoint = endpoint;
-    this._autoConnect = autoConnect;
-
-    if (autoConnect) {
-      this.connect();
-    }
+  if (autoConnect) {
+    connect(self);
   }
 
-  // flowlint-next-line unsafe-getters-setters:off
-  get isConnected (): boolean {
-    return this._isConnected;
-  }
-
-  connect = () => {
-    this._websocket = new WebSocket(this._endpoint);
-
-    this._websocket.onclose = this._onClose;
-    this._websocket.onerror = this._onError;
-    this._websocket.onmessage = this._onMessage;
-    this._websocket.onopen = this._onOpen;
-  }
-
-  _onClose = () => {
-    // console.log('disconnected from', this._endpoint);
-    this._isConnected = false;
-
-    if (this._autoConnect) {
-      setTimeout(this.connect, 1000);
-    }
-  }
-
-  _onError = (error: Event) => {
-    console.error(error);
-  }
-
-  _onOpen = () => {
-    // console.log('connected to', this._endpoint);
-    this._isConnected = true;
-
-    Object.keys(this._queued).forEach((id) => {
-      try {
-        this._websocket.send(
-          // flowlint-next-line unclear-type:off
-          this._queued[((id: any): number)]
-        );
-
-        // flowlint-next-line unclear-type:off
-        delete this._queued[((id: any): number)];
-      } catch (error) {
-        console.error(error);
-      }
-    });
-  }
-
-  _onMessage = (message: MessageEvent): void => {
-    // flowlint-next-line unclear-type:off
-    const response: JsonRpcResponse = JSON.parse(((message.data: any): string));
-    const handler = this._handlers[response.id];
-
-    if (!handler) {
-      console.error(`Unable to find handler for id=${response.id}`);
-      return;
-    }
-
-    try {
-      const result = this.decodeResponse(response);
-
-      handler.callback(null, result);
-    } catch (error) {
-      handler.callback(error);
-    }
-
-    delete this._handlers[response.id];
-  }
-
-  send (method: string, params: Array<mixed>): Promise<mixed> {
-    return new Promise((resolve, reject): void => {
-      try {
-        const json = this.encodeJson(method, params);
-
-        this._handlers[this.id] = {
-          callback: (error: ?Error, result: mixed) => {
-            if (error) {
-              reject(error);
-            } else {
-              resolve(result);
-            }
-          }
-        };
-
-        if (this._isConnected) {
-          this._websocket.send(json);
-        } else {
-          this._queued[this.id] = json;
-        }
-      } catch (error) {
-        reject(error);
-      }
-    });
-  }
+  return {
+    connect: (): void =>
+      connect(self),
+    isConnected: (): boolean =>
+      self.isConnected,
+    send: (method: string, params: Array<mixed>): Promise<mixed> =>
+      send(self, method, params),
+    subscribe: (method: string, params: Array<mixed>, cb: ProviderInterface$Callback): Promise<number> =>
+      subscribe(self, method, params, cb),
+    unsubscribe: (id: number): Promise<boolean> =>
+      unsubscribe(self, id)
+  };
 };

packages/api-provider/src/ws/index.spec.js

@@ -4,223 +4,46 @@
 
 const { mockWs, TEST_WS_URL } = require('../../test/mockWs');
 
-const Ws = require('./index');
+const create = require('./index');
 
 let ws;
 let mock;
-let encodeSpy;
-let decodeSpy;
 
 function createWs (requests, autoConnect) {
   mock = mockWs(requests);
-  ws = new Ws(TEST_WS_URL, autoConnect);
-
-  encodeSpy = jest.spyOn(ws, 'encodeObject');
-  decodeSpy = jest.spyOn(ws, 'decodeResponse');
+  ws = create(TEST_WS_URL, autoConnect);
 
   return ws;
 }
 
 describe('Ws', () => {
   afterEach(() => {
-    if (encodeSpy) {
-      encodeSpy.mockRestore();
-      encodeSpy = null;
-    }
-
-    if (decodeSpy) {
-      decodeSpy.mockRestore();
-      decodeSpy = null;
-    }
-
     if (mock) {
       mock.done();
       mock = null;
     }
   });
 
-  it('requires an ws:// prefixed endpoint', () => {
+  it('returns the connected state', () => {
     expect(
-      () => new Ws('http://')
-    ).toThrow(/with 'ws/);
+      createWs([]).isConnected()
+    ).toEqual(false);
   });
 
-  describe('websocket', () => {
-    let errorSpy;
-
+  describe('pubsub', () => {
     beforeEach(() => {
       ws = createWs([]);
-
-      errorSpy = jest.spyOn(console, 'error');
     });
 
-    afterEach(() => {
-      errorSpy.mockRestore();
-    });
-
-    it('sets up the on* handlers', () => {
-      expect(ws._websocket.onclose[0]).toEqual(ws._onClose);
-      expect(ws._websocket.onerror[0]).toEqual(ws._onError);
-      expect(ws._websocket.onmessage[0]).toEqual(ws._onMessage);
-      expect(ws._websocket.onopen[0]).toEqual(ws._onOpen);
-    });
-
-    describe('_onClose', () => {
-      beforeEach(() => {
-        jest.useFakeTimers();
-      });
-
-      afterEach(() => {
-        jest.useRealTimers();
-      });
-
-      it('reconnects after delay', () => {
-        ws.connect = jest.fn();
-        ws._onClose();
-
-        expect(ws.connect).not.toHaveBeenCalled();
-
-        jest.runAllTimers();
-
-        expect(ws.connect).toHaveBeenCalled();
-      });
-
-      it('sets isConnected false', () => {
-        expect(ws.isConnected).toEqual(false);
-      });
-
-      it('does not reconnect when autoConnect false', () => {
-        mock.done();
-        ws = createWs([], false);
-
-        ws.connect = jest.fn();
-        ws._onClose();
-
-        expect(ws.connect).not.toHaveBeenCalled();
-
-        jest.runAllTimers();
-
-        expect(ws.connect).not.toHaveBeenCalled();
+    it('does not (yet) support subscribe', () => {
+      return ws.subscribe().catch((error) => {
+        expect(error.message).toMatch(/has not been implemented/);
       });
     });
 
-    describe('_onError', () => {
-      it('logs the error', () => {
-        ws._onError('test error');
-
-        expect(errorSpy).toHaveBeenCalledWith('test error');
-      });
-    });
-
-    describe('_onMessage', () => {
-      it('fails with log when handler not found', () => {
-        ws._onMessage({ data: '{"id":2}' });
-
-        expect(errorSpy).toHaveBeenCalledWith('Unable to find handler for id=2');
-      });
-    });
-  });
-
-  describe('send', () => {
-    it('handles internal errors', () => {
-      ws = createWs([]);
-
-      ws._isConnected = true;
-      ws._websocket = null;
-
-      return ws
-        .send('test_encoding', ['param'])
-        .catch((error) => {
-          expect(error).toBeDefined();
-        });
-    });
-
-    it('encodes requests', () => {
-      ws = createWs([{
-        id: 1,
-        method: 'test_encoding',
-        reply: {
-          result: 'ok'
-        }
-      }]);
-
-      return ws
-        .send('test_encoding', ['param'])
-        .then((result) => {
-          expect(encodeSpy).toHaveBeenCalledWith('test_encoding', ['param']);
-        });
-    });
-
-    it('decodes responses', () => {
-      ws = createWs([{
-        id: 1,
-        method: 'test_encoding',
-        reply: {
-          result: 'ok'
-        }
-      }]);
-
-      return ws
-        .send('test_encoding', ['param'])
-        .then((result) => {
-          expect(decodeSpy).toHaveBeenCalledWith({
-            id: 1,
-            jsonrpc: '2.0',
-            result: 'ok'
-          });
-        });
-    });
-
-    it('passes the body through correctly', () => {
-      ws = createWs([{
-        id: 1,
-        method: 'test_body',
-        reply: {
-          result: 'ok'
-        }
-      }]);
-
-      return ws
-        .send('test_body', ['param'])
-        .then((result) => {
-          expect(
-            mock.body['test_body']
-          ).toEqual('{"id":1,"jsonrpc":"2.0","method":"test_body","params":["param"]}');
-        });
-    });
-
-    it('throws error when !response.ok', () => {
-      ws = createWs([{
-        id: 1,
-        error: {
-          code: 666,
-          message: 'error'
-        }
-      }]);
-
-      return ws
-        .send('test_error', [])
-        .catch((error) => {
-          expect(error.message).toMatch(/\[666\]: error/);
-        });
-    });
-  });
-
-  describe('queued', () => {
-    it('sends messages when connected', () => {
-      const ws = createWs([{
-        id: 1,
-        method: 'test_queue',
-        reply: {
-          result: 'ok'
-        }
-      }], false);
-      const sendPromise = ws.send('test_queue', []);
-
-      ws.connect();
-
-      return sendPromise.then((result) => {
-        expect(result).toEqual('ok');
+    it('does not (yet) support unsubscribe', () => {
+      return ws.unsubscribe().catch((error) => {
+        expect(error.message).toMatch(/has not been implemented/);
       });
     });
   });

packages/api-provider/src/ws/onClose.js

@@ -0,0 +1,20 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { WsState } from './types';
+
+const connect = require('./connect');
+
+module.exports = function onClose (self: WsState): () => void {
+  return (): void => {
+    self.l.debug(() => ['disconnected from', self.endpoint]);
+
+    self.isConnected = false;
+
+    if (self.autoConnect) {
+      setTimeout(() => connect(self), 1000);
+    }
+  };
+};

packages/api-provider/src/ws/onClose.spec.js

@@ -0,0 +1,42 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const { TEST_WS_URL } = require('../../test/mockWs');
+
+const createState = require('./state');
+
+describe('onClose', () => {
+  let mockConnect;
+
+  beforeEach(() => {
+    mockConnect = jest.fn();
+
+    jest.mock('./connect', () => mockConnect);
+    jest.useFakeTimers();
+  });
+
+  afterEach(() => {
+    jest.useRealTimers();
+  });
+
+  it('reconnects after delay', () => {
+    require('./onClose')(createState(TEST_WS_URL, true))();
+
+    expect(mockConnect).not.toHaveBeenCalled();
+
+    jest.runAllTimers();
+
+    expect(mockConnect).toHaveBeenCalled();
+  });
+
+  it('does not reconnect when autoConnect false', () => {
+    require('./onClose')(createState(TEST_WS_URL, false))();
+
+    expect(mockConnect).not.toHaveBeenCalled();
+
+    jest.runAllTimers();
+
+    expect(mockConnect).not.toHaveBeenCalled();
+  });
+});

packages/api-provider/src/ws/onError.js

@@ -0,0 +1,12 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { WsState } from './types';
+
+module.exports = function onError (self: WsState): (Event) => void {
+  return (error: Event): void => {
+    self.l.error(error);
+  };
+};

packages/api-provider/src/ws/onError.spec.js

@@ -0,0 +1,28 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const onError = require('./onError');
+const createState = require('./state');
+
+describe('onError', () => {
+  let state;
+  let errorSpy;
+
+  beforeEach(() => {
+    state = createState('ws://127.0.0.1:1234', false);
+    errorSpy = jest.spyOn(console, 'error');
+  });
+
+  afterEach(() => {
+    errorSpy.mockRestore();
+  });
+
+  it('logs the error', () => {
+    onError(state)('test error');
+
+    expect(errorSpy).toHaveBeenCalledWith(
+      expect.anything(), expect.anything(), 'test error'
+    );
+  });
+});

packages/api-provider/src/ws/onMessage.js

@@ -0,0 +1,30 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { JsonRpcResponse } from '../types';
+import type { WsState } from './types';
+
+module.exports = function onMessage (self: WsState): (MessageEvent) => void {
+  return (message: MessageEvent): void => {
+    // flowlint-next-line unclear-type:off
+    const response: JsonRpcResponse = JSON.parse(((message.data: any): string));
+    const handler = self.handlers[response.id];
+
+    if (!handler) {
+      self.l.error(`Unable to find handler for id=${response.id}`);
+      return;
+    }
+
+    try {
+      const result = self.coder.decodeResponse(response);
+
+      handler.callback(null, result);
+    } catch (error) {
+      handler.callback(error);
+    }
+
+    delete self.handlers[response.id];
+  };
+};

packages/api-provider/src/ws/onMessage.spec.js

@@ -0,0 +1,28 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const onMessage = require('./onMessage');
+const createState = require('./state');
+
+describe('onError', () => {
+  let state;
+  let errorSpy;
+
+  beforeEach(() => {
+    state = createState('ws://127.0.0.1:1234', false);
+    errorSpy = jest.spyOn(console, 'error');
+  });
+
+  afterEach(() => {
+    errorSpy.mockRestore();
+  });
+
+  it('fails with log when handler not found', () => {
+    onMessage(state)({ data: '{"id":2}' });
+
+    expect(errorSpy).toHaveBeenCalledWith(
+      expect.anything(), expect.anything(), 'Unable to find handler for id=2'
+    );
+  });
+});

packages/api-provider/src/ws/onOpen.js

@@ -0,0 +1,30 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { WsState } from './types';
+
+module.exports = function onOpen (self: WsState): () => boolean {
+  return (): boolean => {
+    self.l.debug(() => ['connected to', self.endpoint]);
+
+    self.isConnected = true;
+
+    Object.keys(self.queued).forEach((id) => {
+      try {
+        self.websocket.send(
+          // flowlint-next-line unclear-type:off
+          self.queued[((id: any): number)]
+        );
+
+        // flowlint-next-line unclear-type:off
+        delete self.queued[((id: any): number)];
+      } catch (error) {
+        self.l.error(error);
+      }
+    });
+
+    return true;
+  };
+};

packages/api-provider/src/ws/onOpen.spec.js

@@ -0,0 +1,43 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const { mockWs, TEST_WS_URL } = require('../../test/mockWs');
+
+const create = require('./index');
+
+let ws;
+let mock;
+
+function createWs (requests, autoConnect) {
+  mock = mockWs(requests);
+  ws = create(TEST_WS_URL, autoConnect);
+
+  return ws;
+}
+
+describe('onOpen', () => {
+  afterEach(() => {
+    if (mock) {
+      mock.done();
+      mock = null;
+    }
+  });
+
+  it('sends messages when connected', () => {
+    const ws = createWs([{
+      id: 1,
+      method: 'test_queue',
+      reply: {
+        result: 'ok'
+      }
+    }], false);
+    const sendPromise = ws.send('test_queue', []);
+
+    ws.connect();
+
+    return sendPromise.then((result) => {
+      expect(result).toEqual('ok');
+    });
+  });
+});

packages/api-provider/src/ws/send.js

@@ -0,0 +1,34 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { WsState } from './types';
+
+module.exports = async function send (self: WsState, method: string, params: Array<mixed>): Promise<mixed> {
+  return new Promise((resolve, reject): void => {
+    try {
+      const json = self.coder.encodeJson(method, params);
+      const id = self.coder.getId();
+      const callback = (error: ?Error, result: mixed) => {
+        if (error) {
+          reject(error);
+        } else {
+          resolve(result);
+        }
+      };
+
+      self.handlers[id] = {
+        callback
+      };
+
+      if (self.isConnected) {
+        self.websocket.send(json);
+      } else {
+        self.queued[id] = json;
+      }
+    } catch (error) {
+      reject(error);
+    }
+  });
+};

packages/api-provider/src/ws/send.spec.js

@@ -0,0 +1,94 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const { mockWs, TEST_WS_URL } = require('../../test/mockWs');
+
+let ws;
+let mock;
+
+function createMock (requests) {
+  mock = mockWs(requests);
+}
+
+function createWs (autoConnect) {
+  ws = require('./index')(TEST_WS_URL, autoConnect);
+
+  return ws;
+}
+
+describe('send', () => {
+  let globalWs;
+
+  beforeEach(() => {
+    globalWs = global.WebSocket;
+  });
+
+  afterEach(() => {
+    global.WebSocket = globalWs;
+
+    if (mock) {
+      mock.done();
+      mock = null;
+    }
+  });
+
+  it('handles internal errors', () => {
+    createMock([]);
+
+    let websocket;
+
+    global.WebSocket = class {
+      constructor () {
+        websocket = this;
+      }
+
+      send () {
+        throw new Error('send error');
+      }
+    };
+
+    ws = createWs();
+    websocket.onopen();
+
+    return ws
+      .send('test_encoding', ['param'])
+      .catch((error) => {
+        expect(error.message).toEqual('send error');
+      });
+  });
+
+  it('passes the body through correctly', () => {
+    createMock([{
+      id: 1,
+      method: 'test_body',
+      reply: {
+        result: 'ok'
+      }
+    }]);
+
+    return createWs()
+      .send('test_body', ['param'])
+      .then((result) => {
+        expect(
+          mock.body['test_body']
+        ).toEqual('{"id":1,"jsonrpc":"2.0","method":"test_body","params":["param"]}');
+      });
+  });
+
+  it('throws error when !response.ok', () => {
+    createMock([{
+      id: 1,
+      error: {
+        code: 666,
+        message: 'error'
+      }
+    }]);
+
+    return createWs()
+      .send('test_error', [])
+      .catch((error) => {
+        expect(error.message).toMatch(/\[666\]: error/);
+      });
+  });
+});

packages/api-provider/src/ws/state.js

@@ -0,0 +1,25 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { WsState } from './types';
+
+const assert = require('@polkadot/util/assert');
+const l = require('@polkadot/util/logger')('api-ws');
+
+const coder = require('../coder/json');
+
+module.exports = function state (endpoint: string, autoConnect: boolean): $Shape<WsState> {
+  assert(/^ws:\/\//.test(endpoint), `Endpoint should start with 'ws://', received '${endpoint}'`);
+
+  return {
+    autoConnect,
+    coder: coder(),
+    endpoint,
+    handlers: {},
+    isConnected: false,
+    l,
+    queued: {}
+  };
+};

packages/api-provider/src/ws/state.spec.js

@@ -0,0 +1,13 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+
+const state = require('./state');
+
+describe('state', () => {
+  it('requires an ws:// prefixed endpoint', () => {
+    expect(
+      () => state('http://')
+    ).toThrow(/with 'ws/);
+  });
+});

packages/api-provider/src/ws/subscribe.js

@@ -0,0 +1,11 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { ProviderInterface$Callback } from '../types';
+import type { WsState } from './types';
+
+module.exports = async function subscribe (self: WsState, method: string, params: Array<mixed>, cb: ProviderInterface$Callback): Promise<number> {
+  throw new Error('Subscriptions has not been implemented');
+};

packages/api-provider/src/ws/types.js

@@ -0,0 +1,22 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { Logger } from '@polkadot/util/types';
+import type { RpcCoder } from '../coder/json/types';
+
+export type WsState$Awaiting = {
+  callback: (error: ?Error, result: mixed) => void
+};
+
+export type WsState = {
+  autoConnect: boolean,
+  coder: RpcCoder,
+  endpoint: string,
+  handlers: { [number]: WsState$Awaiting },
+  isConnected: boolean,
+  l: Logger,
+  queued: { [number]: string },
+  websocket: WebSocket
+};

packages/api-provider/src/ws/unsubscribe.js

@@ -0,0 +1,10 @@
+// Copyright 2017-2018 Jaco Greeff
+// This software may be modified and distributed under the terms
+// of the ISC license. See the LICENSE file for details.
+// @flow
+
+import type { WsState } from './types';
+
+module.exports = async function unsubscribe (self: WsState, id: number): Promise<boolean> {
+  throw new Error('Subscriptions has not been implemented');
+};