feat(wslink): upgrade to wslink 2

.gitignore

@@ -13,3 +13,4 @@ build
 *.db
 tests/tests_output/
 package-lock.json
+VTK.txt

opengeodeweb_viewer_schemas.json

@@ -3266,7 +3266,7 @@
           "type": "object",
           "properties": {
             "blob": {
-              "type": "string"
+              "description": "Image file content, sent as binary"
             }
           },
           "required": [

opengeodeweb_viewer_typed_schemas.d.ts

@@ -896,7 +896,10 @@ export interface ViewerTakeScreenshotParams {
 export type OutputExtension = "png" | "jpg";
 
 export interface ViewerTakeScreenshotResponse {
-    blob: string;
+    /**
+     * Image file content, sent as binary
+     */
+    blob: any;
 }
 
 export interface ViewerSliceParams {

requirements.in

@@ -1,3 +1,3 @@
-wslink==1.12.4
+wslink==2.6.0
 websocket_client==1.8.0
 vtk==9.6.2

requirements.txt

@@ -28,7 +28,9 @@ kiwisolver==1.*
     # via matplotlib
 matplotlib==3.*
     # via vtk
-multidict==6.*
+msgpack==1.*
+    # via wslink
+multidict==7.*
     # via
     #   aiohttp
     #   yarl
@@ -58,9 +60,8 @@ vtk==9.6.2
     # via -r requirements.in
 websocket-client==1.8.0
     # via -r requirements.in
-wslink==1.12.4
+wslink==2.6.0
     # via -r requirements.in
 yarl==1.*
     # via aiohttp
 
-opengeodeweb-microservice==1.*,>=1.3.2rc1

src/opengeodeweb_viewer/app.py

@@ -155,7 +155,10 @@ def run_server(server_protocol: type[ServerProtocol] = _Server) -> None:
 
     logger.info("args=%s", args)
     server_protocol.configure(args)
-    # Python 3.14 no longer creates a default event loop, which wslink 1.x expects
+    # Wslink 2 writes its ready message without a trailing newline,
+    # which line-based launchers waiting for it never receive
+    os.environ.setdefault("WSLINK_READY_MSG", "wslink: Starting factory\n")
+    # Python 3.14 no longer creates a default event loop, which wslink expects
     asyncio.set_event_loop(asyncio.new_event_loop())
     server.start_webserver(options=args, protocol=server_protocol)
 

src/opengeodeweb_viewer/rpc/viewer/schemas/take_screenshot.json

@@ -28,7 +28,7 @@
     "type": "object",
     "properties": {
       "blob": {
-        "type": "string"
+        "description": "Image file content, sent as binary"
       }
     },
     "required": [

src/opengeodeweb_viewer/rpc/viewer/schemas/take_screenshot.py

@@ -1,5 +1,6 @@
 from pathlib import Path
 from opengeodeweb_microservice.schemas import Route, load_schema
+from typing import Any
 from dataclasses_json import DataClassJsonMixin
 from opengeodeweb_microservice.schemas import print_dataclass
 from enum import Enum
@@ -26,7 +27,8 @@ class TakeScreenshotResponse(DataClassJsonMixin):
     def __post_init__(self) -> None:
         print_dataclass(self)
 
-    blob: str
+    blob: Any
+    """Image file content, sent as binary"""
 
 
 take_screenshot_route = Route(

src/opengeodeweb_viewer/rpc/viewer/viewer_protocols.py

@@ -160,7 +160,7 @@ def take_screenshot(self, params: schemas.TakeScreenshot) -> schemas.TakeScreens
 
         file_content = file_path.read_bytes()
 
-        return schemas.TakeScreenshotResponse(blob=self.addAttachment(file_content))
+        return schemas.TakeScreenshotResponse(blob=file_content)
 
     @typed_rpc(viewer_prefix, schemas.update_data_route)
     def update_data(self, params: schemas.UpdateData) -> schemas.UpdateDataResponse:

src/opengeodeweb_viewer/typed_rpc.py

@@ -2,6 +2,7 @@
 import functools
 import os
 from collections.abc import Callable
+from dataclasses import fields
 from typing import Any, Concatenate, Protocol, cast
 
 # Third party imports
@@ -30,6 +31,17 @@ def _drop_none(value: object) -> object:
     return value
 
 
+def _to_payload(result: DataClassJsonMixin) -> dict[str, Any]:
+    payload = cast("dict[str, Any]", _drop_none(result.to_dict(encode_json=True)))
+    # dataclasses_json encodes bytes as a list of ints: keep them as bytes,
+    # wslink (msgpack) sends them as binary
+    for field in fields(result):  # type: ignore[arg-type]
+        value = getattr(result, field.name)
+        if isinstance(value, bytes):
+            payload[field.name] = value
+    return payload
+
+
 def _validate_responses() -> bool:
     return os.environ.get("PYTHON_ENV", "prod").strip().lower() in ("dev", "test")
 
@@ -65,7 +77,7 @@ def rpc(
                     f"expected {route.response.__name__}"
                 )
                 raise TypeError(msg)
-            payload = cast("dict[str, Any]", _drop_none(result.to_dict(encode_json=True)))
+            payload = _to_payload(result)
             if _validate_responses():
                 validate_response(payload)
             if do_stream:

tests/conftest.py

@@ -1,5 +1,4 @@
 import contextlib
-import json
 import logging
 import os
 import shutil
@@ -9,12 +8,14 @@
 from pathlib import Path
 from typing import Any, ClassVar
 
+import msgpack  # type: ignore[import-untyped]
 import pytest
 from opengeodeweb_microservice.database.connection import get_session, init_database
 from opengeodeweb_microservice.database.data import Data
 from vtkmodules.vtkImagingCore import vtkImageDifference
 from vtkmodules.vtkIOImage import vtkImageReader2, vtkJPEGReader, vtkPNGReader
 from websocket import WebSocketTimeoutException, create_connection
+from wslink.chunking import UnChunker, generate_chunks  # type: ignore[import-untyped]
 from xprocess import ProcessStarter, XProcess  # type: ignore[import-untyped]
 
 from opengeodeweb_viewer.config import TestConfig
@@ -25,28 +26,37 @@
 
 type RpcTestParams = list[dict[str, Any] | int] | None
 
+MAX_MESSAGE_SIZE = 1024**3
+INITIAL_IMAGE_TIMEOUT = 2.0
+
 
 class ServerMonitor:
     def __init__(self, log: str, port: str = "1234") -> None:
         self.log = log
         self.ws = create_connection(f"ws://localhost:{port}/ws")
+        self.unchunker = UnChunker()
+        # The un-chunker refuses messages above 512 bytes by default
+        self.unchunker.set_max_message_size(MAX_MESSAGE_SIZE)
         self.images_dir_path = Path(__file__).parent.resolve() / "data" / "images"
         self.test_output_dir = Path(__file__).parent.resolve() / "tests_output"
         self.test_output_dir.mkdir(exist_ok=True)
         self._init_ws()
         self._drain_initial_messages()
 
+    def send(self, message: dict[str, Any]) -> None:
+        # wslink 2 messages are msgpack-encoded and sent as (here, a single) binary chunk
+        for chunk in generate_chunks(msgpack.packb(message), 0):
+            self.ws.send_binary(chunk)
+
     def call(self, rpc: str, params: RpcTestParams = None) -> None:
         if params is None:
             params = [{}]
-        self.ws.send(
-            json.dumps(
-                {
-                    "id": "rpc:" + rpc,
-                    "method": rpc,
-                    "args": params,
-                }
-            )
+        self.send(
+            {
+                "id": "rpc:" + rpc,
+                "method": rpc,
+                "args": params,
+            }
         )
 
     def print_log(self) -> None:
@@ -59,17 +69,22 @@ def print_log(self) -> None:
                 output += line
         logger.info("%s", output)
 
-    def get_response(self) -> bytes | dict[str, object] | str:
-        response = self.ws.recv()
-        if isinstance(response, bytes):
-            return response
-        try:
-            parsed = json.loads(response)
-            if isinstance(parsed, dict):
-                return parsed
-            return str(parsed)
-        except json.JSONDecodeError:
-            return str(response)
+    def get_response(self) -> dict[str, Any]:
+        while True:
+            # process_chunk returns the msgpack-decoded message once all its chunks are received
+            response: dict[str, Any] | None = self.unchunker.process_chunk(self.ws.recv())
+            if response is not None:
+                return response
+
+    def get_rpc_response(self, rpc: str) -> dict[str, Any]:
+        while True:
+            response = self.get_response()
+            if response.get("id") == "rpc:" + rpc:
+                return response
+
+    def get_result(self, rpc: str) -> dict[str, Any]:
+        result: dict[str, Any] = self.get_rpc_response(rpc)["result"]
+        return result
 
     @staticmethod
     def _reader_for_file(path: Path) -> vtkImageReader2:
@@ -100,43 +115,43 @@ def images_diff(self, first_image_path: Path, second_image_path: Path) -> float:
     def compare_image(self, filename: str) -> bool:
         self.call(VtkViewerView.viewer_prefix + VtkViewerView.viewer_schemas_dict["render"]["rpc"])
         while True:
-            image = self.ws.recv()
-            if isinstance(image, bytes):
-                response = self.ws.recv()
-                logger.info("response=%s", response)
-                result = json.loads(response)["result"]
-                if result["stale"]:
-                    continue
-                test_file_path = self.test_output_dir / f"test.{result['format']}"
-                test_file_path.write_bytes(image)
-                path_image = self.images_dir_path / filename
-                return self.images_diff(test_file_path, path_image) == 0.0
-            logger.info("response = %s", image)
-        return False
+            response = self.get_response()
+            result = response.get("result")
+            if not isinstance(result, dict) or "image" not in result:
+                logger.info("response = %s", response)
+                continue
+            logger.info("response=%s", {**result, "image": f"<{len(result['image'])} bytes>"})
+            if result["stale"]:
+                continue
+            test_file_path = self.test_output_dir / f"test.{result['format']}"
+            test_file_path.write_bytes(result["image"])
+            path_image = self.images_dir_path / filename
+            return self.images_diff(test_file_path, path_image) == 0.0
 
     def _init_ws(self) -> None:
-        self.ws.send(
-            json.dumps(
-                {
-                    "id": "system:hello",
-                    "method": "wslink.hello",
-                    "args": [{"secret": "wslink-secret"}],
-                }
-            )
+        self.send(
+            {
+                "id": "system:hello",
+                "method": "wslink.hello",
+                "args": [{"secret": "wslink-secret"}],
+            }
         )
         self.call("viewport.image.push.observer.add", [-1])
 
-    def _drain_initial_messages(self, max_messages: int = 5, timeout: float = 10.0) -> None:
+    def _drain_initial_messages(self, timeout: float = 10.0) -> None:
         self.ws.settimeout(timeout)
-        for i in range(max_messages):
-            try:
-                self.ws.recv()
-            except WebSocketTimeoutException:
-                logger.warning(
-                    "Timeout on message %s, but continuing to try remaining messages...",
-                    i,
-                )
-                continue
+        self.get_result("viewport.image.push.observer.add")
+        # A fresh viewer then publishes its initial image, a reused one may not
+        self.ws.settimeout(INITIAL_IMAGE_TIMEOUT)
+        try:
+            while True:
+                result = self.get_response().get("result")
+                if isinstance(result, dict) and "image" in result:
+                    return
+        except WebSocketTimeoutException:
+            logger.info("No initial image received")
+        finally:
+            self.ws.settimeout(timeout)
 
 
 class FixtureHelper:

tests/mesh/cells/test_mesh_cells_protocols.py

@@ -110,9 +110,7 @@ def register_grid_3d(server: ServerMonitor, dataset_factory: Callable[..., str])
 
 def call_slice(server: ServerMonitor, ids: list[str], slices: list[dict[str, int]]) -> object:
     server.call(slice_rpc, [{"ids": ids, "slices": slices}])
-    response = server.get_response()
-    assert isinstance(response, dict), f"Unexpected response: {response!r}"
-    return response["result"]
+    return server.get_result(slice_rpc)
 
 
 def test_slice(server: ServerMonitor, dataset_factory: Callable[..., str]) -> None:
@@ -175,7 +173,7 @@ def test_slice_on_non_grid_removed(
         VtkMeshView.mesh_prefix + VtkMeshView.mesh_schemas_dict["register"]["rpc"],
         [{"id": other_mesh_id, "name": "hat.vtp"}],
     )
-    server.get_response()
+    server.get_result(VtkMeshView.mesh_prefix + VtkMeshView.mesh_schemas_dict["register"]["rpc"])
 
     result = call_slice(server, [other_mesh_id], [])
     assert result == {"max_indices": [0, 0, 0]}
@@ -190,7 +188,7 @@ def test_slice_on_non_grid_ignored(
         VtkMeshView.mesh_prefix + VtkMeshView.mesh_schemas_dict["register"]["rpc"],
         [{"id": other_mesh_id, "name": "hat.vtp"}],
     )
-    server.get_response()
+    server.get_result(VtkMeshView.mesh_prefix + VtkMeshView.mesh_schemas_dict["register"]["rpc"])
 
     result = call_slice(server, [other_mesh_id, grid_3d_id], [{"axis": 2, "index": 3}])
     assert result == {"max_indices": [10, 8, 6]}

tests/model/surfaces/test_model_surfaces_protocols.py

@@ -102,9 +102,11 @@ def test_surfaces_polygons_random_collection_id(
             }
         ],
     )
-    response = server.get_response()
-    assert isinstance(response, dict)
-    colors = [result["color"] for result in response["result"]["colors"]]
+    result = server.get_result(
+        VtkModelSurfacesView.model_surfaces_prefix
+        + VtkModelSurfacesView.model_surfaces_schemas_dict["color"]["rpc"]
+    )
+    colors = [surface["color"] for surface in result["colors"]]
     assert len(colors) == 11
     assert all(color == colors[0] for color in colors)
 

tests/model/test_model_protocols.py

@@ -72,11 +72,7 @@ def test_get_blocks_bounds(server: ServerMonitor, dataset_factory: Callable[...,
     rpc = VtkModelView.model_prefix + "get_blocks_bounds"
     server.call(rpc, [{"id": "12345678901234567890123456789012", "block_ids": [2]}])
 
-    response = server.get_response()
-    while isinstance(response, bytes) or response.get("id") != f"rpc:{rpc}":
-        response = server.get_response()
-
-    assert response.get("result") == {"bounds": [4.9, 4.9, 3.1, 3.1, 0.0, 0.0]}
+    assert server.get_result(rpc) == {"bounds": [4.9, 4.9, 3.1, 3.1, 0.0, 0.0]}
 
 
 def test_model_cube_side_view(server: ServerMonitor, dataset_factory: Callable[..., str]) -> None:

tests/test_utils_protocols.py

@@ -32,7 +32,9 @@ def test_reset_project_after_import(
     server.call(
         VtkUtilsView.utils_prefix + VtkUtilsView.utils_schemas_dict["import_project"]["rpc"]
     )
-    server.get_response()
+    server.get_result(
+        VtkUtilsView.utils_prefix + VtkUtilsView.utils_schemas_dict["import_project"]["rpc"]
+    )
 
     server.call(
         VtkViewerView.viewer_prefix