update: yjs callbacks handle Uint8Arrays

Author: 4e6

app/ydoc-channel/src/YjsChannel.ts

@@ -173,7 +173,7 @@ export class YjsChannel<T = unknown> extends ObservableV2<WebSocketEventHandlers
   /**
    * Notifies all subscribed handlers with the received message.
    */
-  private notifyHandlers(message: T): void {
+  protected notifyHandlers(message: any): void {
     // Create a MessageEvent-like object for WebSocket compatibility
     const messageEvent = { data: message } as MessageEvent
 
@@ -229,4 +229,9 @@ export class YjsDataChannel<T = unknown> extends YjsChannel<T> {
       return new YjsDataChannel(doc, channelName, callbacks)
     })
   }
+
+  override notifyHandlers(message: any): void {
+    const arr = message as Uint8Array
+    super.notifyHandlers(arr.buffer)
+  }
 }

engine/language-server/src/main/scala/org/enso/languageserver/http/server/BinaryYdocServer.scala

@@ -48,15 +48,16 @@ object BinaryYdocServer {
     ): Unit = {
       logger.info(s"BinaryServerCallbacks.onMessage ${message.getClass}")
       message match {
-        case bytes: ByteBuffer =>
-          logger.info(s"Received binary message ${bytes.getClass}")
-          decoder.decode(bytes) match {
+        case bytes: Array[Byte] =>
+          //val bytes = value.as(classOf[Array[Byte]])
+          logger.info(s"Received binary message")
+          decoder.decode(ByteBuffer.wrap(bytes)) match {
             case Right(message) =>
               incomingMessageHandler ! message
             case Left(error) =>
               logger.error("Failed to decode binary message", error)
           }
-          messageCallbacks.foreach(cb => cb(bytes))
+          messageCallbacks.foreach(cb => cb(ByteBuffer.wrap(bytes)))
         case _ =>
           logger.error(
             s"Received unsupported message: ${message.getClass}"

lib/java/ydoc-polyfill/src/test/java/org/enso/ydoc/api/CallbacksTest.java

@@ -104,4 +104,31 @@ class YjsChannel {
 
     Assert.assertEquals("World!", res.get());
   }
+
+  @Test
+  public void onConnectSubscribeBuffer() throws Exception {
+    var res = new AtomicReference<>();
+    var code =
+        """
+        class YjsChannel {
+          subscribe(messageHandler) {
+            var arr = new Uint8Array([0, 128, 255]);
+            messageHandler(arr.buffer);
+          }
+        }
+
+        var channel = new YjsChannel();
+        callbacks.onConnect(channel);
+        """;
+
+    var callbacks =
+        new TestCallbacks((channel) -> channel.subscribe((message) -> res.set(message)));
+    context.getBindings("js").putMember("callbacks", callbacks);
+
+    CompletableFuture.runAsync(() -> context.eval("js", code), executor).get();
+    var value = context.asValue(res.get());
+    var arr = value.as(byte[].class);
+
+    Assert.assertArrayEquals(new byte[] {0, -128, -1}, arr);
+  }
 }

lib/java/ydoc-server/src/main/java/org/enso/ydoc/server/Ydoc.java

@@ -146,14 +146,10 @@ public Ydoc build() {
 
       if (jsonChannelCallbacks == null) {
         jsonChannelCallbacks = NoOpMessageCallbacks.INSTANCE;
-      } else {
-        jsonChannelCallbacks = new YjsCallbacksSynchronized(jsonChannelCallbacks, executor);
       }
 
       if (binaryChannelCallbacks == null) {
         binaryChannelCallbacks = NoOpMessageCallbacks.INSTANCE;
-      } else {
-        binaryChannelCallbacks = new YjsCallbacksSynchronized(binaryChannelCallbacks, executor);
       }
 
       return new Ydoc(
@@ -175,6 +171,14 @@ public Context.Builder getContextBuilder() {
     return contextBuilder;
   }
 
+  public MessageCallbacks getJsonChannelCallbacksSynchronized() {
+    return new YjsCallbacksSynchronized(jsonChannelCallbacks, executor);
+  }
+
+  public MessageCallbacks getBinaryChannelCallbacksSynchronized(Context context) {
+    return new YjsBinaryChannelCallbacksSynchronized(binaryChannelCallbacks, executor, context);
+  }
+
   public void start() throws ExecutionException, InterruptedException, IOException {
     var ydoc = Main.class.getResource(YDOC_PATH);
     if (ydoc == null) {
@@ -195,8 +199,10 @@ public void start() throws ExecutionException, InterruptedException, IOException
                   var bindings = ctx.getBindings("js");
                   bindings.putMember("YDOC_HOST", hostname);
                   bindings.putMember("YDOC_PORT", port);
-                  bindings.putMember("YDOC_JSON_CHANNEL_CALLBACKS", jsonChannelCallbacks);
-                  bindings.putMember("YDOC_BINARY_CHANNEL_CALLBACKS", binaryChannelCallbacks);
+                  bindings.putMember(
+                      "YDOC_JSON_CHANNEL_CALLBACKS", getJsonChannelCallbacksSynchronized());
+                  bindings.putMember(
+                      "YDOC_BINARY_CHANNEL_CALLBACKS", getBinaryChannelCallbacksSynchronized(ctx));
                   bindings.putMember("YDOC_LS_DEBUG", "false");
 
                   ctx.eval(ydocJs);

lib/java/ydoc-server/src/main/java/org/enso/ydoc/server/YjsBinaryChannelCallbacksSynchronized.java

@@ -0,0 +1,27 @@
+package org.enso.ydoc.server;
+
+import java.util.concurrent.ExecutorService;
+import org.enso.ydoc.api.MessageCallbacks;
+import org.enso.ydoc.api.YjsChannel;
+import org.graalvm.polyglot.Context;
+
+public class YjsBinaryChannelCallbacksSynchronized implements MessageCallbacks {
+
+  private final MessageCallbacks callbacks;
+  private final ExecutorService executor;
+  private final Context context;
+
+  public YjsBinaryChannelCallbacksSynchronized(
+      MessageCallbacks callbacks, ExecutorService executor, Context context) {
+    this.callbacks = callbacks;
+    this.executor = executor;
+    this.context = context;
+  }
+
+  @Override
+  public void onConnect(YjsChannel channel) {
+    var synchronizedChannel =
+        new YjsBinaryChannelSynchronized(channel, this.executor, this.context);
+    this.callbacks.onConnect(synchronizedChannel);
+  }
+}

lib/java/ydoc-server/src/main/java/org/enso/ydoc/server/YjsBinaryChannelSynchronized.java

@@ -0,0 +1,45 @@
+package org.enso.ydoc.server;
+
+import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
+import org.enso.ydoc.api.YjsChannel;
+import org.graalvm.polyglot.Context;
+
+public class YjsBinaryChannelSynchronized implements YjsChannel {
+
+  private final YjsChannel channel;
+  private final ExecutorService executor;
+  private final Context context;
+
+  public YjsBinaryChannelSynchronized(
+      YjsChannel channel, ExecutorService executor, Context context) {
+    this.channel = channel;
+    this.executor = executor;
+    this.context = context;
+  }
+
+  @Override
+  public void send(Object message) {
+    executor.execute(() -> channel.send(message));
+  }
+
+  @Override
+  public void subscribe(Consumer<Object> messageHandler) {
+    executor.execute(
+        () ->
+            channel.subscribe(
+                (message) -> {
+                  System.out.println("DBG YjsBinaryChannel message " + message);
+                  var value = context.asValue(message);
+                  System.out.println("DBG YjsBinaryChannel value " + value);
+                  byte[] bytes = null;
+                  try {
+                    bytes = value.as(byte[].class);
+                  } catch (Throwable e) {
+                    System.out.println("DBG YjsBinaryChannel ERR " + e);
+                  }
+                  System.out.println("DBG YjsBinaryChannel bytes " + bytes);
+                  messageHandler.accept(bytes);
+                }));
+  }
+}