Server push (#111)

* Asynchronous bidirectional communication
* Use Atmosphere as a backend
* Use websockets if available, fallback to HTTP streaming
* Push mode (disabled, manual, automatic)
  * Configurable via servlet parameter pushMode
  * Disabled: The default; regular AJAX communication
  * Manual: Need explicit UI.push() call
  * Automatic: push all UIs in session when lock released
* UI.push()
  * Push pending state and RPC to client asynchronously
  * Must hold session lock when invoking

Change-Id: Idb5978ac81f7ff1e66665df4e3f96e29e4c419d4

3 files changed, 195 insertions, 11 deletions

diff --git a/client/src/com/vaadin/client/ApplicationConfiguration.java b/client/src/com/vaadin/client/ApplicationConfiguration.java
index 2291f21361..a6cc3cf531 100644
--- a/client/src/com/vaadin/client/ApplicationConfiguration.java
+++ b/client/src/com/vaadin/client/ApplicationConfiguration.java
@@ -36,6 +36,7 @@ import com.vaadin.client.metadata.NoDataException;
 import com.vaadin.client.metadata.TypeData;
 import com.vaadin.client.ui.UnknownComponentConnector;
 import com.vaadin.shared.ApplicationConstants;
+import com.vaadin.shared.communication.PushMode;
 import com.vaadin.shared.ui.ui.UIConstants;
 
 public class ApplicationConfiguration implements EntryPoint {
@@ -201,6 +202,7 @@ public class ApplicationConfiguration implements EntryPoint {
     private ErrorMessage authorizationError;
     private ErrorMessage sessionExpiredError;
     private int heartbeatInterval;
+    private PushMode pushMode;
 
     private HashMap<Integer, String> unknownComponents;
 
@@ -304,6 +306,10 @@ public class ApplicationConfiguration implements EntryPoint {
         return heartbeatInterval;
     }
 
+    public PushMode getPushMode() {
+        return pushMode;
+    }
+
     public JavaScriptObject getVersionInfoJSObject() {
         return getJsoConfiguration(id).getVersionInfoJSObject();
     }
@@ -357,6 +363,14 @@ public class ApplicationConfiguration implements EntryPoint {
         heartbeatInterval = jsoConfiguration
                 .getConfigInteger("heartbeatInterval");
 
+        String pushMode = jsoConfiguration.getConfigString("pushMode");
+        if (pushMode != null) {
+            this.pushMode = Enum
+                    .valueOf(PushMode.class, pushMode.toUpperCase());
+        } else {
+            this.pushMode = PushMode.DISABLED;
+        }
+
         communicationError = jsoConfiguration.getConfigError("comErrMsg");
         authorizationError = jsoConfiguration.getConfigError("authErrMsg");
         sessionExpiredError = jsoConfiguration.getConfigError("sessExpMsg");
@@ -365,7 +379,6 @@ public class ApplicationConfiguration implements EntryPoint {
         if (jsoConfiguration.getConfigBoolean("initPending") == Boolean.FALSE) {
             setBrowserDetailsSent();
         }
-
     }
 
     /**
diff --git a/client/src/com/vaadin/client/ApplicationConnection.java b/client/src/com/vaadin/client/ApplicationConnection.java
index d59abc892a..0341a9d5c4 100644
--- a/client/src/com/vaadin/client/ApplicationConnection.java
+++ b/client/src/com/vaadin/client/ApplicationConnection.java
@@ -66,6 +66,7 @@ import com.vaadin.client.communication.HasJavaScriptConnectorHelper;
 import com.vaadin.client.communication.JavaScriptMethodInvocation;
 import com.vaadin.client.communication.JsonDecoder;
 import com.vaadin.client.communication.JsonEncoder;
+import com.vaadin.client.communication.PushConnection;
 import com.vaadin.client.communication.RpcManager;
 import com.vaadin.client.communication.StateChangeEvent;
 import com.vaadin.client.extensions.AbstractExtensionConnector;
@@ -88,6 +89,7 @@ import com.vaadin.shared.ApplicationConstants;
 import com.vaadin.shared.Version;
 import com.vaadin.shared.communication.LegacyChangeVariablesInvocation;
 import com.vaadin.shared.communication.MethodInvocation;
+import com.vaadin.shared.communication.PushMode;
 import com.vaadin.shared.communication.SharedState;
 import com.vaadin.shared.ui.ui.UIConstants;
 
@@ -222,6 +224,8 @@ public class ApplicationConnection {
 
     private final RpcManager rpcManager;
 
+    private PushConnection push;
+
     /**
      * If responseHandlingLocks contains any objects, response handling is
      * suspended until the collection is empty or a timeout has occurred.
@@ -439,6 +443,8 @@ public class ApplicationConnection {
 
         scheduleHeartbeat();
 
+        initializePush();
+
         Window.addWindowClosingHandler(new ClosingHandler() {
             @Override
             public void onWindowClosing(ClosingEvent event) {
@@ -831,13 +837,16 @@ public class ApplicationConnection {
                         response.getText().length() - 1);
                 handleJSONText(jsonText, statusCode);
             }
-
         };
-        try {
-            doAjaxRequest(uri, payload, requestCallback);
-        } catch (RequestException e) {
-            VConsole.error(e);
-            endRequest();
+        if (push != null) {
+            push.push(payload);
+        } else {
+            try {
+                doAjaxRequest(uri, payload, requestCallback);
+            } catch (RequestException e) {
+                VConsole.error(e);
+                endRequest();
+            }
         }
     }
 
@@ -848,7 +857,7 @@ public class ApplicationConnection {
      * @param jsonText
      * @param statusCode
      */
-    private void handleJSONText(String jsonText, int statusCode) {
+    public void handleJSONText(String jsonText, int statusCode) {
         final Date start = new Date();
         final ValueMap json;
         try {
@@ -952,7 +961,7 @@ public class ApplicationConnection {
      * servicing the session so far. These values are always one request behind,
      * since they cannot be measured before the request is finished.
      */
-    private ValueMap serverTimingInfo;
+    public ValueMap serverTimingInfo;
 
     static final int MAX_CSS_WAITS = 100;
 
@@ -1462,7 +1471,11 @@ public class ApplicationConnection {
                         + jsonText.length() + " characters of JSON");
                 VConsole.log("Referenced paintables: " + connectorMap.size());
 
-                endRequest();
+                if (meta == null || !meta.containsKey("async")) {
+                    // End the request if the received message was a response,
+                    // not sent asynchronously
+                    endRequest();
+                }
 
                 if (Profiler.isEnabled()) {
                     Scheduler.get().scheduleDeferred(new ScheduledCommand() {
@@ -1473,7 +1486,6 @@ public class ApplicationConnection {
                         }
                     });
                 }
-
             }
 
             /**
@@ -3314,4 +3326,28 @@ public class ApplicationConnection {
         return Util.getConnectorForElement(this, getUIConnector().getWidget(),
                 focusedElement);
     }
+
+    private void initializePush() {
+        if (getConfiguration().getPushMode() != PushMode.DISABLED) {
+            push = GWT.create(PushConnection.class);
+            push.init(this);
+
+            final String pushUri = addGetParameters(
+                    translateVaadinUri(ApplicationConstants.APP_PROTOCOL_PREFIX
+                            + ApplicationConstants.PUSH_PATH + '/'),
+                    UIConstants.UI_ID_PARAMETER + "="
+                            + getConfiguration().getUIId());
+
+            Scheduler.get().scheduleDeferred(new Command() {
+                @Override
+                public void execute() {
+                    push.connect(pushUri);
+                }
+            });
+        }
+    }
+
+    public void handlePushMessage(String message) {
+        handleJSONText(message, 200);
+    }
 }
diff --git a/client/src/com/vaadin/client/communication/PushConnection.java b/client/src/com/vaadin/client/communication/PushConnection.java
new file mode 100644
index 0000000000..f87ddf430a
--- /dev/null
+++ b/client/src/com/vaadin/client/communication/PushConnection.java
@@ -0,0 +1,135 @@
+/*
+ * Copyright 2000-2013 Vaadin Ltd.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package com.vaadin.client.communication;
+
+import java.util.ArrayList;
+
+import com.google.gwt.core.client.JavaScriptObject;
+import com.vaadin.client.ApplicationConnection;
+import com.vaadin.client.VConsole;
+
+/**
+ * Represents the client-side endpoint of a bidirectional ("push") communication
+ * channel. Can be used to send UIDL request messages to the server and to
+ * receive UIDL messages from the server (either asynchronously or as a response
+ * to a UIDL request.) Delegates the UIDL handling to the
+ * {@link ApplicationConnection}.
+ * 
+ * @author Vaadin Ltd
+ * @since 7.1
+ */
+public class PushConnection {
+
+    private ApplicationConnection connection;
+
+    private JavaScriptObject socket;
+
+    private ArrayList<String> messageQueue = new ArrayList<String>();
+
+    private boolean connected = false;
+
+    private JavaScriptObject config = createConfig();
+
+    public PushConnection() {
+    }
+
+    /**
+     * Two-phase construction to allow using GWT.create()
+     * 
+     * @param ac
+     *            The ApplicationConnection
+     */
+    public void init(ApplicationConnection ac) {
+        this.connection = ac;
+    }
+
+    public void connect(String uri) {
+        VConsole.log("Establishing Atmosphere connection");
+        socket = doConnect(uri, getConfig());
+    }
+
+    public void push(String message) {
+        if (!connected) {
+            VConsole.log("Queuing Atmosphere message: " + message);
+            messageQueue.add(message);
+        } else {
+            VConsole.log("Pushing Atmosphere message: " + message);
+            doPush(socket, message);
+        }
+    }
+
+    protected JavaScriptObject getConfig() {
+        return config;
+    }
+
+    protected void onOpen() {
+        VConsole.log("Atmosphere connection established");
+        connected = true;
+        for (String message : messageQueue) {
+            push(message);
+        }
+        messageQueue.clear();
+    }
+
+    protected void onMessage(String message) {
+        if (message.startsWith("for(;;);")) {
+            VConsole.log("Received Atmosphere message: " + message);
+            // "for(;;);[{json}]" -> "{json}"
+            message = message.substring(9, message.length() - 1);
+            connection.handlePushMessage(message);
+        }
+    }
+
+    protected void onError() {
+        VConsole.error("Atmosphere connection failed!");
+    }
+
+    private static native JavaScriptObject createConfig()
+    /*-{
+        return {
+            transport: 'websocket',
+            fallbackTransport: 'streaming',
+            contentType: 'application/json; charset=UTF-8',
+            reconnectInterval: '5000',
+            trackMessageLength: true 
+        };
+    }-*/;
+
+    private native JavaScriptObject doConnect(String uri,
+            JavaScriptObject config)
+    /*-{
+        var self = this;
+
+        config.url = uri;
+        config.onOpen = $entry(function(response) {
+            self.@com.vaadin.client.communication.PushConnection::onOpen()();
+        });
+        config.onMessage = $entry(function(response) {
+            self.@com.vaadin.client.communication.PushConnection::onMessage(*)(response.responseBody);
+        });
+        config.onError = $entry(function(response) {
+            self.@com.vaadin.client.communication.PushConnection::onError()();
+        });
+                                 
+        return $wnd.atmosphere.subscribe(config);
+    }-*/;
+
+    private native void doPush(JavaScriptObject socket, String message)
+    /*-{
+       socket.push(message);
+    }-*/;
+}