Handle pubsub interfaces

packages/api/src/create/interface.js

@@ -9,23 +9,22 @@ import type { ApiInterface$Section } from '../types';
 
 const interfaces = require('@polkadot/api-jsonrpc');
 
-const createMethod = require('./method');
-const createSubscribe = require('./subscribe');
+const methodSend = require('./methodSend');
+const methodSubscribe = require('./methodSubscribe');
 
 module.exports = function createInterface (provider: ProviderInterface, section: InterfaceTypes): ApiInterface$Section {
   const exposed: $Shape<ApiInterface$Section> = {};
   const { methods } = interfaces[section];
 
-  exposed.subscribe = createSubscribe(provider, section, methods);
-  exposed.unsubscribe = provider.unsubscribe;
-
   return Object
     .keys(methods)
     .reduce((exposed, name: string) => {
       const rpcName = `${section}_${name}`;
-      const method = createMethod(provider, rpcName, methods[name]);
+      const def = methods[name];
 
-      exposed[name] = method;
+      exposed[name] = def.isSubscription
+        ? methodSubscribe(provider, rpcName, name, methods[name])
+        : methodSend(provider, rpcName, name, methods[name]);
 
       return exposed;
     }, exposed);

packages/api/src/create/interface.spec.js

@@ -16,6 +16,11 @@ jest.mock('./@polkadot/api-jsonrpc', () => ({
       bleh: {
         inputs: [],
         output: { type: 'Address' }
+      },
+      pubsub: {
+        isSubscription: true,
+        inputs: [],
+        output: { type: 'Address' }
       }
     }
   }
@@ -38,7 +43,7 @@ describe('createInterface', () => {
 
   it('adds the specified methods to the interface', () => {
     expect(Object.keys(container)).toEqual(
-      ['subscribe', 'unsubscribe', 'blah', 'bleh']
+      ['blah', 'bleh', 'pubsub']
     );
   });
 

packages/api/src/create/method.js → packages/api/src/create/methodSend.js

@@ -5,17 +5,17 @@
 
 import type { InterfaceMethodDefinition } from '@polkadot/api-jsonrpc/types';
 import type { ProviderInterface } from '@polkadot/api-provider/types';
+import type { ApiInterface$Section$Method } from '../types';
 
 const formatOutput = require('@polkadot/api-format/output');
 const ExtError = require('@polkadot/util/ext/error');
 const jsonrpcSignature = require('@polkadot/util/jsonrpc/signature');
 
-type Method = (..._params: Array<mixed>) => Promise<mixed>;
-
 const createParams = require('./params');
 
-module.exports = function createMethod (provider: ProviderInterface, rpcName: string, { inputs, output }: InterfaceMethodDefinition): Method {
+module.exports = function createMethodSend (provider: ProviderInterface, rpcName: string, name: string, { inputs, output }: InterfaceMethodDefinition): $Shape<ApiInterface$Section$Method> {
   return async (..._params: Array<mixed>): Promise<mixed> => {
+    // TODO: Deprecated warning
     try {
       const params = createParams(rpcName, _params, inputs);
       const result = await provider.send(rpcName, params);

packages/api/src/create/method.spec.js → packages/api/src/create/methodSend.spec.js

@@ -2,9 +2,9 @@
 // This software may be modified and distributed under the terms
 // of the ISC license. See the LICENSE file for details.
 
-const createMethod = require('./method');
+const createMethod = require('./methodSend');
 
-describe('createMethod', () => {
+describe('methodCall', () => {
   let methods;
   let provider;
 
@@ -30,23 +30,23 @@ describe('createMethod', () => {
   });
 
   it('wraps errors with the call signature', () => {
-    const method = createMethod(provider, 'test_blah', methods.blah);
+    const method = createMethod(provider, 'test_blah', 'blah', methods.blah);
 
     return method().catch((error) => {
       expect(error.message).toMatch(/test_blah \(foo: Bytes\): Bytes/);
     });
   });
 
   it('checks for mismatched parameters', () => {
-    const method = createMethod(provider, 'test_bleh', methods.bleh);
+    const method = createMethod(provider, 'test_bleh', 'bleh', methods.bleh);
 
     return method(1).catch((error) => {
       expect(error.message).toMatch(/0 params expected, found 1 instead/);
     });
   });
 
   it('calls the provider with the correct parameters', () => {
-    const method = createMethod(provider, 'test_blah', methods.blah);
+    const method = createMethod(provider, 'test_blah', 'blah', methods.blah);
 
     return method(new Uint8Array([0x12, 0x34])).then(() => {
       expect(provider.send).toHaveBeenCalledWith('test_blah', ['0x1234']);

packages/api/src/create/subscribeMethod.js → packages/api/src/create/methodSubscribe.js

@@ -5,18 +5,17 @@
 
 import type { InterfaceMethodDefinition } from '@polkadot/api-jsonrpc/types';
 import type { ProviderInterface, ProviderInterface$Callback } from '@polkadot/api-provider/types';
+import type { ApiInterface$Section$Method } from '../types';
 
 const formatOutput = require('@polkadot/api-format/output');
 const assert = require('@polkadot/util/assert');
 const ExtError = require('@polkadot/util/ext/error');
 const isFunction = require('@polkadot/util/is/function');
 const jsonrpcSignature = require('@polkadot/util/jsonrpc/signature');
 
-type Method = (...params: Array<mixed>) => Promise<number>;
-
 const createParams = require('./params');
 
-module.exports = function createSubscribeMethod (provider: ProviderInterface, rpcName: string, { inputs, output }: InterfaceMethodDefinition): Method {
+module.exports = function methodSubscribe (provider: ProviderInterface, rpcName: string, name: string, { inputs, output }: InterfaceMethodDefinition): $Shape<ApiInterface$Section$Method> {
   return async (..._params: Array<mixed>): Promise<number> => {
     try {
       // flowlint-next-line unclear-type:off
@@ -25,14 +24,24 @@ module.exports = function createSubscribeMethod (provider: ProviderInterface, rp
       assert(isFunction(cb), `Expected callback in last position of params`);
 
       const params = createParams(rpcName, _params, inputs);
-
-      return provider.subscribe(rpcName, params, (error: ?Error, result: mixed) => {
-        if (error) {
-          cb(error);
-        } else {
-          cb(null, formatOutput(output, result));
-        }
-      });
+      let subscriptionId = -1;
+      const promise = provider
+        .subscribe(`subscribe_${name}`, params, (error: ?Error, result: mixed) => {
+          if (error) {
+            cb(error);
+          } else {
+            cb(null, formatOutput(output, result));
+          }
+        })
+        .then((_subscriptionId) => {
+          subscriptionId = _subscriptionId;
+        });
+
+      promise.unsubscribe = (): Promise<boolean> => {
+        return provider.send(`unsubscribe_${name}`, subscriptionId);
+      };
+
+      return promise;
     } catch (error) {
       throw new ExtError(`${jsonrpcSignature(rpcName, inputs, output)}:: ${error.message}`, (error: ExtError).code);
     }

packages/api/src/types.js

@@ -5,13 +5,13 @@
 
 import type { InterfaceTypes } from '@polkadot/api-jsonrpc/types';
 
-export type ApiInterface$Section$Method = (...params: Array<mixed>) => Promise<mixed>;
+export type ApiInterface$Section$Method = {
+  (...params: Array<mixed>): Promise<mixed>;
+  unsubscribe: (id: number) => Promise<boolean>;
+}
 
 export type ApiInterface$Section = {
-  [string]: ApiInterface$Section$Method,
-
-  subscribe: (name: string, ...params: Array<mixed>) => Promise<number>,
-  unsubscribe: (id: number) => Promise<boolean>
+  [string]: ApiInterface$Section$Method
 };
 
 export type ApiInterface = {