Split client-to-server messages when websockets are used (#11648)

* First part of a fragmented message prefixed by total message length plus '|'
* Atmosphere websocket buffer size set to 65536
* Fragment size is 65535 characters (Jetty requires less than buffer size)

Change-Id: I8bf872bbb03b86386070fdc99c14ea805dd2ef3a

1 files changed, 55 insertions, 4 deletions

diff --git a/client/src/com/vaadin/client/communication/AtmospherePushConnection.java b/client/src/com/vaadin/client/communication/AtmospherePushConnection.java
index f6b704ed9b..e42f91ea67 100644
--- a/client/src/com/vaadin/client/communication/AtmospherePushConnection.java
+++ b/client/src/com/vaadin/client/communication/AtmospherePushConnection.java
@@ -62,6 +62,45 @@ public class AtmospherePushConnection implements PushConnection {
         DISCONNECTED;
     }
 
+    /**
+     * Represents a message that should be sent as multiple fragments.
+     */
+    protected static class FragmentedMessage {
+
+        // Jetty requires length less than buffer size
+        private int FRAGMENT_LENGTH = ApplicationConstants.WEBSOCKET_BUFFER_SIZE - 1;
+
+        private String message;
+        private int index = 0;
+
+        public FragmentedMessage(String message) {
+            this.message = message;
+        }
+
+        public boolean hasNextFragment() {
+            return index < message.length();
+        }
+
+        public String getNextFragment() {
+            String result;
+            if (index == 0) {
+                String header = "" + message.length()
+                        + ApplicationConstants.WEBSOCKET_MESSAGE_DELIMITER;
+                int fragmentLen = FRAGMENT_LENGTH - header.length();
+                result = header + getFragment(0, fragmentLen);
+                index += fragmentLen;
+            } else {
+                result = getFragment(index, index + FRAGMENT_LENGTH);
+                index += FRAGMENT_LENGTH;
+            }
+            return result;
+        }
+
+        private String getFragment(int begin, int end) {
+            return message.substring(begin, Math.min(message.length(), end));
+        }
+    }
+
     private ApplicationConnection connection;
 
     private JavaScriptObject socket;
@@ -74,6 +113,8 @@ public class AtmospherePushConnection implements PushConnection {
 
     private String uri;
 
+    private String transport;
+
     /**
      * Keeps track of the disconnect confirmation command for cases where
      * pending messages should be pushed before actually disconnecting.
@@ -151,7 +192,15 @@ public class AtmospherePushConnection implements PushConnection {
         case CONNECTED:
             assert isActive();
             VConsole.log("Sending push message: " + message);
-            doPush(socket, message);
+
+            if (transport.equals("websocket")) {
+                FragmentedMessage fragmented = new FragmentedMessage(message);
+                while (fragmented.hasNextFragment()) {
+                    doPush(socket, fragmented.getNextFragment());
+                }
+            } else {
+                doPush(socket, message);
+            }
             break;
         case DISCONNECT_PENDING:
         case DISCONNECTED:
@@ -167,10 +216,12 @@ public class AtmospherePushConnection implements PushConnection {
     }
 
     protected void onOpen(AtmosphereResponse response) {
-        VConsole.log("Push connection established using "
-                + response.getTransport());
+        transport = response.getTransport();
+
+        VConsole.log("Push connection established using " + transport);
+
         for (String message : messageQueue) {
-            doPush(socket, message);
+            push(message);
         }
         messageQueue.clear();