Expose WS subscribe/unsubscribe (#63)

* Expose WS subscribe

* Support unsubscribe

* Align unsubscribe with method

* Basic e2e sub test

* Cleanup parameter positions

* Small update

* Handle non-supplied in output formatter

packages/api-format/src/output.js

@@ -8,6 +8,7 @@ import type { InterfaceOutputType } from '@polkadot/api-jsonrpc/types';
 const bnDecode = require('@polkadot/primitives-json/bn/decode');
 const bytesDecode = require('@polkadot/primitives-json/bytes/decode');
 const headerDecode = require('@polkadot/primitives-json/header/decode');
+const isUndefined = require('@polkadot/util/is/undefined');
 
 const util = require('./util');
 
@@ -18,6 +19,10 @@ const formatters = {
   'U64': bnDecode
 };
 
-module.exports = function formatOutput (output: InterfaceOutputType, value: mixed): mixed {
+module.exports = function formatOutput (output: InterfaceOutputType, value?: mixed): ?mixed {
+  if (isUndefined(value)) {
+    return void 0;
+  }
+
   return util.format(formatters, output.type, value);
 };

packages/api-jsonrpc/src/chain/newHead.js

@@ -16,6 +16,6 @@ module.exports = ({
   isSubscription: true,
   inputs: [],
   output: {
-    type: 'HeaderHash'
+    type: 'Header'
   }
 }: InterfaceMethodDefinition);

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

@@ -3,27 +3,40 @@
 // of the ISC license. See the LICENSE file for details.
 // @flow
 
-import type { JsonRpcResponse } from '../../types';
+import type { JsonRpcResponse, JsonRpcResponseBase$Error } 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}`);
+function checkError (error?: JsonRpcResponseBase$Error) {
+  if (!error) {
+    return;
   }
 
-  assert(!isUndefined(response.result), 'No result found in JsonRpc response');
+  const { code, message } = error;
+
+  throw new Error(`[${code}]: ${message}`);
+}
+
+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');
+
+  const isSubscription = !isUndefined(response.params) && !isUndefined(response.method);
+
+  assert(isNumber(response.id) || (isSubscription && isNumber(response.params.subscription)), 'Invalid id field in decoded object');
+
+  checkError(response.error);
+
+  assert(!isUndefined(response.result) || isSubscription, 'No result found in JsonRpc response');
+
+  if (isSubscription) {
+    checkError(response.params.error);
+
+    return response.params.result;
+  }
 
   return response.result;
 };

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

@@ -22,7 +22,7 @@ module.exports = function httpProvider (endpoint: string): ProviderInterface {
       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)
+    unsubscribe: (method: string, id: number): Promise<boolean> =>
+      unsubscribe(self, method, id)
   };
 };

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

@@ -5,6 +5,6 @@
 
 import type { HttpState } from './types';
 
-module.exports = async function unsubscribe (self: HttpState, id: number): Promise<boolean> {
+module.exports = async function unsubscribe (self: HttpState, method: string, id: number): Promise<boolean> {
   throw new Error('Subscriptions has not been implemented');
 };

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

@@ -23,7 +23,7 @@ module.exports = function mockProvider (): ProviderInterface {
       send(self, method, params),
     subscribe: (method: string, ...params: Array<mixed>): Promise<number> =>
       subscribe(self, method, params),
-    unsubscribe: (id: number): Promise<boolean> =>
-      unsubscribe(self, id)
+    unsubscribe: (method: string, id: number): Promise<boolean> =>
+      unsubscribe(self, method, id)
   };
 };

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

@@ -5,7 +5,7 @@
 
 import type { MockState } from './types';
 
-module.exports = async function subscribe (self: MockState, id: number): Promise<boolean> {
+module.exports = async function subscribe (self: MockState, _: string, id: number): Promise<boolean> {
   const method = self.subscriptionMap[id];
 
   self.l.debug(() => ['unsubscribe', id, method]);

packages/api-provider/src/mock/unsubscribe.spec.js

@@ -19,17 +19,17 @@ describe('unsubscribe', () => {
   });
 
   it('fails on unknown ids', () => {
-    return unsubscribe(state, 5).catch((error) => {
+    return unsubscribe(state, 'extra_getBlockNumber', 5).catch((error) => {
       expect(error.message).toMatch(/Unable to find/);
     });
   });
 
   it('unsubscribes successfully', () => {
-    return unsubscribe(state, id);
+    return unsubscribe(state, 'extra_getBlockNumber', id);
   });
 
   it('fails on double unsubscribe', () => {
-    return unsubscribe(state, id)
+    return unsubscribe(state, 'extra_getBlockNumber', id)
       .then(() => unsubscribe(state, id))
       .catch((error) => {
         expect(error.message).toMatch(/Unable to find/);
@@ -37,7 +37,7 @@ describe('unsubscribe', () => {
   });
 
   it('clears the subscriptions', () => {
-    return unsubscribe(state, id).then(() => {
+    return unsubscribe(state, 'extra_getBlockNumber', id).then(() => {
       expect(state.subscriptionMap[id]).not.toBeDefined();
       expect(state.subscriptions['extra_getBlockNumber'].callbacks[id]).not.toBeDefined();
     });

packages/api-provider/src/types.js

@@ -13,13 +13,26 @@ export type JsonRpcRequest = JsonRpcObject & {
   params: Array<mixed>;
 };
 
-export type JsonRpcResponseBase = {
-  error?: {
-    code: number,
-    message: string
-  };
+export type JsonRpcResponseBase$Error = {
+  code: number,
+  message: string
+};
+
+type JsonRpcResponse$Single = {
+  error?: JsonRpcResponseBase$Error;
   result?: mixed;
-}
+};
+
+type JsonRpcResponse$Subscription = {
+  method?: string;
+  params: {
+    error?: JsonRpcResponseBase$Error;
+    result: mixed;
+    subscription: number;
+  }
+};
+
+export type JsonRpcResponseBase = JsonRpcResponse$Single & JsonRpcResponse$Subscription;
 
 export type JsonRpcResponse = JsonRpcObject & JsonRpcResponseBase;
 
@@ -29,5 +42,5 @@ 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>
+  unsubscribe (method: string, id: number): Promise<boolean>
 }

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

@@ -29,7 +29,7 @@ module.exports = function wsProvider (endpoint: string, autoConnect: boolean = t
       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)
+    unsubscribe: (method: string, id: number): Promise<boolean> =>
+      unsubscribe(self, method, id)
   };
 };

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

@@ -29,22 +29,4 @@ describe('Ws', () => {
       createWs([]).isConnected()
     ).toEqual(false);
   });
-
-  describe('pubsub', () => {
-    beforeEach(() => {
-      ws = createWs([]);
-    });
-
-    it('does not (yet) support subscribe', () => {
-      return ws.subscribe().catch((error) => {
-        expect(error.message).toMatch(/has not been implemented/);
-      });
-    });
-
-    it('does not (yet) support unsubscribe', () => {
-      return ws.unsubscribe().catch((error) => {
-        expect(error.message).toMatch(/has not been implemented/);
-      });
-    });
-  });
 });

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

@@ -6,25 +6,20 @@
 import type { JsonRpcResponse } from '../types';
 import type { WsState } from './types';
 
+const isUndefined = require('@polkadot/util/is/undefined');
+
+const onMessageResult = require('./onMessageResult');
+const onMessageSubscribe = require('./onMessageSubscribe');
+
 module.exports = function onMessage (self: WsState): (MessageEvent) => void {
   return (message: MessageEvent): void => {
+    self.l.debug(() => ['received', message.data]);
+
     // 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];
+    return isUndefined(response.method)
+      ? onMessageResult(self, response)
+      : onMessageSubscribe(self, response);
   };
 };

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

@@ -0,0 +1,36 @@
+// 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 onMessageResult (self: WsState, response: JsonRpcResponse): void {
+  self.l.debug(() => ['handling: response =', response, 'id =', response.id]);
+
+  const handler = self.handlers[response.id];
+
+  if (!handler) {
+    self.l.error(`Unable to find handler for id=${response.id}`);
+    return;
+  }
+
+  try {
+    const { subscription } = self.handlers[response.id];
+    const result = self.coder.decodeResponse(response);
+
+    if (subscription) {
+      // flowlint-next-line unclear-type:off
+      self.subscriptions[((result: any): number)] = {
+        callback: subscription
+      };
+    }
+
+    handler.callback(null, result);
+  } catch (error) {
+    handler.callback(error);
+  }
+
+  delete self.handlers[response.id];
+};

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

@@ -5,7 +5,7 @@
 const onMessage = require('./onMessage');
 const createState = require('./state');
 
-describe('onError', () => {
+describe('onMessageResult', () => {
   let state;
   let errorSpy;
 
@@ -25,4 +25,15 @@ describe('onError', () => {
       expect.anything(), expect.anything(), 'Unable to find handler for id=2'
     );
   });
+
+  it('calls the handler when found', (done) => {
+    state.handlers[5] = {
+      callback: (_, result) => {
+        expect(result).toEqual('test');
+        done();
+      }
+    };
+
+    onMessage(state)({ data: '{"jsonrpc":"2.0","id":5,"result":"test"}' });
+  });
 });

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

@@ -0,0 +1,26 @@
+// 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 onMessageSubscribe (self: WsState, response: JsonRpcResponse): void {
+  self.l.debug(() => ['handling: response =', response, 'subscription =', response.params.subscription]);
+
+  const handler = self.subscriptions[response.params.subscription];
+
+  if (!handler) {
+    self.l.error(`Unable to find handler for subscription=${response.params.subscription}`);
+    return;
+  }
+
+  try {
+    const result = self.coder.decodeResponse(response);
+
+    handler.callback(null, result);
+  } catch (error) {
+    handler.callback(error);
+  }
+};

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

@@ -0,0 +1,54 @@
+// 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('onMessageSubscribe', () => {
+  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: '{"method":"test","params":{"subscription":2}}' });
+
+    expect(errorSpy).toHaveBeenCalledWith(
+      expect.anything(), expect.anything(), 'Unable to find handler for subscription=2'
+    );
+  });
+
+  it('calls the subscriber with data', (done) => {
+    state.handlers[11] = {
+      callback: (_, id) => {},
+      subscription: (_, result) => {
+        expect(result).toEqual('test');
+        done();
+      }
+    };
+
+    onMessage(state)({ data: '{"jsonrpc":"2.0","id":11,"result":22}' });
+    onMessage(state)({ data: '{"jsonrpc":"2.0","method":"test","params":{"subscription":22,"result":"test"}}' });
+  });
+
+  it('calls the subscriber with error', (done) => {
+    state.handlers[11] = {
+      callback: (_, id) => {},
+      subscription: (error) => {
+        expect(error.message).toMatch(/test/);
+        done();
+      }
+    };
+
+    onMessage(state)({ data: '{"jsonrpc":"2.0","id":11,"result":22}' });
+    onMessage(state)({ data: '{"jsonrpc":"2.0","method":"test","params":{"subscription":22,"error":{"message":"test"}}}' });
+  });
+});

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

@@ -3,9 +3,10 @@
 // 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 send (self: WsState, method: string, params: Array<mixed>): Promise<mixed> {
+module.exports = async function send (self: WsState, method: string, params: Array<mixed>, subscription?: ProviderInterface$Callback): Promise<mixed> {
   return new Promise((resolve, reject): void => {
     try {
       const json = self.coder.encodeJson(method, params);
@@ -18,8 +19,11 @@ module.exports = async function send (self: WsState, method: string, params: Arr
         }
       };
 
+      self.l.debug(() => ['calling', method, params, json, !!subscription]);
+
       self.handlers[id] = {
-        callback
+        callback,
+        subscription
       };
 
       if (self.isConnected) {

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

@@ -91,4 +91,20 @@ describe('send', () => {
         expect(error.message).toMatch(/\[666\]: error/);
       });
   });
+
+  it('adds subscriptions', () => {
+    createMock([{
+      id: 1,
+      method: 'test_sub',
+      reply: {
+        result: 1
+      }
+    }]);
+
+    return createWs()
+      .send('test_sub', [], () => {})
+      .then((id) => {
+        expect(id).toEqual(1);
+      });
+  });
 });

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

@@ -20,6 +20,7 @@ module.exports = function state (endpoint: string, autoConnect: boolean): $Shape
     handlers: {},
     isConnected: false,
     l,
-    queued: {}
+    queued: {},
+    subscriptions: {}
   };
 };

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

@@ -6,6 +6,11 @@
 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');
+const send = require('./send');
+
+module.exports = async function subscribe (self: WsState, method: string, params: Array<mixed>, subscription: ProviderInterface$Callback): Promise<number> {
+  const id = await send(self, method, params, subscription);
+
+  // flowlint-next-line unclear-type:off
+  return ((id: any): number);
 };

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

@@ -0,0 +1,51 @@
+// 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('subscribe', () => {
+  let globalWs;
+
+  beforeEach(() => {
+    globalWs = global.WebSocket;
+  });
+
+  afterEach(() => {
+    global.WebSocket = globalWs;
+
+    if (mock) {
+      mock.done();
+      mock = null;
+    }
+  });
+
+  it('adds subscriptions', () => {
+    createMock([{
+      id: 1,
+      method: 'test_sub',
+      reply: {
+        result: 1
+      }
+    }]);
+
+    return createWs()
+      .subscribe('test_sub', [], () => {})
+      .then((id) => {
+        expect(id).toEqual(1);
+      });
+  });
+});

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

@@ -5,9 +5,15 @@
 
 import type { Logger } from '@polkadot/util/types';
 import type { RpcCoder } from '../coder/json/types';
+import type { ProviderInterface$Callback } from '../types';
 
 export type WsState$Awaiting = {
-  callback: (error: ?Error, result: mixed) => void
+  callback: ProviderInterface$Callback,
+  subscription?: ProviderInterface$Callback
+};
+
+export type WsState$Subscription = {
+  callback: ProviderInterface$Callback
 };
 
 export type WsState = {
@@ -18,5 +24,6 @@ export type WsState = {
   isConnected: boolean,
   l: Logger,
   queued: { [number]: string },
+  subscriptions: { [number]: WsState$Subscription },
   websocket: WebSocket
 };

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

@@ -5,6 +5,18 @@
 
 import type { WsState } from './types';
 
-module.exports = async function unsubscribe (self: WsState, id: number): Promise<boolean> {
-  throw new Error('Subscriptions has not been implemented');
+const assert = require('@polkadot/util/assert');
+const isUndefined = require('@polkadot/util/is/undefined');
+
+const send = require('./send');
+
+module.exports = async function unsubscribe (self: WsState, method: string, id: number): Promise<boolean> {
+  assert(!isUndefined(self.subscriptions[id]), `Unable to find active subscription=${id}`);
+
+  delete self.subscriptions[id];
+
+  const result = await send(self, method, [id]);
+
+  // flowlint-next-line unclear-type:off
+  return ((result: any): boolean);
 };

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

@@ -0,0 +1,83 @@
+// 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('subscribe', () => {
+  let globalWs;
+
+  beforeEach(() => {
+    globalWs = global.WebSocket;
+  });
+
+  afterEach(() => {
+    global.WebSocket = globalWs;
+
+    if (mock) {
+      mock.done();
+      mock = null;
+    }
+  });
+
+  it('removes subscriptions', () => {
+    createMock([
+      {
+        id: 1,
+        method: 'subscribe_test',
+        reply: {
+          result: 1
+        }
+      },
+      {
+        id: 2,
+        method: 'unsubscribe_test',
+        reply: {
+          result: true
+        }
+      }
+    ]);
+
+    const ws = createWs();
+
+    return ws
+      .subscribe('subscribe_test', [], () => {})
+      .then((id) => {
+        return ws.unsubscribe('unsubscribe_test', id);
+      });
+  });
+
+  it('fails when sub not found', () => {
+    createMock([{
+      id: 1,
+      method: 'subscribe_test',
+      reply: {
+        result: 1
+      }
+    }]);
+
+    const ws = createWs();
+
+    return ws
+      .subscribe('subscribe_test', [], () => {})
+      .then((id) => {
+        return ws.unsubscribe('unsubscribe_test', 111);
+      })
+      .catch((error) => {
+        expect(error.message).toMatch(/find active subscription/);
+      });
+  });
+});

packages/api-provider/test/e2e.test.js

@@ -0,0 +1,33 @@
+// 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 createApi = require('../../api/src');
+const createProvider = require('../src/ws');
+
+describe.skip('e2e', () => {
+  let api;
+
+  beforeEach(() => {
+    jest.setTimeout(30000);
+    api = createApi(createProvider('ws://127.0.0.1:9944'));
+  });
+
+  it('subscribes to client_newHead', (done) => {
+    let count = 0;
+
+    api.chain
+      .newHead((error, data) => {
+        console.log('newHead: result =', [error, data]);
+
+        expect(data).toBeDefined();
+
+        if (++count === 3) {
+          done();
+        }
+      })
+      .then((subscriptionId) => {
+        console.log('newHead: subscriptionId =', subscriptionId);
+      });
+  });
+});

packages/api/src/create/methodSubscribe.js

@@ -26,14 +26,11 @@ module.exports = function methodSubscribe (provider: ProviderInterface, rpcName:
       assert(isFunction(cb), `Expected callback in last position of params`);
 
       const params = createParams(rpcName, _params, inputs);
+      const update = (error: ?Error, result?: mixed) => {
+        cb(error, formatOutput(output, result));
+      };
 
-      return provider.subscribe(`subscribe_${name}`, params, (error: ?Error, result: mixed) => {
-        if (error) {
-          cb(error);
-        } else {
-          cb(null, formatOutput(output, result));
-        }
-      });
+      return provider.subscribe(`subscribe_${name}`, params, update);
     } catch (error) {
       throw new ExtError(`${jsonrpcSignature(rpcName, inputs, output)}:: ${error.message}`, (error: ExtError).code);
     }