liveliness, keyexpr, serialization, querier tests added. Bug in querier fixed: parameters were not passed (#199)

* liveliness test added, vscode library resolving fixed

* tests for keyexpr, liveliness, serialization added

* fixed bug: querier parameters were not passed, test for querier added

* run_test function which prints test name added

* update run_test calls to use function references instead of names

* delay for flaky test added

* avoid vscode errors in analyzing zenoh-ts import

Author: milyin

zenoh-plugin-remote-api/src/handle_control_message.rs

@@ -471,6 +471,7 @@ pub(crate) async fn handle_control_message(
         ControlMsg::QuerierGet {
             get_id,
             querier_id,
+            parameters,
             encoding,
             payload,
             attachment,
@@ -501,6 +502,7 @@ pub(crate) async fn handle_control_message(
                 add_if_some!(encoding, get_builder);
                 add_if_some!(payload, get_builder);
                 add_if_some!(attachment, get_builder);
+                add_if_some!(parameters, get_builder);
 
                 let ws_tx = state_map.websocket_tx.clone();
                 let finish_msg = RemoteAPIMsg::Control(ControlMsg::GetFinished { id: get_id });

zenoh-plugin-remote-api/src/interface/mod.rs

@@ -352,6 +352,8 @@ pub enum ControlMsg {
         querier_id: Uuid,
         get_id: Uuid,
         #[ts(type = "string | undefined")]
+        parameters: Option<String>,
+        #[ts(type = "string | undefined")]
         encoding: Option<String>,
         #[ts(type = "string | undefined")]
         payload: Option<B64String>,

zenoh-ts/examples/deno/tsconfig.json

@@ -12,6 +12,10 @@
     "resolveJsonModule": true,
     "isolatedModules": true,
     "noEmit": true,
+    "paths": {
+      "@eclipse-zenoh/zenoh-ts": ["../../src/index.ts"],
+      "@eclipse-zenoh/zenoh-ts/*": ["../../src/*"]
+    },
 
     /* Linting */
     "strict": true,

zenoh-ts/scripts/start.sh

@@ -32,7 +32,7 @@ else
   echo
   echo "Available options:"
   echo
-  echo "yarn start test [test-name] | ALL"
+  echo "yarn start test test-name|ALL [DAEMON]"
   echo "yarn start example deno [example-name]"
   echo "yarn start example browser"
   echo

zenoh-ts/src/remote_api/interface/ControlMsg.ts

@@ -4,4 +4,4 @@ import type { HandlerChannel } from "./HandlerChannel.js";
 import type { LivelinessMsg } from "./LivelinessMsg.js";
 import type { OwnedKeyExprWrapper } from "./OwnedKeyExprWrapper.js";
 
-export type ControlMsg = "OpenSession" | "CloseSession" | { "Session": string } | { "NewTimestamp": string } | "SessionInfo" | { "Get": { key_expr: OwnedKeyExprWrapper, parameters: string | null, handler: HandlerChannel, id: string, consolidation: number | undefined, timeout: number | undefined, congestion_control: number | undefined, priority: number | undefined, target: number | undefined, express: boolean | undefined, encoding: string | undefined, payload: string | undefined, attachment: string | undefined, } } | { "GetFinished": { id: string, } } | { "Put": { key_expr: OwnedKeyExprWrapper, payload: B64String, encoding: string | undefined, congestion_control: number | undefined, priority: number | undefined, express: boolean | undefined, attachment: string | undefined, timestamp: string | undefined, } } | { "Delete": { key_expr: OwnedKeyExprWrapper, congestion_control: number | undefined, priority: number | undefined, express: boolean | undefined, attachment: string | undefined, timestamp: string | undefined, } } | { "DeclareSubscriber": { key_expr: OwnedKeyExprWrapper, handler: HandlerChannel, id: string, } } | { "Subscriber": string } | { "UndeclareSubscriber": string } | { "DeclarePublisher": { key_expr: OwnedKeyExprWrapper, encoding: string | undefined, congestion_control: number | undefined, priority: number | undefined, reliability: number | undefined, express: boolean | undefined, id: string, } } | { "UndeclarePublisher": string } | { "DeclareQueryable": { key_expr: OwnedKeyExprWrapper, id: string, complete: boolean, handler: HandlerChannel, } } | { "UndeclareQueryable": string } | { "DeclareQuerier": { id: string, key_expr: OwnedKeyExprWrapper, target: number | undefined, timeout: number | undefined, accept_replies: number | undefined, allowed_destination: number | undefined, congestion_control: number | undefined, priority: number | undefined, consolidation: number | undefined, express: boolean | undefined, } } | { "UndeclareQuerier": string } | { "QuerierGet": { querier_id: string, get_id: string, encoding: string | undefined, payload: string | undefined, attachment: string | undefined, handler: HandlerChannel, } } | { "Liveliness": LivelinessMsg };
+export type ControlMsg = "OpenSession" | "CloseSession" | { "Session": string } | { "NewTimestamp": string } | "SessionInfo" | { "Get": { key_expr: OwnedKeyExprWrapper, parameters: string | null, handler: HandlerChannel, id: string, consolidation: number | undefined, timeout: number | undefined, congestion_control: number | undefined, priority: number | undefined, target: number | undefined, express: boolean | undefined, encoding: string | undefined, payload: string | undefined, attachment: string | undefined, } } | { "GetFinished": { id: string, } } | { "Put": { key_expr: OwnedKeyExprWrapper, payload: B64String, encoding: string | undefined, congestion_control: number | undefined, priority: number | undefined, express: boolean | undefined, attachment: string | undefined, timestamp: string | undefined, } } | { "Delete": { key_expr: OwnedKeyExprWrapper, congestion_control: number | undefined, priority: number | undefined, express: boolean | undefined, attachment: string | undefined, timestamp: string | undefined, } } | { "DeclareSubscriber": { key_expr: OwnedKeyExprWrapper, handler: HandlerChannel, id: string, } } | { "Subscriber": string } | { "UndeclareSubscriber": string } | { "DeclarePublisher": { key_expr: OwnedKeyExprWrapper, encoding: string | undefined, congestion_control: number | undefined, priority: number | undefined, reliability: number | undefined, express: boolean | undefined, id: string, } } | { "UndeclarePublisher": string } | { "DeclareQueryable": { key_expr: OwnedKeyExprWrapper, id: string, complete: boolean, handler: HandlerChannel, } } | { "UndeclareQueryable": string } | { "DeclareQuerier": { id: string, key_expr: OwnedKeyExprWrapper, target: number | undefined, timeout: number | undefined, accept_replies: number | undefined, allowed_destination: number | undefined, congestion_control: number | undefined, priority: number | undefined, consolidation: number | undefined, express: boolean | undefined, } } | { "UndeclareQuerier": string } | { "QuerierGet": { querier_id: string, get_id: string, parameters: string | undefined, encoding: string | undefined, payload: string | undefined, attachment: string | undefined, handler: HandlerChannel, } } | { "Liveliness": LivelinessMsg };

zenoh-ts/src/remote_api/querier.ts

@@ -42,33 +42,34 @@ export class RemoteQuerier {
     }
 
     get(
-        _handler_type: HandlerChannel,
-        _encoding?: string,
-        _parameters?: string,
-        _attachment?: Array<number>,
-        _payload?: Array<number>,
+        handler_type: HandlerChannel,
+        encoding?: string,
+        parameters?: string,
+        attachment?: Array<number>,
+        payload?: Array<number>,
     ): SimpleChannel<ReplyWS> {
         let get_id = uuidv4();
         let channel: SimpleChannel<ReplyWS> = new SimpleChannel<ReplyWS>();
         this.session_ref.get_receiver.set(get_id, channel);
 
-        let payload = undefined;
-        if (_payload != undefined) {
-            payload = b64_str_from_bytes(new Uint8Array(_payload))
+        let payload_str = undefined;
+        if (payload != undefined) {
+            payload_str = b64_str_from_bytes(new Uint8Array(payload))
         }
-        let attachment = undefined;
-        if (_attachment != undefined) {
-            attachment = b64_str_from_bytes(new Uint8Array(_attachment))
+        let attachment_str = undefined;
+        if (attachment != undefined) {
+            attachment_str = b64_str_from_bytes(new Uint8Array(attachment))
         }
 
         let control_msg: ControlMsg = {
             QuerierGet: {
                 querier_id: this.querier_id as string,
                 get_id: get_id,
-                encoding: _encoding,
-                payload: payload,
-                attachment: attachment,
-                handler: _handler_type
+                parameters: parameters,
+                encoding: encoding,
+                payload: payload_str,
+                attachment: attachment_str,
+                handler: handler_type
             }
         };
 

zenoh-ts/tests/src/common/assertions.ts

@@ -18,6 +18,45 @@ export function assert(condition: boolean, message: string): void {
   }
 }
 
+function deepEqual(actual: any, expected: any): boolean {
+    if (actual === expected) return true;
+    
+    if (actual instanceof Map && expected instanceof Map) {
+        if (actual.size !== expected.size) return false;
+        for (const [key, value] of actual) {
+            if (!expected.has(key)) return false;
+            if (!deepEqual(value, expected.get(key))) return false;
+        }
+        return true;
+    }
+
+    if (Array.isArray(actual) && Array.isArray(expected)) {
+        if (actual.length !== expected.length) return false;
+        return actual.every((val, idx) => deepEqual(val, expected[idx]));
+    }
+
+    if (actual instanceof Float32Array && expected instanceof Float32Array ||
+        actual instanceof Float64Array && expected instanceof Float64Array) {
+        if (actual.length !== expected.length) return false;
+        return actual.every((val, idx) => Math.abs(val - expected[idx]) < 0.0001);
+    }
+
+    if (typeof actual === 'object' && actual !== null &&
+        typeof expected === 'object' && expected !== null) {
+        return Object.keys(actual).length === Object.keys(expected).length &&
+               Object.keys(actual).every(key => deepEqual(actual[key], expected[key]));
+    }
+
+    return false;
+}
+
 export function assert_eq<T>(actual: T, expected: T, message: string): void {
-  assert(actual === expected, `${message}: expected ${expected}, but got ${actual}`);
+    if (!deepEqual(actual, expected)) {
+        throw new Error(`${message}: expected ${expected}, but got ${actual}`);
+    }
+}
+
+export async function run_test(fn: () => Promise<void>): Promise<void> {
+    console.warn(`Test function: ${fn.name}`);
+    await fn();
 }

zenoh-ts/tests/src/z_api_pub_sub.ts

@@ -13,9 +13,9 @@
 //
 
 import { Config, Session, Subscriber, Sample } from "@eclipse-zenoh/zenoh-ts";
-import { assert, assert_eq } from "./common/assertions.ts";
+import { assert, assert_eq, run_test } from "./common/assertions.ts";
 
-export async function putSubTest() {
+async function putSubTest() {
   // Open two sessions
   const session1 = await Session.open(new Config("ws/127.0.0.1:10000"));
   const session2 = await Session.open(new Config("ws/127.0.0.1:10000"));
@@ -57,4 +57,4 @@ export async function putSubTest() {
   await session2.close();
 }
 
-putSubTest();
+await run_test(putSubTest);

zenoh-ts/tests/src/z_api_queryable_get.ts

@@ -12,10 +12,10 @@
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
 
-import { Config, Session, Query, Reply, KeyExpr, Receiver, ZBytes, Selector, ReplyError, Parameters, Sample } from "@eclipse-zenoh/zenoh-ts";
-import { assert_eq } from "./common/assertions.ts";
+import { Config, Session, Query, Reply, KeyExpr, Receiver, ZBytes, Selector, ReplyError, Parameters, Sample, QueryTarget } from "@eclipse-zenoh/zenoh-ts";
+import { assert_eq, run_test } from "./common/assertions.ts";
 
-async function queryableGetCallbackTest() {
+async function queryableSessionGetCallbackTest() {
     const ke_queryable = new KeyExpr("zenoh/test/*");
     const ke_get = new KeyExpr("zenoh/test/1");
     const queries: Query[] = [];
@@ -84,7 +84,7 @@ async function queryableGetCallbackTest() {
     await session2.close();
 }
 
-async function queryableGetChannelTest() {
+async function queryableSessionGetChannelTest() {
     const ke = new KeyExpr("zenoh/test/*");
     const selector = new KeyExpr("zenoh/test/1");
 
@@ -147,5 +147,160 @@ async function queryableGetChannelTest() {
     await session2.close();
 }
 
-await queryableGetCallbackTest();
-await queryableGetChannelTest();
+async function queriableQuerierGetChannelTest() {
+    const ke = new KeyExpr("zenoh/test/*");
+    const selector = new KeyExpr("zenoh/test/1");
+
+    const session1 = await Session.open(new Config("ws/127.0.0.1:10000"));
+    const session2 = await Session.open(new Config("ws/127.0.0.1:10000"));
+
+    const queryable = session1.declare_queryable(ke, { complete: true });
+    
+    // sleep for 1 second to ensure the queryable is ready
+    await new Promise((resolve) => setTimeout(resolve, 1000));
+    
+    const querier = session2.declare_querier(selector, {
+        target: QueryTarget.BestMatching
+    });
+
+    // First query with ok parameters
+    const receiver1 = await querier.get(new Parameters("p=ok"), { payload: "1" });
+    let query = await queryable.receive();
+    assert_eq(query instanceof Query, true, "Expected query to be an instance of Query");
+    if (query instanceof Query) {
+        assert_eq(query.key_expr().toString(), selector.toString(), "Key mismatch");
+        assert_eq(query.parameters().toString(), "p=ok", "Parameters mismatch");
+        assert_eq(query.payload()?.to_string(), "1", "Payload mismatch");
+        query.reply(query.key_expr(), "1");
+    }
+
+    // sleep to ensure the reply is processed
+    await new Promise((resolve) => setTimeout(resolve, 100));
+
+    if (receiver1) {
+        let reply = await receiver1.receive();
+        assert_eq(reply instanceof Reply, true, "Expected reply to be an instance of Reply");
+        if (reply instanceof Reply) {
+            const result = reply.result();
+            assert_eq(result instanceof ReplyError, false, "Reply should be OK");
+            if (!(result instanceof ReplyError)) {
+                assert_eq(result.payload().to_string(), "1", "Reply payload mismatch");
+            }
+        }
+    }
+
+    // Second query using the same querier with error parameters
+    const receiver2 = await querier.get(new Parameters("p=err"), { payload: "2" });
+    query = await queryable.receive();
+    assert_eq(query instanceof Query, true, "Expected query to be an instance of Query");
+    if (query instanceof Query) {
+        assert_eq(query.key_expr().toString(), selector.toString(), "Key mismatch");
+        assert_eq(query.parameters().toString(), "p=err", "Parameters mismatch");
+        assert_eq(query.payload()?.to_string(), "2", "Payload mismatch");
+        query.reply_err("err");
+    }
+
+    // sleep to ensure the reply is processed
+    await new Promise((resolve) => setTimeout(resolve, 100));
+
+    if (receiver2) {
+        let reply = await receiver2.receive();
+        assert_eq(reply instanceof Reply, true, "Expected reply to be an instance of Reply");
+        if (reply instanceof Reply) {
+            const result = reply.result();
+            assert_eq(result instanceof ReplyError, true, "Reply should be an error");
+            if (result instanceof ReplyError) {
+                assert_eq(result.payload().to_string(), "err", "Error payload mismatch");
+            }
+        }
+    }
+
+    querier.undeclare();
+    await queryable.undeclare();
+    await session1.close();
+    await session2.close();
+}
+
+async function queriableQuerierGetCallbackTest() {
+    const ke = new KeyExpr("zenoh/test/*");
+    const selector = new KeyExpr("zenoh/test/1");
+    const queries: Query[] = [];
+    const replies: Reply[] = [];
+
+    const session1 = await Session.open(new Config("ws/127.0.0.1:10000"));
+    const session2 = await Session.open(new Config("ws/127.0.0.1:10000"));
+
+    const queryable = session1.declare_queryable(ke, { complete: true });
+    
+    // sleep for 1 second to ensure the queryable is ready
+    await new Promise((resolve) => setTimeout(resolve, 1000));
+    
+    const querier = session2.declare_querier(selector, {
+        target: QueryTarget.BestMatching
+    });
+
+    const handler = async (reply: Reply) => { replies.push(reply); };
+
+    // First query with ok parameters
+    querier.get(new Parameters("p=ok"), {
+        payload: "1",
+        handler: handler
+    });
+
+    // sleep to ensure the request is handled
+    await new Promise((resolve) => setTimeout(resolve, 100));
+
+    let query = await queryable.receive();
+    assert_eq(query instanceof Query, true, "Expected query to be an instance of Query");
+    if (query instanceof Query) {
+        queries.push(query);
+        assert_eq(query.key_expr().toString(), selector.toString(), "Key mismatch");
+        assert_eq(query.parameters().toString(), "p=ok", "Parameters mismatch");
+        assert_eq(query.payload()?.to_string(), "1", "Payload mismatch");
+        query.reply(query.key_expr(), "1");
+    }
+
+    // Second query using the same querier with error parameters
+    querier.get(new Parameters("p=err"), {
+        payload: "2",
+        handler: handler
+    });
+
+    // sleep to ensure the request is handled
+    await new Promise((resolve) => setTimeout(resolve, 100));
+
+    query = await queryable.receive();
+    assert_eq(query instanceof Query, true, "Expected query to be an instance of Query");
+    if (query instanceof Query) {
+        queries.push(query);
+        assert_eq(query.key_expr().toString(), selector.toString(), "Key mismatch");
+        assert_eq(query.parameters().toString(), "p=err", "Parameters mismatch");
+        assert_eq(query.payload()?.to_string(), "2", "Payload mismatch");
+        query.reply_err("err");
+    }
+
+    // sleep to ensure replies are processed
+    await new Promise((resolve) => setTimeout(resolve, 100));
+
+    assert_eq(queries.length, 2, "Queries received");
+    assert_eq(replies.length, 2, "Replies received");
+    
+    assert_eq(replies[0].result() instanceof Sample, true, "Reply 0 should be Sample");
+    if (!(replies[0].result() instanceof ReplyError)) {
+        assert_eq(replies[0].result().payload().to_string(), "1", "Reply payload mismatch");
+    }
+    assert_eq(replies[1].result() instanceof ReplyError, true, "Reply 1 should be ReplyError");
+    if (replies[1].result() instanceof ReplyError) {
+        assert_eq(replies[1].result().payload().to_string(), "err", "Error payload mismatch");
+    }
+
+    querier.undeclare();
+    await queryable.undeclare();
+    await session1.close();
+    await session2.close();
+}
+
+await run_test(queryableSessionGetCallbackTest);
+await run_test(queryableSessionGetChannelTest);
+await run_test(queriableQuerierGetChannelTest);
+await run_test(queriableQuerierGetCallbackTest);