finos · texodus · May 26, 2019 · May 25, 2019 · May 26, 2019 · May 26, 2019
diff --git a/examples/remote/index.css b/examples/remote/index.css
@@ -0,0 +1,31 @@
+/******************************************************************************
+ *
+ * Copyright (c) 2017, the Perspective Authors.
+ *
+ * This file is part of the Perspective library, distributed under the terms of
+ * the Apache License 2.0.  The full license can be found in the LICENSE file.
+ *
+ */
+
+ perspective-viewer {
+    position: absolute;
+    top: 0;
+    left: 0;
+    right: 0;
+    bottom: 0;
+}
+
+@media (max-width: 600px) {
+    html {
+        overflow: hidden;
+    }
+
+    body {
+        position: fixed;
+        height: 100%;
+        width: 100%;
+        margin: 0;
+        overflow: hidden;
+        touch-action: none;
+    }
+}
diff --git a/examples/remote/index.html b/examples/remote/index.html
@@ -0,0 +1,58 @@
+<!--
+
+   Copyright (c) 2017, the Perspective Authors.
+
+   This file is part of the Perspective library, distributed under the terms of
+   the Apache License 2.0.  The full license can be found in the LICENSE file.
+
+-->
+
+<!DOCTYPE html>
+<html>
+
+<head>
+
+    <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1, minimum-scale=1, user-scalable=no">
+
+    <script src="perspective.view.js"></script>
+    <script src="hypergrid.plugin.js"></script>
+    <script src="d3fc.plugin.js"></script>
+
+    <script src="perspective.js"></script>
+
+    <link rel='stylesheet' href="index.css">
+    <link rel='stylesheet' href="material.css">
+
+</head>
+
+<body>
+
+    <perspective-viewer 
+        id="view1"
+        view="d3_heatmap"
+        row-pivots='["client"]'
+        columns='["chg"]'
+        column-pivots='["name"]'>
+
+    </perspective-viewer>
+
+    <script>
+
+
+
+        window.addEventListener('WebComponentsReady', async function() {
+            var elem = document.getElementById('view1');
+            var client = perspective.worker(window.location.origin.replace('http', 'ws'));
+            var view = client.open_view('data_source_one');
+            let arrow = await view.to_arrow();
+            elem.load(arrow, {limit: 10000});
+            view.on_update(x => {
+                elem.update(x);
+            }, {mode: "row"});
+
+        });
+    </script>
+
+</body>
+
+</html>
diff --git a/examples/remote/package.json b/examples/remote/package.json
@@ -0,0 +1,17 @@
+{
+    "name": "remote",
+    "private": true,
+    "version": "0.3.0-rc.2",
+    "description": "An example of 2 Perspectives, one client and one server, streaming via Apache Arrow.",
+    "scripts": {
+        "start": "node server.js"
+    },
+    "keywords": [],
+    "license": "Apache-2.0",
+    "dependencies": {
+        "@finos/perspective": "^0.3.0-rc.2",
+        "@finos/perspective-viewer": "^0.3.0-rc.2",
+        "@finos/perspective-viewer-highcharts": "^0.3.0-rc.2",
+        "@finos/perspective-viewer-hypergrid": "^0.3.0-rc.2"
+    }
+}
diff --git a/examples/remote/server.js b/examples/remote/server.js
@@ -0,0 +1,77 @@
+/******************************************************************************
+ *
+ * Copyright (c) 2017, the Perspective Authors.
+ *
+ * This file is part of the Perspective library, distributed under the terms of
+ * the Apache License 2.0.  The full license can be found in the LICENSE file.
+ *
+ */
+
+const {WebSocketHost, table} = require("@finos/perspective");
+
+/******************************************************************************
+ *
+ * Copyright (c) 2017, the Perspective Authors.
+ *
+ * This file is part of the Perspective library, distributed under the terms of
+ * the Apache License 2.0.  The full license can be found in the LICENSE file.
+ *
+ */
+
+var SECURITIES = ["AAPL.N", "AMZN.N", "QQQ.N", "NVDA.N", "TSLA.N", "FB.N", "MSFT.N", "TLT.N", "XIV.N", "YY.N", "CSCO.N", "GOOGL.N", "PCLN.N"];
+var CLIENTS = ["Homer", "Marge", "Bart", "Lisa", "Maggie", "Moe", "Lenny", "Carl", "Krusty"];
+var CACHE_INPUT = false;
+var CACHE_ENTRIES = 200;
+var TABLE_SIZE = 10000;
+
+var __CACHE__ = {};
+
+function newRows() {
+    var rows = [];
+    for (var x = 0; x < 50; x++) {
+        rows.push({
+            name: SECURITIES[Math.floor(Math.random() * SECURITIES.length)],
+            client: CLIENTS[Math.floor(Math.random() * CLIENTS.length)],
+            lastUpdate: new Date(),
+            chg: Math.random() * 20 - 10,
+            bid: Math.random() * 10 + 90,
+            ask: Math.random() * 10 + 100,
+            vol: Math.random() * 10 + 100
+        });
+    }
+    return rows;
+}
+
+async function newArrow() {
+    var tbl = table(newRows());
+    var vw = tbl.view();
+    var arrow = await vw.to_arrow();
+    vw.delete();
+    tbl.delete();
+    return arrow;
+}
+
+const host = new WebSocketHost({assets: [__dirname]});
+
+async function init() {
+    if (CACHE_INPUT) {
+        for (let x = 0; x < CACHE_ENTRIES; x++) {
+            let arrow = await newArrow();
+            __CACHE__[x] = arrow;
+        }
+    }
+    var tbl = table(CACHE_INPUT ? __CACHE__[0] : newRows(), {
+        limit: TABLE_SIZE
+    });
+    host.host_view("data_source_one", tbl.view());
+    (function postRow() {
+        if (CACHE_INPUT) {
+            tbl.update(__CACHE__[Math.floor(Math.random() * __CACHE__.length)]);
+        } else {
+            tbl.update(newRows());
+        }
+        setTimeout(postRow, 20);
+    })();
+}
+
+init();
diff --git a/package.json b/package.json
@@ -29,7 +29,7 @@
         "html-loader-jest": "^0.2.1",
         "jest": "^24.5.0",
         "js-beautify": "^1.8.6",
