Add Fragmentation support for WebRTC

src/main/java/net/rptools/clientserver/hessian/client/MethodClientDecorator.java

@@ -20,17 +20,19 @@
 import net.rptools.clientserver.simple.DisconnectHandler;
 import net.rptools.clientserver.simple.MessageHandler;
 import net.rptools.clientserver.simple.client.ClientConnection;
+import org.apache.log4j.Logger;
 
 /** @author drice */
 public class MethodClientDecorator implements MethodClientConnection {
+  private static final Logger log = Logger.getLogger(MethodClientDecorator.class);
   private ClientConnection connection;
 
   public MethodClientDecorator(ClientConnection connection) {
     this.connection = connection;
   }
 
   public void callMethod(String method, Object... parameters) {
-
+    log.debug(connection.getId() + ": will call " + method);
     byte[] message = HessianUtils.methodToBytesGZ(method, parameters);
     sendMessage(message);
   }

src/main/java/net/rptools/clientserver/hessian/server/MethodServerDecorator.java

@@ -22,30 +22,37 @@
 import net.rptools.clientserver.simple.MessageHandler;
 import net.rptools.clientserver.simple.server.ServerConnection;
 import net.rptools.clientserver.simple.server.ServerObserver;
+import org.apache.log4j.Logger;
 
 /** @author drice */
 public class MethodServerDecorator implements MethodServerConnection {
+  private static final Logger log = Logger.getLogger(MethodServerDecorator.class);
+
   private ServerConnection connection;
 
   public MethodServerDecorator(ServerConnection connection) {
     this.connection = connection;
   }
 
   public void broadcastCallMethod(String method, Object... parameters) {
+    log.debug("will broadcast " + method);
     broadcastMessage(HessianUtils.methodToBytesGZ(method, parameters));
   }
 
   public void broadcastCallMethod(String[] exclude, String method, Object... parameters) {
+    log.debug("will broadcast " + method);
     byte[] data = HessianUtils.methodToBytesGZ(method, parameters);
     broadcastMessage(exclude, data);
   }
 
   public void callMethod(String id, String method, Object... parameters) {
+    log.debug("will call " + method + " to " + id);
     byte[] data = HessianUtils.methodToBytesGZ(method, parameters);
     sendMessage(id, null, data);
   }
 
   public void callMethod(String id, Object channel, String method, Object... parameters) {
+    log.debug("will call " + method + " to " + id + "(" + channel + ")");
     byte[] data = HessianUtils.methodToBytesGZ(method, parameters);
     sendMessage(id, channel, data);
   }

src/main/java/net/rptools/clientserver/simple/AbstractConnection.java

@@ -17,6 +17,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -185,5 +186,37 @@ public final byte[] readMessage(InputStream in) throws IOException {
     return ret;
   }
 
+  private ByteBuffer messageBuffer = null;
+
+  public final byte[] readMessage(ByteBuffer part) {
+    if (messageBuffer == null) {
+      int length = part.getInt();
+      notifyListeners(Direction.Inbound, State.Start, length, 0);
+
+      if (part.remaining() == length) {
+        var ret = new byte[length];
+        part.get(ret);
+        notifyListeners(Direction.Inbound, State.Complete, length, length);
+        return ret;
+      }
+
+      messageBuffer = ByteBuffer.allocate(length);
+    }
+
+    messageBuffer.put(part);
+    notifyListeners(
+        Direction.Inbound, State.Progress, messageBuffer.capacity(), messageBuffer.position());
+
+    if (messageBuffer.capacity() == messageBuffer.position()) {
+      notifyListeners(
+          Direction.Inbound, State.Complete, messageBuffer.capacity(), messageBuffer.capacity());
+      var ret = messageBuffer.array();
+      messageBuffer = null;
+      return ret;
+    }
+
+    return null;
+  }
+
   public abstract String getError();
 }

src/main/java/net/rptools/clientserver/simple/client/WebRTCClientConnection.java

@@ -17,9 +17,7 @@
 import com.google.gson.Gson;
 import dev.onvoid.webrtc.*;
 import dev.onvoid.webrtc.media.MediaStream;
-import java.io.ByteArrayInputStream;
 import java.io.IOException;
-import java.io.InputStream;
 import java.net.URI;
 import java.nio.ByteBuffer;
 import net.rptools.clientserver.simple.AbstractConnection;
@@ -349,7 +347,7 @@ public void onIceCandidate(RTCIceCandidate candidate) {
 
   @Override
   public void onIceCandidateError(RTCPeerConnectionIceErrorEvent event) {
-    log.info(
+    log.debug(
         prefix()
             + "PeerConnection.onIceCandidateError: code:"
             + event.getErrorCode()
@@ -424,13 +422,20 @@ public void onBufferedAmountChange(long previousAmount) {
   // dataChannel
   @Override
   public void onStateChange() {
-    log.info(prefix() + "localDataChannel onStateChange " + localDataChannel.getState());
-    if (localDataChannel.getState() == RTCDataChannelState.OPEN) {
-      // connection established we don't need the signaling server anymore
-      // for now disabled. We may get additional ice candidates.
-      if (!isServerSide() && signalingClient.isOpen()) signalingClient.close();
-
-      sendThread.start();
+    var state = localDataChannel.getState();
+    log.info(prefix() + "localDataChannel onStateChange " + state);
+    switch (state) {
+      case OPEN -> {
+        // connection established we don't need the signaling server anymore
+        // for now disabled. We may get additional ice candidates.
+        if (!isServerSide() && signalingClient.isOpen()) signalingClient.close();
+
+        sendThread.start();
+      }
+      case CLOSED -> {
+        close();
+        fireDisconnectAsync();
+      }
     }
   }
 
@@ -439,26 +444,14 @@ public void onStateChange() {
   public void onMessage(RTCDataChannelBuffer channelBuffer) {
     log.debug(
         prefix() + "localDataChannel onMessage: got " + channelBuffer.data.capacity() + " bytes");
-    var buffer = channelBuffer.data;
 
     if (Thread.currentThread().getContextClassLoader() == null) {
       ClassLoader cl = ClassLoader.getSystemClassLoader();
       Thread.currentThread().setContextClassLoader(cl);
     }
 
-    int len = buffer.capacity();
-    var byteArray = new byte[len];
-    buffer.get(byteArray);
-
-    InputStream byteStream = new ByteArrayInputStream(byteArray);
-    try {
-      while (byteStream.available() > 0) {
-        var message = readMessage(byteStream);
-        dispatchMessage(id, message);
-      }
-    } catch (IOException e) {
-      e.printStackTrace();
-    }
+    var message = readMessage(channelBuffer.data);
+    if (message != null) dispatchMessage(id, message);
   }
 
   private void fireDisconnectAsync() {
@@ -518,17 +511,29 @@ public void run() {
               continue;
             }
 
-            var length = message.length;
-            var buffer = ByteBuffer.allocate(length + 4);
+            ByteBuffer buffer = ByteBuffer.allocate(message.length + Integer.BYTES);
+            buffer.putInt(message.length).put(message).rewind();
+
+            int chunkSize = 16 * 1024;
 
-            buffer.put((byte) (length >> 24));
-            buffer.put((byte) (length >> 16));
-            buffer.put((byte) (length >> 8));
-            buffer.put((byte) (length));
-            buffer.put(message);
+            while (buffer.remaining() > 0) {
+              var amountToSend = buffer.remaining() <= chunkSize ? buffer.remaining() : chunkSize;
+              ByteBuffer part = buffer;
 
-            localDataChannel.send(new RTCDataChannelBuffer(buffer, true));
-            log.debug(prefix() + " sent " + (length + 4) + " bytes");
+              if (amountToSend != buffer.capacity()) {
+                // we need to allocation a new ByteBuffer because send calls ByteBuffer.array()
+                // which would return
+                // the whole byte[] and not only the slice. But the lib doesn't use
+                // ByteBuffer.arrayOffset().
+                var slice = buffer.slice(buffer.position(), amountToSend);
+                part = ByteBuffer.allocate(amountToSend);
+                part.put(slice);
+              }
+
+              buffer.position(buffer.position() + amountToSend);
+              localDataChannel.send(new RTCDataChannelBuffer(part, true));
+              log.debug(prefix() + " sent " + part.capacity() + " bytes");
+            }
           }
           synchronized (this) {
             if (!stopRequested) {

src/main/java/net/rptools/lib/BackupManager.java

@@ -77,7 +77,11 @@ public void backup(File file) throws IOException {
   private List<File> getFiles() {
 
     List<File> fileList = new LinkedList<File>(Arrays.asList(backupDir.listFiles()));
-    fileList.sort((o1, o2) -> o1.lastModified() < o2.lastModified() ? -1 : 1);
+    fileList.sort(
+        (o1, o2) -> {
+          if (o1.lastModified() == o2.lastModified()) return 0;
+          return o1.lastModified() < o2.lastModified() ? -1 : 1;
+        });
 
     return fileList;
   }

src/main/java/net/rptools/maptool/client/ClientMethodHandler.java

@@ -63,6 +63,7 @@
 import net.rptools.maptool.transfer.AssetChunk;
 import net.rptools.maptool.transfer.AssetConsumer;
 import net.rptools.maptool.transfer.AssetHeader;
+import org.apache.log4j.Logger;
 
 /**
  * This class is used by the clients to receive server commands sent through {@link
@@ -71,11 +72,15 @@
  * @author drice
  */
 public class ClientMethodHandler extends AbstractMethodHandler {
+  private static final Logger log = Logger.getLogger(ClientMethodHandler.class);
+
   public ClientMethodHandler() {}
 
   public void handleMethod(final String id, final String method, final Object... parameters) {
     final ClientCommand.COMMAND cmd = Enum.valueOf(ClientCommand.COMMAND.class, method);
 
+    log.debug("from " + id + " got " + method);
+
     // These commands are safe to do in the background, any events that cause model updates need
     // to be on the EDT (See next section)
     switch (cmd) {

src/main/java/net/rptools/maptool/server/ServerMethodHandler.java

@@ -48,6 +48,7 @@
 import net.rptools.maptool.model.drawing.Pen;
 import net.rptools.maptool.model.framework.dropinlibrary.TransferableAddOnLibrary;
 import net.rptools.maptool.transfer.AssetProducer;
+import org.apache.log4j.Logger;
 
 /**
  * This class is used by the server host to receive client commands sent through {@link
@@ -61,6 +62,7 @@
 public class ServerMethodHandler extends AbstractMethodHandler implements ServerCommand {
   private final MapToolServer server;
   private final Object MUTEX = new Object();
+  private static final Logger log = Logger.getLogger(ServerMethodHandler.class);
 
   public ServerMethodHandler(MapToolServer server) {
     this.server = server;
@@ -70,6 +72,8 @@ public ServerMethodHandler(MapToolServer server) {
   public void handleMethod(String id, String method, Object... parameters) {
     ServerCommand.COMMAND cmd = Enum.valueOf(ServerCommand.COMMAND.class, method);
 
+    log.debug("from " + id + " got " + method);
+
     try {
       RPCContext context = new RPCContext(id, method, parameters);
       RPCContext.setCurrent(context);