Separate server RPC queue to its own class (#11733)

Change-Id: Ib810406801305cd23a943a4e468610896d166958

9 files changed, 394 insertions, 190 deletions

diff --git a/client/src/com/vaadin/client/ApplicationConnection.java b/client/src/com/vaadin/client/ApplicationConnection.java
index 5772242cff..7dbcf5e1cf 100644
--- a/client/src/com/vaadin/client/ApplicationConnection.java
+++ b/client/src/com/vaadin/client/ApplicationConnection.java
@@ -21,8 +21,6 @@ import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -68,21 +66,18 @@ import com.vaadin.client.ResourceLoader.ResourceLoadEvent;
 import com.vaadin.client.ResourceLoader.ResourceLoadListener;
 import com.vaadin.client.communication.HasJavaScriptConnectorHelper;
 import com.vaadin.client.communication.Heartbeat;
-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.ServerRpcQueue;
 import com.vaadin.client.communication.StateChangeEvent;
 import com.vaadin.client.componentlocator.ComponentLocator;
 import com.vaadin.client.extensions.AbstractExtensionConnector;
 import com.vaadin.client.metadata.ConnectorBundleLoader;
-import com.vaadin.client.metadata.Method;
 import com.vaadin.client.metadata.NoDataException;
 import com.vaadin.client.metadata.Property;
 import com.vaadin.client.metadata.Type;
 import com.vaadin.client.metadata.TypeData;
-import com.vaadin.client.metadata.TypeDataStore;
 import com.vaadin.client.ui.AbstractComponentConnector;
 import com.vaadin.client.ui.AbstractConnector;
 import com.vaadin.client.ui.FontIcon;
@@ -109,7 +104,6 @@ import com.vaadin.shared.util.SharedUtil;
 import elemental.json.Json;
 import elemental.json.JsonArray;
 import elemental.json.JsonObject;
-import elemental.json.JsonValue;
 
 /**
  * This is the client side communication "engine", managing client-server
@@ -188,19 +182,6 @@ public class ApplicationConnection implements HasHandlers {
 
     private final HashMap<String, String> resourcesMap = new HashMap<String, String>();
 
-    /**
-     * The pending method invocations that will be send to the server by
-     * {@link #sendPendingCommand}. The key is defined differently based on
-     * whether the method invocation is enqueued with lastonly. With lastonly
-     * enabled, the method signature ( {@link MethodInvocation#getLastOnlyTag()}
-     * ) is used as the key to make enable removing a previously enqueued
-     * invocation. Without lastonly, an incremental id based on
-     * {@link #lastInvocationTag} is used to get unique values.
-     */
-    private LinkedHashMap<String, MethodInvocation> pendingInvocations = new LinkedHashMap<String, MethodInvocation>();
-
-    private int lastInvocationTag = 0;
-
     private WidgetSet widgetSet;
 
     private VContextMenu contextMenu = null;
@@ -504,6 +485,8 @@ public class ApplicationConnection implements HasHandlers {
         tooltip = GWT.create(VTooltip.class);
         loadingIndicator = GWT.create(VLoadingIndicator.class);
         loadingIndicator.setConnection(this);
+        serverRpcQueue = GWT.create(ServerRpcQueue.class);
+        serverRpcQueue.setConnection(this);
     }
 
     public void init(WidgetSet widgetSet, ApplicationConfiguration cnf) {
@@ -1141,6 +1124,7 @@ public class ApplicationConnection implements HasHandlers {
      */
     private int lastSeenServerSyncId = UNDEFINED_SYNC_ID;
 
+    protected ServerRpcQueue serverRpcQueue;
     /**
      * The value of an undefined sync id.
      * <p>
@@ -1260,7 +1244,7 @@ public class ApplicationConnection implements HasHandlers {
         webkitMaybeIgnoringRequests = false;
 
         if (isApplicationRunning()) {
-            if (sendServerRpcWhenConnectionAvailable) {
+            if (serverRpcQueue.isFlushPending()) {
                 sendInvocationsToServer();
             }
             runPostRequestHooks(configuration.getRootPanelId());
@@ -1271,7 +1255,8 @@ public class ApplicationConnection implements HasHandlers {
             @Override
             public void execute() {
                 if (!isApplicationRunning()
-                        || !(hasActiveRequest() || deferredSendPending)) {
+                        || !(hasActiveRequest() || serverRpcQueue
+                                .isFlushPending())) {
                     getLoadingIndicator().hide();
 
                     // If on Liferay and session expiration management is in
@@ -2636,96 +2621,26 @@ public class ApplicationConnection implements HasHandlers {
             Object value, boolean immediate) {
         boolean lastOnly = !immediate;
         // note that type is now deduced from value
-        addMethodInvocationToQueue(new LegacyChangeVariablesInvocation(
-                connectorId, variableName, value), lastOnly, lastOnly);
-    }
-
-    /**
-     * Adds an explicit RPC method invocation to the send queue.
-     * 
-     * @since 7.0
-     * 
-     * @param invocation
-     *            RPC method invocation
-     * @param delayed
-     *            <code>false</code> to trigger sending within a short time
-     *            window (possibly combining subsequent calls to a single
-     *            request), <code>true</code> to let the framework delay sending
-     *            of RPC calls and variable changes until the next non-delayed
-     *            change
-     * @param lastOnly
-     *            <code>true</code> to remove all previously delayed invocations
-     *            of the same method that were also enqueued with lastonly set
-     *            to <code>true</code>. <code>false</code> to add invocation to
-     *            the end of the queue without touching previously enqueued
-     *            invocations.
-     */
-    public void addMethodInvocationToQueue(MethodInvocation invocation,
-            boolean delayed, boolean lastOnly) {
-        if (!isApplicationRunning()) {
-            getLogger()
-                    .warning(
-                            "Trying to invoke method on not yet started or stopped application");
-            return;
-        }
-        String tag;
-        if (lastOnly) {
-            tag = invocation.getLastOnlyTag();
-            assert !tag.matches("\\d+") : "getLastOnlyTag value must have at least one non-digit character";
-            pendingInvocations.remove(tag);
-        } else {
-            tag = Integer.toString(lastInvocationTag++);
-        }
-        pendingInvocations.put(tag, invocation);
-        if (!delayed) {
-            sendPendingVariableChanges();
+        serverRpcQueue.add(new LegacyChangeVariablesInvocation(connectorId,
+                variableName, value), lastOnly);
+        if (immediate) {
+            serverRpcQueue.flush();
         }
     }
 
     /**
-     * Removes any pending invocation of the given method from the queue
-     * 
-     * @param invocation
-     *            The invocation to remove
-     */
-    public void removePendingInvocations(MethodInvocation invocation) {
-        Iterator<MethodInvocation> iter = pendingInvocations.values()
-                .iterator();
-        while (iter.hasNext()) {
-            MethodInvocation mi = iter.next();
-            if (mi.equals(invocation)) {
-                iter.remove();
-            }
-        }
-    }
-
-    /**
-     * Triggers a send of server RPC and legacy variable changes to the server.
+     * @deprecated as of 7.6, use {@link ServerRpcQueue#flush()}
      */
+    @Deprecated
     public void sendPendingVariableChanges() {
-        if (!deferredSendPending) {
-            deferredSendPending = true;
-            Scheduler.get().scheduleFinally(sendPendingCommand);
-        }
+        serverRpcQueue.flush();
     }
 
-    private final ScheduledCommand sendPendingCommand = new ScheduledCommand() {
-        @Override
-        public void execute() {
-            deferredSendPending = false;
-            doSendPendingVariableChanges();
-        }
-    };
-    private boolean deferredSendPending = false;
-
-    private boolean sendServerRpcWhenConnectionAvailable = false;
-
-    private void doSendPendingVariableChanges() {
+    public void doSendPendingVariableChanges() {
         if (isApplicationRunning()) {
             if (hasActiveRequest() || (push != null && !push.isActive())) {
                 // There is an active request or push is enabled but not active
                 // -> send when current request completes or push becomes active
-                sendServerRpcWhenConnectionAvailable = true;
             } else {
                 sendInvocationsToServer();
             }
@@ -2743,69 +2658,17 @@ public class ApplicationConnection implements HasHandlers {
      * 
      */
     private void sendInvocationsToServer() {
-        boolean showLoadingIndicator = false;
-        JsonArray reqJson = Json.createArray();
-        if (pendingInvocations.isEmpty()) {
+        if (serverRpcQueue.isEmpty()) {
             return;
         }
 
         if (ApplicationConfiguration.isDebugMode()) {
-            Util.logMethodInvocations(this, pendingInvocations.values());
+            Util.logMethodInvocations(this, serverRpcQueue.getAll());
         }
 
-        for (MethodInvocation invocation : pendingInvocations.values()) {
-            String connectorId = invocation.getConnectorId();
-            if (!getConnectorMap().connectorExists(connectorId)) {
-                getLogger().info(
-                        "Not sending RPC for removed connector: " + connectorId
-                                + ": "
-                                + Util.getInvocationDebugString(invocation));
-                continue;
-            }
-
-            JsonArray invocationJson = Json.createArray();
-            invocationJson.set(0, connectorId);
-            invocationJson.set(1, invocation.getInterfaceName());
-            invocationJson.set(2, invocation.getMethodName());
-            JsonArray paramJson = Json.createArray();
-
-            Type[] parameterTypes = null;
-            if (!isLegacyVariableChange(invocation)
-                    && !isJavascriptRpc(invocation)) {
-                try {
-                    Type type = new Type(invocation.getInterfaceName(), null);
-                    Method method = type.getMethod(invocation.getMethodName());
-                    parameterTypes = method.getParameterTypes();
-
-                    showLoadingIndicator |= !TypeDataStore
-                            .isNoLoadingIndicator(method);
-                } catch (NoDataException e) {
-                    throw new RuntimeException("No type data for "
-                            + invocation.toString(), e);
-                }
-            } else {
-                // Always show loading indicator for legacy requests
-                showLoadingIndicator = true;
-            }
-
-            for (int i = 0; i < invocation.getParameters().length; ++i) {
-                // TODO non-static encoder?
-                Type type = null;
-                if (parameterTypes != null) {
-                    type = parameterTypes[i];
-                }
-                Object value = invocation.getParameters()[i];
-                JsonValue jsonValue = JsonEncoder.encode(value, type, this);
-                paramJson.set(i, jsonValue);
-            }
-            invocationJson.set(3, paramJson);
-            reqJson.set(reqJson.length(), invocationJson);
-        }
-
-        pendingInvocations.clear();
-        // Keep tag string short
-        lastInvocationTag = 0;
-        sendServerRpcWhenConnectionAvailable = false;
+        boolean showLoadingIndicator = serverRpcQueue.showLoadingIndicator();
+        JsonArray reqJson = serverRpcQueue.toJson();
+        serverRpcQueue.clear();
 
         if (reqJson.length() == 0) {
             // Nothing to send, all invocations were filtered out (for
@@ -2832,17 +2695,6 @@ public class ApplicationConnection implements HasHandlers {
         makeUidlRequest(reqJson, extraParams);
     }
 
-    private boolean isJavascriptRpc(MethodInvocation invocation) {
-        return invocation instanceof JavaScriptMethodInvocation;
-    }
-
-    private boolean isLegacyVariableChange(MethodInvocation invocation) {
-        return ApplicationConstants.UPDATE_VARIABLE_METHOD.equals(invocation
-                .getInterfaceName())
-                && ApplicationConstants.UPDATE_VARIABLE_METHOD
-                        .equals(invocation.getMethodName());
-    }
-
     /**
      * Sends a new value for the given paintables given variable to the server.
      * <p>
@@ -3642,7 +3494,7 @@ public class ApplicationConnection implements HasHandlers {
                      * Send anything that was enqueued while we waited for the
                      * connection to close
                      */
-                    if (pendingInvocations.size() > 0) {
+                    if (serverRpcQueue.isFlushPending()) {
                         sendPendingVariableChanges();
                     }
                 }
@@ -3707,4 +3559,13 @@ public class ApplicationConnection implements HasHandlers {
     public State getState() {
         return state;
     }
+
+    /**
+     * Gets the server RPC queue for this application
+     * 
+     * @return the server RPC queue
+     */
+    public ServerRpcQueue getServerRpcQueue() {
+        return serverRpcQueue;
+    }
 }
diff --git a/client/src/com/vaadin/client/ConnectorMap.java b/client/src/com/vaadin/client/ConnectorMap.java
index dd1a819b8e..3200dd6ab4 100644
--- a/client/src/com/vaadin/client/ConnectorMap.java
+++ b/client/src/com/vaadin/client/ConnectorMap.java
@@ -101,10 +101,6 @@ public class ConnectorMap {
         return idToComponentDetail.containsKey(connectorId);
     }
 
-    boolean connectorExists(String connectorId) {
-        return hasConnector(connectorId) || isDragAndDropPaintable(connectorId);
-    }
-
     /**
      * Removes all registered connectors
      */
diff --git a/client/src/com/vaadin/client/JavaScriptConnectorHelper.java b/client/src/com/vaadin/client/JavaScriptConnectorHelper.java
index a473073712..eee597b706 100644
--- a/client/src/com/vaadin/client/JavaScriptConnectorHelper.java
+++ b/client/src/com/vaadin/client/JavaScriptConnectorHelper.java
@@ -28,6 +28,7 @@ import com.google.gwt.core.client.JavaScriptObject;
 import com.google.gwt.core.client.JsArray;
 import com.google.gwt.dom.client.Element;
 import com.vaadin.client.communication.JavaScriptMethodInvocation;
+import com.vaadin.client.communication.ServerRpcQueue;
 import com.vaadin.client.communication.StateChangeEvent;
 import com.vaadin.client.communication.StateChangeEvent.StateChangeHandler;
 import com.vaadin.client.ui.layout.ElementResizeEvent;
@@ -357,9 +358,10 @@ public class JavaScriptConnectorHelper {
         for (int i = 0; i < parameters.length; i++) {
             parameters[i] = argumentsArray.get(i);
         }
-        connector.getConnection().addMethodInvocationToQueue(
-                new JavaScriptMethodInvocation(connector.getConnectorId(),
-                        iface, method, parameters), false, false);
+        ServerRpcQueue rpcQueue = ServerRpcQueue.get(connector.getConnection());
+        rpcQueue.add(new JavaScriptMethodInvocation(connector.getConnectorId(),
+                iface, method, parameters), false);
+        rpcQueue.flush();
     }
 
     private String findWildcardInterface(String method) {
@@ -390,8 +392,9 @@ public class JavaScriptConnectorHelper {
                 connector.getConnectorId(),
                 "com.vaadin.ui.JavaScript$JavaScriptCallbackRpc", "call",
                 new Object[] { name, arguments });
-        connector.getConnection().addMethodInvocationToQueue(invocation, false,
-                false);
+        ServerRpcQueue rpcQueue = ServerRpcQueue.get(connector.getConnection());
+        rpcQueue.add(invocation, false);
+        rpcQueue.flush();
     }
 
     public void setNativeState(JavaScriptObject state) {
diff --git a/client/src/com/vaadin/client/Util.java b/client/src/com/vaadin/client/Util.java
index 326a6068c1..b4f777f50e 100644
--- a/client/src/com/vaadin/client/Util.java
+++ b/client/src/com/vaadin/client/Util.java
@@ -759,7 +759,7 @@ public class Util {
      * @param invocation
      * @return
      */
-    static String getInvocationDebugString(MethodInvocation invocation) {
+    private static String getInvocationDebugString(MethodInvocation invocation) {
         Object[] parameters = invocation.getParameters();
         String formattedParams = null;
         if (ApplicationConstants.UPDATE_VARIABLE_METHOD.equals(invocation
diff --git a/client/src/com/vaadin/client/communication/RpcProxy.java b/client/src/com/vaadin/client/communication/RpcProxy.java
index 31b5c92707..b757590f77 100644
--- a/client/src/com/vaadin/client/communication/RpcProxy.java
+++ b/client/src/com/vaadin/client/communication/RpcProxy.java
@@ -58,8 +58,12 @@ public class RpcProxy {
             MethodInvocation invocation = new MethodInvocation(
                     connector.getConnectorId(), rpcInterface.getName(),
                     method.getName(), params);
-            connector.getConnection().addMethodInvocationToQueue(invocation,
-                    method.isDelayed(), method.isLastOnly());
+            ServerRpcQueue serverRpcQueue = ServerRpcQueue.get(connector
+                    .getConnection());
+            serverRpcQueue.add(invocation, method.isLastOnly());
+            if (!method.isDelayed()) {
+                serverRpcQueue.flush();
+            }
             // No RPC iface should have a return value
             return null;
         }
diff --git a/client/src/com/vaadin/client/communication/ServerRpcQueue.java b/client/src/com/vaadin/client/communication/ServerRpcQueue.java
new file mode 100644
index 0000000000..d9241e4c63
--- /dev/null
+++ b/client/src/com/vaadin/client/communication/ServerRpcQueue.java
@@ -0,0 +1,339 @@
+/*
+ * Copyright 2000-2014 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.Collection;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.logging.Logger;
+
+import com.google.gwt.core.client.Scheduler;
+import com.google.gwt.core.client.Scheduler.ScheduledCommand;
+import com.vaadin.client.ApplicationConnection;
+import com.vaadin.client.ConnectorMap;
+import com.vaadin.client.metadata.Method;
+import com.vaadin.client.metadata.NoDataException;
+import com.vaadin.client.metadata.Type;
+import com.vaadin.client.metadata.TypeDataStore;
+import com.vaadin.shared.ApplicationConstants;
+import com.vaadin.shared.communication.MethodInvocation;
+
+import elemental.json.Json;
+import elemental.json.JsonArray;
+import elemental.json.JsonValue;
+
+/**
+ * Manages the queue of server invocations (RPC) which are waiting to be sent to
+ * the server.
+ * 
+ * @since
+ * @author Vaadin Ltd
+ */
+public class ServerRpcQueue {
+
+    /**
+     * The pending method invocations that will be send to the server by
+     * {@link #sendPendingCommand}. The key is defined differently based on
+     * whether the method invocation is enqueued with lastonly. With lastonly
+     * enabled, the method signature ( {@link MethodInvocation#getLastOnlyTag()}
+     * ) is used as the key to make enable removing a previously enqueued
+     * invocation. Without lastonly, an incremental id based on
+     * {@link #lastInvocationTag} is used to get unique values.
+     */
+    private LinkedHashMap<String, MethodInvocation> pendingInvocations = new LinkedHashMap<String, MethodInvocation>();
+
+    private int lastInvocationTag = 0;
+
+    protected ApplicationConnection connection;
+    private boolean flushPending = false;
+
+    private boolean flushScheduled = false;
+
+    public ServerRpcQueue() {
+
+    }
+
+    /**
+     * Sets the application connection this queue is connected to
+     *
+     * @param connection
+     *            the application connection this queue is connected to
+     */
+    public void setConnection(ApplicationConnection connection) {
+        this.connection = connection;
+    }
+
+    public static Logger getLogger() {
+        return Logger.getLogger(ServerRpcQueue.class.getName());
+    }
+
+    /**
+     * Removes any pending invocation of the given method from the queue
+     * 
+     * @param invocation
+     *            The invocation to remove
+     */
+    public void removeMatching(MethodInvocation invocation) {
+        Iterator<MethodInvocation> iter = pendingInvocations.values()
+                .iterator();
+        while (iter.hasNext()) {
+            MethodInvocation mi = iter.next();
+            if (mi.equals(invocation)) {
+                iter.remove();
+            }
+        }
+    }
+
+    /**
+     * Adds an explicit RPC method invocation to the send queue.
+     * 
+     * @param invocation
+     *            RPC method invocation
+     * @param delayed
+     *            <code>false</code> to trigger sending within a short time
+     *            window (possibly combining subsequent calls to a single
+     *            request), <code>true</code> to let the framework delay sending
+     *            of RPC calls and variable changes until the next non-delayed
+     *            change
+     * @param lastOnly
+     *            <code>true</code> to remove all previously delayed invocations
+     *            of the same method that were also enqueued with lastonly set
+     *            to <code>true</code>. <code>false</code> to add invocation to
+     *            the end of the queue without touching previously enqueued
+     *            invocations.
+     */
+    public void add(MethodInvocation invocation, boolean lastOnly) {
+        if (!connection.isApplicationRunning()) {
+            getLogger()
+                    .warning(
+                            "Trying to invoke method on not yet started or stopped application");
+            return;
+        }
+        String tag;
+        if (lastOnly) {
+            tag = invocation.getLastOnlyTag();
+            assert !tag.matches("\\d+") : "getLastOnlyTag value must have at least one non-digit character";
+            pendingInvocations.remove(tag);
+        } else {
+            tag = Integer.toString(lastInvocationTag++);
+        }
+        pendingInvocations.put(tag, invocation);
+    }
+
+    /**
+     * Returns a collection of all queued method invocations
+     * 
+     * @return a collection of all queued method invocations
+     */
+    public Collection<MethodInvocation> getAll() {
+        return pendingInvocations.values();
+    }
+
+    /**
+     * Clears the queue
+     */
+    public void clear() {
+        pendingInvocations.clear();
+        // Keep tag string short
+        lastInvocationTag = 0;
+        flushPending = false;
+    }
+
+    /**
+     * Returns the current size of the queue
+     * 
+     * @return the number of invocations in the queue
+     */
+    public int size() {
+        return pendingInvocations.size();
+    }
+
+    /**
+     * Returns the server RPC queue for the given application
+     * 
+     * @param connection
+     *            the application connection which owns the queue
+     * @return the server rpc queue for the given application
+     */
+    public static ServerRpcQueue get(ApplicationConnection connection) {
+        return connection.getServerRpcQueue();
+    }
+
+    /**
+     * Checks if the queue is empty
+     * 
+     * @return true if the queue is empty, false otherwise
+     */
+    public boolean isEmpty() {
+        return size() == 0;
+    }
+
+    /**
+     * Triggers a send of server RPC and legacy variable changes to the server.
+     */
+    public void flush() {
+        if (flushScheduled) {
+            return;
+        }
+        flushPending = true;
+        flushScheduled = true;
+        Scheduler.get().scheduleFinally(scheduledFlushCommand);
+    }
+
+    private final ScheduledCommand scheduledFlushCommand = new ScheduledCommand() {
+        @Override
+        public void execute() {
+            flushScheduled = false;
+            if (!isFlushPending()) {
+                // Somebody else cleared the queue before we had the chance
+                return;
+            }
+            connection.doSendPendingVariableChanges();
+        }
+    };
+
+    /**
+     * Checks if a flush operation is pending
+     * 
+     * @return true if a flush is pending, false otherwise
+     */
+    public boolean isFlushPending() {
+        return flushPending;
+    }
+
+    /**
+     * Checks if a loading indicator should be shown when the RPCs have been
+     * sent to the server and we are waiting for a response
+     * 
+     * @return true if a loading indicator should be shown, false otherwise
+     */
+    public boolean showLoadingIndicator() {
+        for (MethodInvocation invocation : getAll()) {
+            if (isLegacyVariableChange(invocation)) {
+                // Always show loading indicator for legacy requests
+                return true;
+            } else if (!isJavascriptRpc(invocation)) {
+                Type type = new Type(invocation.getInterfaceName(), null);
+                Method method = type.getMethod(invocation.getMethodName());
+                if (!TypeDataStore.isNoLoadingIndicator(method)) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
+    /**
+     * Returns the current invocations as JSON
+     * 
+     * @return the current invocations in a JSON format ready to be sent to the
+     *         server
+     */
+    public JsonArray toJson() {
+        JsonArray json = Json.createArray();
+        if (isEmpty()) {
+            return json;
+        }
+
+        for (MethodInvocation invocation : getAll()) {
+            String connectorId = invocation.getConnectorId();
+            if (!connectorExists(connectorId)) {
+                getLogger().info(
+                        "Ignoring RPC for removed connector: " + connectorId
+                                + ": " + invocation.toString());
+                continue;
+            }
+
+            JsonArray invocationJson = Json.createArray();
+            invocationJson.set(0, connectorId);
+            invocationJson.set(1, invocation.getInterfaceName());
+            invocationJson.set(2, invocation.getMethodName());
+            JsonArray paramJson = Json.createArray();
+
+            Type[] parameterTypes = null;
+            if (!isLegacyVariableChange(invocation)
+                    && !isJavascriptRpc(invocation)) {
+                try {
+                    Type type = new Type(invocation.getInterfaceName(), null);
+                    Method method = type.getMethod(invocation.getMethodName());
+                    parameterTypes = method.getParameterTypes();
+                } catch (NoDataException e) {
+                    throw new RuntimeException("No type data for "
+                            + invocation.toString(), e);
+                }
+            }
+
+            for (int i = 0; i < invocation.getParameters().length; ++i) {
+                // TODO non-static encoder?
+                Type type = null;
+                if (parameterTypes != null) {
+                    type = parameterTypes[i];
+                }
+                Object value = invocation.getParameters()[i];
+                JsonValue jsonValue = JsonEncoder.encode(value, type,
+                        connection);
+                paramJson.set(i, jsonValue);
+            }
+            invocationJson.set(3, paramJson);
+            json.set(json.length(), invocationJson);
+        }
+
+        return json;
+    }
+
+    /**
+     * Checks if the connector with the given id is still ok to use (has not
+     * been removed)
+     * 
+     * @param connectorId
+     *            the connector id to check
+     * @return true if the connector exists, false otherwise
+     */
+    private boolean connectorExists(String connectorId) {
+        ConnectorMap connectorMap = ConnectorMap.get(connection);
+        return connectorMap.hasConnector(connectorId)
+                || connectorMap.isDragAndDropPaintable(connectorId);
+    }
+
+    /**
+     * Checks if the given method invocation originates from Javascript
+     * 
+     * @param invocation
+     *            the invocation to check
+     * @return true if the method invocation originates from javascript, false
+     *         otherwise
+     */
+    public static boolean isJavascriptRpc(MethodInvocation invocation) {
+        return invocation instanceof JavaScriptMethodInvocation;
+    }
+
+    /**
+     * Checks if the given method invocation represents a Vaadin 6 variable
+     * change
+     * 
+     * @param invocation
+     *            the invocation to check
+     * @return true if the method invocation is a legacy variable change, false
+     *         otherwise
+     */
+    public static boolean isLegacyVariableChange(MethodInvocation invocation) {
+        return ApplicationConstants.UPDATE_VARIABLE_METHOD.equals(invocation
+                .getInterfaceName())
+                && ApplicationConstants.UPDATE_VARIABLE_METHOD
+                        .equals(invocation.getMethodName());
+    }
+
+}
diff --git a/client/src/com/vaadin/client/extensions/javascriptmanager/JavaScriptManagerConnector.java b/client/src/com/vaadin/client/extensions/javascriptmanager/JavaScriptManagerConnector.java
index d48571452e..364b948573 100644
--- a/client/src/com/vaadin/client/extensions/javascriptmanager/JavaScriptManagerConnector.java
+++ b/client/src/com/vaadin/client/extensions/javascriptmanager/JavaScriptManagerConnector.java
@@ -24,6 +24,7 @@ import com.google.gwt.core.client.JsArray;
 import com.vaadin.client.ServerConnector;
 import com.vaadin.client.Util;
 import com.vaadin.client.communication.JavaScriptMethodInvocation;
+import com.vaadin.client.communication.ServerRpcQueue;
 import com.vaadin.client.communication.StateChangeEvent;
 import com.vaadin.client.extensions.AbstractExtensionConnector;
 import com.vaadin.shared.extension.javascriptmanager.ExecuteJavaScriptRpc;
@@ -122,10 +123,11 @@ public class JavaScriptManagerConnector extends AbstractExtensionConnector {
          * Must invoke manually as the RPC interface can't be used in GWT
          * because of the JSONArray parameter
          */
-        getConnection().addMethodInvocationToQueue(
-                new JavaScriptMethodInvocation(getConnectorId(),
-                        "com.vaadin.ui.JavaScript$JavaScriptCallbackRpc",
-                        "call", parameters), false, false);
+        ServerRpcQueue rpcQueue = ServerRpcQueue.get(getConnection());
+        rpcQueue.add(new JavaScriptMethodInvocation(getConnectorId(),
+                "com.vaadin.ui.JavaScript$JavaScriptCallbackRpc", "call",
+                parameters), false);
+        rpcQueue.flush();
     }
 
     @Override
diff --git a/client/src/com/vaadin/client/ui/datefield/DateFieldConnector.java b/client/src/com/vaadin/client/ui/datefield/DateFieldConnector.java
index 042b8db8e2..c6b410fa74 100644
--- a/client/src/com/vaadin/client/ui/datefield/DateFieldConnector.java
+++ b/client/src/com/vaadin/client/ui/datefield/DateFieldConnector.java
@@ -32,7 +32,6 @@ import com.vaadin.shared.ui.Connect;
 import com.vaadin.shared.ui.datefield.PopupDateFieldState;
 import com.vaadin.shared.ui.datefield.Resolution;
 import com.vaadin.ui.DateField;
-import com.vaadin.ui.PopupDateField;
 
 @Connect(DateField.class)
 public class DateFieldConnector extends TextualDateConnector {
@@ -60,7 +59,7 @@ public class DateFieldConnector extends TextualDateConnector {
                  * communicated to the server.
                  */
                 if (getWidget().isImmediate()) {
-                    getConnection().sendPendingVariableChanges();
+                    getConnection().getServerRpcQueue().flush();
                 }
             }
         });
diff --git a/client/src/com/vaadin/client/ui/ui/UIConnector.java b/client/src/com/vaadin/client/ui/ui/UIConnector.java
index cfb444ada6..60e9587493 100644
--- a/client/src/com/vaadin/client/ui/ui/UIConnector.java
+++ b/client/src/com/vaadin/client/ui/ui/UIConnector.java
@@ -173,7 +173,7 @@ public class UIConnector extends AbstractSingleComponentContainerConnector
                         event.getWidth(), Window.getClientWidth(),
                         Window.getClientHeight());
                 if (getState().immediate || getPageState().hasResizeListeners) {
-                    getConnection().sendPendingVariableChanges();
+                    getConnection().getServerRpcQueue().flush();
                 }
             }
         });
@@ -775,13 +775,13 @@ public class UIConnector extends AbstractSingleComponentContainerConnector
                     }
                     getRpcProxy(UIServerRpc.class).poll();
                     // Send changes even though poll is @Delayed
-                    getConnection().sendPendingVariableChanges();
+                    getConnection().getServerRpcQueue().flush();
                 }
             };
             pollTimer.scheduleRepeating(getState().pollInterval);
         } else {
             // Ensure no more polls are sent as polling has been disabled
-            getConnection().removePendingInvocations(
+            getConnection().getServerRpcQueue().removeMatching(
                     new MethodInvocation(getConnectorId(), UIServerRpc.class
                             .getName(), "poll"));
         }