-        "jsdoc": "^3.5.5",
+        "jsdoc": "3.5.5",
         "jsdoc-babel": "^0.5.0",
         "jsdoc-to-markdown": "^4.0.1",
         "lerna": "^2.11.0",

diff --git a/packages/perspective-bench/src/html/benchmark.html b/packages/perspective-bench/src/html/benchmark.html
@@ -20,7 +20,7 @@
     <script src="d3fc.plugin.js"></script>
 
     <link rel='stylesheet' href="index.css">
-    <link rel='stylesheet' href="material.css" is="custom-style">   
+    <link rel='stylesheet' href="material.css" is="custom-style">
 
 </head>
 

diff --git a/packages/perspective/package.json b/packages/perspective/package.json
@@ -62,7 +62,6 @@
         "papaparse": "^4.3.6",
         "text-encoding-utf-8": "^1.0.2",
         "tslib": "^1.9.3",
-        "websocket-heartbeat-js": "^1.0.7",
         "ws": "^6.1.2"
     },
     "devDependencies": {

diff --git a/packages/perspective/src/js/API/host.js b/packages/perspective/src/js/API/host.js
@@ -128,22 +128,29 @@ export class Host {
      */
     process_subscribe(msg, obj) {
         try {
-            obj[msg.method](ev => {
+            const _callback = ev => {
                 let result = {
                     id: msg.id,
                     data: ev
                 };
+                try {
+                    // post transferable data for arrow
+                    if (msg.args && msg.args[0]) {
+                        if (msg.method === "on_update" && msg.args[0]["mode"] === "row") {
+                            this.post(result, [ev]);
+                            return;
+                        }
+                    }
 
-                // post transferable data for arrow
-                if (msg.args && msg.args[0]) {
-                    if (msg.method === "on_update" && msg.args[0]["mode"] === "row") {
-                        this.post(result, [ev]);
-                        return;
+                    this.post(result);
+                } catch (e) {
+                    console.error("Removing callback after failed on_update() (presumably due to closed connection)");
+                    if (msg.method === "on_update") {
+                        obj["remove_update"](_callback);
                     }
                 }
-
-                this.post(result);
-            }, ...msg.args); // make sure we are passing arguments into the callback
+            };
+            obj[msg.method](_callback, ...msg.args); // make sure we are passing arguments into the callback
         } catch (error) {
             this.process_error(msg, error);
             return;

diff --git a/packages/perspective/src/js/perspective.js b/packages/perspective/src/js/perspective.js
@@ -789,9 +789,25 @@ export default function(Module) {
         });
     };
 
+    function filterInPlace(a, condition) {
+        let i = 0,
+            j = 0;
+
+        while (i < a.length) {
+            const val = a[i];
+            if (condition(val, i, a)) a[j++] = val;
+            i++;
+        }
+
+        a.length = j;
+        return a;
+    }
+
     view.prototype.remove_update = function(callback) {
         _clear_process(this.pool);
-        this.callbacks = this.callbacks.filter(x => x.callback !== callback);
+        const total = this.callbacks.length;
+        filterInPlace(this.callbacks, x => x.orig_callback !== callback);
+        console.assert(total > this.callbacks.length, `"callback" does not match a registered updater`);
     };
 
     /**

diff --git a/packages/perspective/src/js/perspective.node.js b/packages/perspective/src/js/perspective.node.js
@@ -134,6 +134,7 @@ class WebSocketHost extends module.exports.Host {
         this._server = http.createServer(create_http_server(assets, host_psp));
 
         this.REQS = {};
+        this.REQ_ID_MAP = new Map();
 
         // Serve Worker API through WebSockets
         this._wss = new WebSocket.Server({noServer: true, perMessageDeflate: true});
@@ -151,7 +152,10 @@ class WebSocketHost extends module.exports.Host {
                     return;
                 }
                 msg = JSON.parse(msg);
-                this.REQS[msg.id] = ws;
+                const compound_id = msg.id + ws.id;
+                this.REQ_ID_MAP.set(compound_id, msg.id);
+                msg.id = compound_id;
+                this.REQS[msg.id] = {ws, msg};
                 try {
                     // Send all messages to the handler defined in Perspective.Host
                     this.process(msg, ws.id);
@@ -209,14 +213,24 @@ class WebSocketHost extends module.exports.Host {
      * @param {*} transferable a transferable object to be sent to the client
      */
     post(msg, transferable) {
+        const req = this.REQS[msg.id];
+        const id = msg.id;
+        if (req.ws.readyState > 1) {
+            delete this.REQS[id];
+            throw new Error("Connection closed");
+        }
+        msg.id = this.REQ_ID_MAP.get(id);
         if (transferable) {
             msg.is_transferable = true;
-            this.REQS[msg.id].send(JSON.stringify(msg));
-            this.REQS[msg.id].send(transferable[0]);
+            req.ws.send(JSON.stringify(msg));
+            req.ws.send(transferable[0]);
         } else {
-            this.REQS[msg.id].send(JSON.stringify(msg));
+            req.ws.send(JSON.stringify(msg));
+        }
+        if (!req.msg.subscribe) {
+            this.REQ_ID_MAP.delete(id);
+            delete this.REQS[id];
         }
-        delete this.REQS[msg.id];
     }
 
     /**