Pubsub for postMessage Provider (UI2)

Proper method-name displaying on request approval is missing

js/src/api/api.js

@@ -17,8 +17,8 @@
 import EventEmitter from 'eventemitter3';
 
 import Contract from './contract';
-import { PromiseProvider, Http as HttpProvider, PostMessage as PostMessageProvider, WsSecure as WsSecureProvider } from './provider';
-import { Http as HttpTransport, WsSecure as WsSecureTransport } from './transport';
+import { PromiseProvider, Http as HttpProvider, PostMessage as PostMessageProvider, Ws as WsProvider } from './provider';
+import { Http as HttpTransport, Ws as WsTransport } from './transport';
 
 import { Db, Eth, Parity, Net, Personal, Shell, Shh, Signer, Trace, Web3 } from './rpc';
 import Subscriptions from './subscriptions';
@@ -36,7 +36,6 @@ export default class Api extends EventEmitter {
       console.log(provider);
       console.warn(new Error('deprecated: Api needs provider with send() function, old-style Transport found instead'));
     }
-
     // does use new provider interface (not promiseProvider)
     if (provider && isFunction(provider.subscribe)) {
       this._pubsub = new Pubsub(provider);
@@ -187,12 +186,12 @@ export default class Api extends EventEmitter {
   static Provider = {
     Http: HttpProvider,
     PostMessage: PostMessageProvider,
-    WsSecure: WsSecureProvider
+    Ws: WsProvider
   }
 
   // NOTE: kept for backwards compatibility
   static Transport = {
     Http: HttpTransport,
-    WsSecure: WsSecureTransport
+    Ws: WsTransport
   }
 }

js/src/api/provider/index.js

@@ -18,4 +18,4 @@ export PromiseProvider from './promise';
 
 export Http from './http';
 export PostMessage from './postMessage';
-export WsSecure from './wsSecure';
+export Ws from './ws';

js/src/api/provider/postMessage.js

@@ -16,7 +16,7 @@
 
 export default class PostMessage {
   id = 0;
-  _callbacks = {};
+  _messages = {};
 
   constructor (token, destination) {
     this._token = token;
@@ -28,17 +28,55 @@ export default class PostMessage {
   addMiddleware () {
   }
 
+  _send (data) {
+    this._destination.postMessage(data, '*');
+  }
+
   send = (method, params, callback) => {
     const id = ++this.id;
 
-    this._callbacks[id] = callback;
-    this._destination.postMessage({
+    this._messages[id] = { callback };
+    this._send({
       id,
       from: this._token,
       method,
       params,
       token: this._token
-    }, '*');
+    });
+  }
+
+  subscribe = (api, callback, params) => {
+    console.log('paritySubscribe', JSON.stringify(params), api, callback);
+    return new Promise((resolve, reject) => {
+      const id = ++this.id;
+
+      this._messages[id] = { callback, resolve, reject, subscription: true, initial: true };
+      this._send({
+        id,
+        from: this._token,
+        api,
+        params,
+        token: this._token
+      });
+    });
+  }
+
+  unsubscribe = (subId) => {
+    return new Promise((resolve, reject) => {
+      const id = ++this.id;
+
+      this._messages[id] = { callback: (e, v) => e ? reject(e) : resolve(v) };
+      this._send({
+        id,
+        from: this._token,
+        subId,
+        token: this._token
+      });
+    });
+  }
+
+  unsubscribeAll () {
+    return this.unsubscribe('*');
   }
 
   receiveMessage = ({ data: { id, error, from, token, result }, origin, source }) => {
@@ -50,7 +88,13 @@ export default class PostMessage {
       console.error(from, error);
     }
 
-    this._callbacks[id](error && new Error(error), result);
-    this._callbacks[id] = null;
+    if (this._messages[id].subscription) {
+      console.log('subscription', result, 'initial?', this._messages[id].initial);
+      this._messages[id].initial ? this._messages[id].resolve(result) : this._messages[id].callback(error && new Error(error), result);
+      this._messages[id].initial = false;
+    } else {
+      this._messages[id].callback(error && new Error(error), result);
+      this._messages[id] = null;
+    }
   }
 }

js/src/api/provider/ws.js

@@ -14,9 +14,9 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
-import { WsSecure as Transport } from '../transport';
+import { Ws as Transport } from '../transport';
 
-export default class WsSecure extends Transport {
+export default class Ws extends Transport {
   send = (method, params, callback) => {
     this
       ._execute(method, params)

js/src/api/pubsub/pubsubBase.js

@@ -20,11 +20,11 @@ export default class PubsubBase {
     this._provider = provider;
   }
 
-  addListener (module, eventName, callback, eventParams) {
+  addListener (module, eventName, handler, eventParams) {
     return eventParams
-     ? this._provider.subscribe(module, callback, [eventName, eventParams])
-     : this._provider.subscribe(module, callback, [eventName, []]);
-    // this._transport.subscribe(module, callback, eventName);  After Patch from tomac is merged to master! =>  eth_subscribe does not support empty array as params
+     ? this._provider.subscribe(module, handler, [eventName, eventParams])
+     : this._provider.subscribe(module, handler, [eventName, []]);
+    // this._transport.subscribe(module, handler, eventName);  After Patch from tomac is merged to master! =>  eth_subscribe does not support empty array as params
   }
 
   removeListener (subscriptionIds) {

js/src/api/rpc/trace/trace.js

@@ -29,8 +29,8 @@ export default class Trace {
   }
 
   call (options, whatTrace = ['trace'], blockNumber = 'latest') {
-    return this._transport
-      .execute('trace_call', inOptions(options), inTraceType(whatTrace), inBlockNumber(blockNumber))
+    return this._provider
+      .send('trace_call', inOptions(options), inTraceType(whatTrace), inBlockNumber(blockNumber))
       .then(outTraceReplay);
   }
 

js/src/api/transport/index.js

@@ -15,6 +15,6 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 export Http from './http';
-export WsSecure from './wsSecure';
+export Ws from './ws';
 export TransportError from './error';
 export Middleware from './middleware';

js/src/api/transport/ws/index.js

@@ -14,4 +14,4 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
-export default from './wsSecure';
+export default from './ws';

js/src/api/transport/ws/ws.e2e.js

@@ -14,9 +14,9 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
-import WsSecure from './wsSecure';
+import Ws from './ws';
 
-const ws = new WsSecure('ws://localhost:8546/');
+const ws = new Ws('ws://localhost:8546/');
 
 describe('transport/WsSecure', () => {
   it('connects and makes a call to web3_clientVersion', () => {

js/src/api/transport/ws/ws.js

@@ -21,7 +21,7 @@ import JsonRpcBase from '../jsonRpcBase';
 import TransportError from '../error';
 
 /* global WebSocket */
-export default class WsSecure extends JsonRpcBase {
+export default class Ws extends JsonRpcBase {
   // token is optional (secure API)
   constructor (url, token = null, autoconnect = true) {
     super();
@@ -321,6 +321,16 @@ export default class WsSecure extends JsonRpcBase {
     });
   }
 
+  unsubscribeAll () {
+    return new Promise((resolve, reject) => {
+      var unsubscribed = 0;
+      let keys = Object.keys(this._messages);
+
+      keys.forEach(i => this._messages[i].subscription ? this.unsubscribe(this._messages[i].subId).then(_ => unsubscribed++, reject) : null);
+      resolve(unsubscribed);
+    });
+  }
+
   unsubscribe (messageId) {
     return new Promise((resolve, reject) => {
       const id = this.id;

js/src/api/transport/ws/ws.spec.js

@@ -15,16 +15,16 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 import { TEST_WS_URL, mockWs } from '../../../../test/mockRpc';
-import WsSecure from './wsSecure';
+import Ws from './ws';
 
-describe('api/transport/WsSecure', () => {
+describe('api/transport/ws', () => {
   let transport;
   let scope;
 
   describe('transport emitter', () => {
     const connect = () => {
       const scope = mockWs();
-      const transport = new WsSecure(TEST_WS_URL);
+      const transport = new Ws(TEST_WS_URL);
 
       return { transport, scope };
     };
@@ -57,7 +57,7 @@ describe('api/transport/WsSecure', () => {
 
     beforeEach(() => {
       scope = mockWs([{ method: 'test_anyCall', reply: 'TestResult' }]);
-      transport = new WsSecure(TEST_WS_URL);
+      transport = new Ws(TEST_WS_URL);
 
       return transport
         .execute('test_anyCall', [1, 2, 3])
@@ -98,7 +98,7 @@ describe('api/transport/WsSecure', () => {
   describe('errors', () => {
     beforeEach(() => {
       scope = mockWs([{ method: 'test_anyCall', reply: { error: { code: 1, message: 'TestError' } } }]);
-      transport = new WsSecure(TEST_WS_URL);
+      transport = new Ws(TEST_WS_URL);
     });
 
     afterEach(() => {

js/src/secureApi.js

@@ -35,7 +35,7 @@ export default class SecureApi extends Api {
   static getTransport (url, sysuiToken, protocol) {
     const transportUrl = SecureApi.transportUrl(url, protocol);
 
-    return new Api.Provider.WsSecure(transportUrl, sysuiToken, false);
+    return new Api.Provider.Ws(transportUrl, sysuiToken, false);
   }
 
   static transportUrl (url, protocol) {

js/src/shell/DappRequests/store.js

@@ -91,16 +91,20 @@ export default class Store {
 
   @action approveSingleRequest = ({ queueId, request: { data, source } }) => {
     this.removeRequest(queueId);
-    this.executeMethodCall(data, source);
+    if (data.api) {
+      this.executePubsubCall(data, source);
+    } else {
+      this.executeMethodCall(data, source);
+    }
   }
 
   @action approveRequest = (queueId, approveAll) => {
     const queued = this.findRequest(queueId);
 
     if (approveAll) {
-      const { request: { data: { method, token } } } = queued;
+      const { request: { data: { method, token, params } } } = queued;
 
-      this.getFilteredSection(method).methods.forEach((m) => {
+      this.getFilteredSection(method || params[0]).methods.forEach((m) => {
         this.addTokenPermission(m, token);
         this.findMatchingRequests(m, token).forEach(this.approveSingleRequest);
       });
@@ -152,7 +156,7 @@ export default class Store {
   }
 
   findMatchingRequests (_method, _token) {
-    return this.requests.filter(({ request: { data: { method, token } } }) => method === _method && token === _token);
+    return this.requests.filter(({ request: { data: { method, token, params } } }) => (method === _method || (params && params[0] === _method)) && token === _token);
   }
 
   _methodCallbackPost = (id, source, token) => {
@@ -169,6 +173,16 @@ export default class Store {
     };
   }
 
+  executePubsubCall = ({ api, id, token, params }, source) => {
+    const callback = this._methodCallbackPost(id, source, token);
+
+    // TODO: enable security pubsub
+    this.provider.subscribe(api, callback, params).then((v, e) => {
+      console.log('Error and result', v, e);
+      this._methodCallbackPost(id, source, token)(null, v);
+    });
+  }
+
   executeMethodCall = ({ id, from, method, params, token }, source) => {
     const visibleStore = VisibleStore.get();
     const callback = this._methodCallbackPost(id, source, token);
@@ -225,18 +239,27 @@ export default class Store {
       return;
     }
 
-    const { from, method, token } = data;
+    const { from, method, token, params, api, subId, id } = data;
 
     if (!from || from === 'shell' || from !== token) {
       return;
     }
 
-    if (this.getFilteredSection(method) && !this.hasTokenPermission(method, token)) {
+    if ((method && this.getFilteredSection(method) && !this.hasTokenPermission(method, token)) ||
+        (api && this.getFilteredSection(params[0]) && !this.hasTokenPermission(method, token))) {
       this.queueRequest({ data, origin, source });
       return;
     }
-
-    this.executeMethodCall(data, source);
+    if (api) {
+      console.log('apiCall', data);
+      this.executePubsubCall(data, source);
+    } else if (subId) {
+      subId === '*'
+        ? this.provider.unsubscribeAll().then(v => this._methodCallbackPost(id, source, token)(null, v))
+        : this.provider.unsubscribe(subId).then(v => this._methodCallbackPost(id, source, token)(null, v));
+    } else {
+      this.executeMethodCall(data, source);
+    }
   }
 
   static instance = null;

js/src/shell/index.js

@@ -27,6 +27,7 @@ import { IndexRoute, Redirect, Route, Router, hashHistory } from 'react-router';
 import qs from 'querystring';
 
 import Api from '@parity/api';
+import Abi from '@parity/abi';
 import builtinDapps from '@parity/shared/config/dappsBuiltin.json';
 import viewsDapps from '@parity/shared/config/dappsViews.json';
 import ContractInstances from '@parity/shared/contracts';
@@ -80,6 +81,13 @@ const dappsHistory = HistoryStore.get('dapps');
 function onEnterDapp ({ params: { id } }) {
   const token = DappRequestsStore.get().createToken(id);
 
+  // on app switch unsubscribe all subscriptions
+  if (window.ethereum) {
+    window.ethereum.unsubscribeAll();
+  }
+  // old API uses window.parity
+  window.parity = { Api, Abi };
+
   window.ethereum = new Api.Provider.PostMessage(token, window);
 
   if (!dapps[id] || !dapps[id].skipHistory) {