JS ProxySession

.devcontainer/devcontainer.json

@@ -0,0 +1,21 @@
+// For format details, see https://aka.ms/devcontainer.json. For config options, see the
+// README at: https://github.com/devcontainers/templates/tree/main/src/universal
+{
+	"name": "Ubuntu: pnpm, rust, cmake, for JS dev",
+	"image": "mcr.microsoft.com/devcontainers/base:ubuntu",
+	"customizations": {
+		"vscode": {
+			"extensions": [
+				"rust-lang.rust-analyzer",
+				"ms-playwright.playwright"
+			]
+		}
+	},
+
+	"features": {
+		"ghcr.io/devcontainers/features/rust:1": {},
+		"ghcr.io/devcontainers-extra/features/pnpm:2": {}
+	},
+
+	"postCreateCommand": "./.devcontainer/postcreate.sh"
+}

.devcontainer/postcreate.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+
+set -e
+
+sudo apt-get update
+
+# Install cmake
+test -f /usr/share/doc/kitware-archive-keyring/copyright ||
+wget -O - https://apt.kitware.com/keys/kitware-archive-latest.asc 2>/dev/null | gpg --dearmor - | sudo tee /usr/share/keyrings/kitware-archive-keyring.gpg >/dev/null
+
+echo 'deb [signed-by=/usr/share/keyrings/kitware-archive-keyring.gpg] https://apt.kitware.com/ubuntu/ jammy main' | sudo tee /etc/apt/sources.list.d/kitware.list >/dev/null
+sudo apt-get update
+sudo apt-get install -y kitware-archive-keyring
+sudo apt-get install -y cmake
+
+# Repo setup
+pnpm install

.vscode/extensions.json

@@ -1,3 +1,3 @@
 {
-    "recommendations": ["rust-lang.rust", "ms-vscode.cpptools-extension-pack"]
+    "recommendations": ["rust-lang.rust-analyzer", "ms-vscode.cpptools-extension-pack"]
 }

rust/perspective-client/src/rust/session.rs

@@ -90,7 +90,7 @@ pub struct ProxySession {
 }
 
 impl ProxySession {
-    pub async fn new(
+    pub fn new(
         client: Client,
         send_response: impl Fn(&[u8]) -> Result<(), Box<dyn std::error::Error + Send + Sync>>
             + Send

rust/perspective-js/src/rust/client.rs

@@ -14,7 +14,7 @@ use js_sys::{Function, Uint8Array};
 use macro_rules_attribute::apply;
 #[cfg(doc)]
 use perspective_client::SystemInfo;
-use perspective_client::{TableData, TableInitOptions};
+use perspective_client::{Session, TableData, TableInitOptions};
 use wasm_bindgen::prelude::*;
 
 pub use crate::table::*;
@@ -32,6 +32,51 @@ extern "C" {
     pub type JsTableInitOptions;
 }
 
+#[wasm_bindgen]
+#[derive(Clone)]
+pub struct ProxySession(perspective_client::ProxySession);
+
+#[wasm_bindgen]
+impl ProxySession {
+    #[wasm_bindgen(constructor)]
+    pub fn new(client: &Client, on_response: &Function) -> Self {
+        let poll_loop = LocalPollLoop::new({
+            let on_response = on_response.clone();
+            move |msg: Vec<u8>| {
+                let msg = Uint8Array::from(&msg[..]);
+                on_response.call1(&JsValue::UNDEFINED, &JsValue::from(msg))?;
+                Ok(JsValue::null())
+            }
+        });
+        // NB: This swallows any errors raised by the inner callback
+        let on_response = Box::new(move |msg: &[u8]| {
+            wasm_bindgen_futures::spawn_local(poll_loop.poll(msg.to_vec()));
+            Ok(())
+        });
+        Self(perspective_client::ProxySession::new(
+            client.client.clone(),
+            on_response,
+        ))
+    }
+
+    #[wasm_bindgen]
+    pub async fn handle_request(&self, data: &[u8]) -> ApiResult<()> {
+        use perspective_client::Session;
+        self.0.handle_request(data).await?;
+        Ok(())
+    }
+
+    #[wasm_bindgen]
+    pub async fn poll(&self) -> ApiResult<()> {
+        self.0.poll().await?;
+        Ok(())
+    }
+
+    pub async fn close(self) {
+        self.0.close().await;
+    }
+}
+
 #[apply(inherit_docs)]
 #[inherit_doc = "client.md"]
 #[wasm_bindgen]
@@ -61,6 +106,11 @@ impl Client {
         Client { close, client }
     }
 
+    #[wasm_bindgen]
+    pub fn new_proxy_session(&self, on_response: &Function) -> ProxySession {
+        ProxySession::new(self, on_response)
+    }
+
     #[wasm_bindgen]
     pub async fn init(&self) -> ApiResult<()> {
         self.client.clone().init().await?;

rust/perspective-js/test/js/proxy_session.spec.js

@@ -0,0 +1,99 @@
+// ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
+// ┃ ██████ ██████ ██████       █      █      █      █      █ █▄  ▀███ █       ┃
+// ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█  ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄  ▀█ █ ▀▀▀▀▀ ┃
+// ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄   █ ▄▄▄▄▄ ┃
+// ┃ █      ██████ █  ▀█▄       █ ██████      █      ███▌▐███ ███████▄ █       ┃
+// ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫
+// ┃ Copyright (c) 2017, the Perspective Authors.                              ┃
+// ┃ ╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ ┃
+// ┃ This file is part of the Perspective library, distributed under the terms ┃
+// ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃
+// ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
+
+import { test, expect } from "@finos/perspective-test";
+import { make_client, make_server } from "@finos/perspective";
+
+test("Proxy session tunnels requests through client", async () => {
+    // test.setTimeout(2000);
+    const { client, server } = connectClientToServer();
+    const { proxyClient } = connectProxyClient(client);
+
+    // Verify that main client + proxy client observe the same server
+    const clientTables1 = await client.get_hosted_table_names();
+    const proxyTables1 = await proxyClient.get_hosted_table_names();
+    expect(proxyTables1).toStrictEqual([]);
+    expect(proxyTables1).toStrictEqual(clientTables1);
+    const name = "abc-" + Math.random();
+    const _table = await client.table({ abc: [123] }, { name });
+    const clientTables2 = await client.get_hosted_table_names();
+    const proxyTables2 = await proxyClient.get_hosted_table_names();
+    expect(proxyTables2).toStrictEqual([name]);
+    expect(proxyTables2).toStrictEqual(clientTables2);
+});
+
+test("Proxy session tunnels on_update callbacks through client", async () => {
+    // test.setTimeout(2000);
+    const { client } = connectClientToServer();
+    const { proxyClient } = connectProxyClient(client);
+    const name = "abc-" + Math.random();
+    const clientTable = await client.table({ abc: [123] }, { name });
+
+    // Add an on_update callback to the proxy client's view of the table
+    const proxyTable = await proxyClient.open_table(name);
+    const proxyView = await proxyTable.view();
+    let resolveUpdate;
+    const onUpdateResp = new Promise((r) => (resolveUpdate = r));
+    await proxyView.on_update(
+        (x) => {
+            resolveUpdate(x);
+        },
+        { mode: "row" }
+    );
+
+    // Enact table update through client's table handle, and assert that proxy
+    // client's on_update callback is called
+    await clientTable.update({ abc: [999] });
+    const expectUpdate = expect.poll(
+        async () => {
+            const data = await onUpdateResp;
+            // TODO: construct table out of onUpdateResp, assert contents?
+            console.log("onUpdateResp, with data", data);
+            return data;
+        },
+        {
+            message: "Ensure proxy view updates with table",
+            timeout: 10000,
+        }
+    );
+
+    expect(await proxyView.to_columns()).toStrictEqual({ abc: [123, 999] });
+
+    await expectUpdate.toHaveProperty("delta");
+    await expectUpdate.toHaveProperty("port_id");
+});
+
+function connectClientToServer() {
+    const server = make_server();
+    const session = server.make_session((msg) => {
+        client.handle_response(msg);
+    });
+    const client = make_client((msg) => {
+        session.handle_request(msg);
+    });
+    return {
+        client,
+        server,
+    };
+}
+
+function connectProxyClient(client) {
+    const sess = client.new_proxy_session((res) => {
+        proxyClient.handle_response(res);
+    });
+    const proxyClient = make_client((msg) => {
+        sess.handle_request(msg);
+    });
+    return {
+        proxyClient,
+    };
+}

rust/perspective-python/src/client/client_sync.rs

@@ -47,10 +47,10 @@ impl ProxySession {
             }
         };
 
-        Ok(ProxySession(
-            perspective_client::ProxySession::new(client.borrow(py).0.client.clone(), callback)
-                .py_block_on(py),
-        ))
+        Ok(ProxySession(perspective_client::ProxySession::new(
+            client.borrow(py).0.client.clone(),
+            callback,
+        )))
     }
 
     pub fn handle_request(&self, py: Python<'_>, data: Vec<u8>) -> PyResult<()> {