Improve comet support. Add status to callback return type to determine when browser client has closed the page. Upgrade to atmosphere-jquery-0.7.1. Add support for multiple tabs. Fix and improve reliability of long polling technique.

git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@1127080 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 111 insertions, 136 deletions

diff --git a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometComponentContext.java b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometComponentContext.java
index ce19da7e7b..9e93457c7e 100644
--- a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometComponentContext.java
+++ b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometComponentContext.java
@@ -24,9 +24,14 @@ import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.tuscany.sca.interfacedef.Operation;
 import org.apache.tuscany.sca.runtime.RuntimeEndpoint;
+import org.atmosphere.cpr.Broadcaster;
+
+import com.google.gson.Gson;
 
 public class CometComponentContext {
 
+	public static Map<String, Broadcaster> broadcasters = new ConcurrentHashMap<String, Broadcaster>();
+	public static Gson gson = new Gson();
 	private Map<String, RuntimeEndpoint> endpoints;
 	private Map<String, Operation> operations;
 
diff --git a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometInvoker.java b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometInvoker.java
index 5e6375480e..99eb30892e 100644
--- a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometInvoker.java
+++ b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometInvoker.java
@@ -20,11 +20,13 @@
 package org.apache.tuscany.sca.binding.comet.runtime;
 
 import org.apache.tuscany.sca.assembly.EndpointReference;
-import org.apache.tuscany.sca.binding.comet.runtime.handler.CometBindingHandler;
+import org.apache.tuscany.sca.binding.comet.runtime.callback.Status;
+import org.apache.tuscany.sca.core.invocation.Constants;
 import org.apache.tuscany.sca.core.invocation.impl.MessageImpl;
 import org.apache.tuscany.sca.interfacedef.Operation;
 import org.apache.tuscany.sca.invocation.Invoker;
 import org.apache.tuscany.sca.invocation.Message;
+import org.atmosphere.cpr.Broadcaster;
 
 public class CometInvoker implements Invoker {
 
@@ -38,18 +40,28 @@ public class CometInvoker implements Invoker {
 
 	@Override
 	public Message invoke(final Message msg) {
-		String operation = msg.getOperation().getName();
-		CometMessageContext context = msg.getBindingContext();
-		CometBindingHandler handler = context.getCometHandler();
-		Message message = new MessageImpl();
-		if (operation.equals("sendResponse")) {
-			String callbackMethod = context.getCallbackMethod();
+		return handleSendMessage(msg);
+	}
+
+	private Message handleSendMessage(Message msg) {
+		String sessionId = (String) msg.getHeaders().get(Constants.RELATES_TO);
+		Broadcaster broadcaster = CometComponentContext.broadcasters.get(sessionId);
+		Message response = new MessageImpl();
+		if (broadcaster == null) {
+			System.out.println("Broadcaster already removed.");
+			response.setBody(Status.CLIENT_DISCONNECTED);
+		} else if (broadcaster.getAtmosphereResources().isEmpty()) {
+			System.out.println("Removing broadcaster " + sessionId + "...");
+			CometComponentContext.broadcasters.remove(sessionId);
+			response.setBody(Status.CLIENT_DISCONNECTED);
+		} else {
+			System.out.println("Using broadcaster " + sessionId + "...");
+			String callbackMethod = msg.getTo().getURI();
 			Object[] body = msg.getBody();
-			handler.respondToClient(callbackMethod, body[0]);
-		} else if (operation.equals("isClientConnected")) {
-			message.setBody(handler.isClientConnected());
+			broadcaster.broadcast(callbackMethod + "($.secureEvalJSON('" + CometComponentContext.gson.toJson(body[0])
+					+ "'))");
+			response.setBody(Status.OK);
 		}
-		return message;
+		return response;
 	}
-
 }
diff --git a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometMessageContext.java b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometMessageContext.java
deleted file mode 100644
index 21dcc9287f..0000000000
--- a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometMessageContext.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you 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 org.apache.tuscany.sca.binding.comet.runtime;
-
-import org.apache.tuscany.sca.binding.comet.runtime.handler.CometBindingHandler;
-
-public class CometMessageContext {
-
-	private CometBindingHandler cometHandler;
-
-	private String callbackMethod;
-
-	public CometMessageContext(CometBindingHandler cometHandler, String callbackMethod) {
-		this.cometHandler = cometHandler;
-		this.callbackMethod = callbackMethod;
-	}
-
-	public CometBindingHandler getCometHandler() {
-		return cometHandler;
-	}
-
-	public void setCometHandler(CometBindingHandler cometHandler) {
-		this.cometHandler = cometHandler;
-	}
-
-	public String getCallbackMethod() {
-		return callbackMethod;
-	}
-
-	public void setCallbackMethod(String callbackMethod) {
-		this.callbackMethod = callbackMethod;
-	}
-
-}
diff --git a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometServiceBindingProvider.java b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometServiceBindingProvider.java
index 7b6a2f6e1b..96b688aa4f 100644
--- a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometServiceBindingProvider.java
+++ b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometServiceBindingProvider.java
@@ -33,59 +33,63 @@ import org.apache.tuscany.sca.runtime.RuntimeEndpoint;
  */
 public class CometServiceBindingProvider implements ServiceBindingProvider {
 
-    /**
-     * Service's endpoint.
-     */
-    private final RuntimeEndpoint endpoint;
+	/**
+	 * Service's endpoint.
+	 */
+	private final RuntimeEndpoint endpoint;
 
-    /**
-     * The underlying servlet host.
-     */
-    private final ServletHost servletHost;
+	/**
+	 * The underlying servlet host.
+	 */
+	private final ServletHost servletHost;
 
-    /**
-     * Constructor.
-     * 
-     * @param endpoint the given endpoint
-     * @param servletHost the given servlet host
-     */
-    public CometServiceBindingProvider(final RuntimeEndpoint endpoint, final ServletHost servletHost) {
-        this.endpoint = endpoint;
-        this.servletHost = servletHost;
-    }
+	/**
+	 * Constructor.
+	 * 
+	 * @param endpoint
+	 *            the given endpoint
+	 * @param servletHost
+	 *            the given servlet host
+	 */
+	public CometServiceBindingProvider(final RuntimeEndpoint endpoint, final ServletHost servletHost) {
+		this.endpoint = endpoint;
+		this.servletHost = servletHost;
+	}
 
-    /**
-     * This method is used to start the provider.
-     */
-    @Override
-    public void start() {
-        String deployedURI = ServletFactory.registerServlet(this.servletHost);
-        endpoint.setDeployedURI(deployedURI);
-        final ComponentService service = this.endpoint.getService();
-        final Interface serviceInterface = service.getInterfaceContract().getInterface();
-        JavascriptGenerator.generateServiceProxy(service);
-        for (final Operation operation : serviceInterface.getOperations()) {
-            JavascriptGenerator.generateMethodProxy(service, operation);
-            ServletFactory.registerOperation(this.endpoint, operation);
-        }
-    }
+	/**
+	 * This method is used to start the provider.
+	 */
+	@Override
+	public void start() {
+		String deployedURI = ServletFactory.registerServlet(this.servletHost);
+		endpoint.setDeployedURI(deployedURI);
+		final ComponentService service = this.endpoint.getService();
+		final Interface serviceInterface = service.getInterfaceContract().getInterface();
+		JavascriptGenerator.generateServiceProxy(service);
+		for (final Operation operation : serviceInterface.getOperations()) {
+			JavascriptGenerator.generateMethodProxy(service, operation);
+			ServletFactory.registerOperation(this.endpoint, operation);
+		}
+	}
 
-    /**
-     * This method is used to stop the provider.
-     */
-    @Override
-    public void stop() {
-        ServletFactory.unregisterServlet(this.servletHost);
-    }
+	/**
+	 * This method is used to stop the provider.
+	 */
+	@Override
+	public void stop() {
+		ServletFactory.unregisterServlet(this.servletHost);
+	}
 
-    @Override
-    public InterfaceContract getBindingInterfaceContract() {
-        return endpoint.getService().getInterfaceContract();
-    }
+	@Override
+	public InterfaceContract getBindingInterfaceContract() {
+		return endpoint.getService().getInterfaceContract();
+	}
 
-    @Override
-    public boolean supportsOneWayInvocation() {
-        return true;
-    }
+	@Override
+	public boolean supportsOneWayInvocation() {
+		// set to false so the runtime will add a nonBlocking interceptor to
+		// handle @OneWay calls
+		return false;
+	}
 
 }
diff --git a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/callback/CometCallback.java b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/callback/CometCallback.java
index fb9facfd35..cf457e9d7b 100644
--- a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/callback/CometCallback.java
+++ b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/callback/CometCallback.java
@@ -24,8 +24,6 @@ import org.oasisopen.sca.annotation.Remotable;
 @Remotable
 public interface CometCallback {
 
-	void sendResponse(Object response);
-
-	boolean isClientConnected();
+	Status sendMessage(Object message);
 
 }
diff --git a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/callback/Status.java b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/callback/Status.java
new file mode 100644
index 0000000000..6c29d0fa2e
--- /dev/null
+++ b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/callback/Status.java
@@ -0,0 +1,5 @@
+package org.apache.tuscany.sca.binding.comet.runtime.callback;
+
+public enum Status {
+	OK, CLIENT_DISCONNECTED
+}
diff --git a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/handler/CometBindingHandler.java b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/handler/CometBindingHandler.java
index a62d73d467..0d47b72f8b 100644
--- a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/handler/CometBindingHandler.java
+++ b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/handler/CometBindingHandler.java
@@ -32,24 +32,27 @@ import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
 
 import org.apache.tuscany.sca.assembly.EndpointReference;
 import org.apache.tuscany.sca.binding.comet.runtime.CometComponentContext;
-import org.apache.tuscany.sca.binding.comet.runtime.CometMessageContext;
 import org.apache.tuscany.sca.binding.comet.runtime.ServletFactory;
 import org.apache.tuscany.sca.core.assembly.impl.RuntimeEndpointImpl;
 import org.apache.tuscany.sca.core.assembly.impl.RuntimeEndpointReferenceImpl;
+import org.apache.tuscany.sca.core.invocation.Constants;
 import org.apache.tuscany.sca.core.invocation.impl.MessageImpl;
 import org.apache.tuscany.sca.interfacedef.DataType;
 import org.apache.tuscany.sca.interfacedef.Operation;
 import org.apache.tuscany.sca.invocation.Message;
 import org.apache.tuscany.sca.runtime.RuntimeEndpoint;
-import org.atmosphere.annotation.Broadcast;
+import org.atmosphere.cache.SessionBroadcasterCache;
 import org.atmosphere.cpr.Broadcaster;
 import org.atmosphere.cpr.DefaultBroadcaster;
+import org.atmosphere.cpr.DefaultBroadcasterFactory;
+import org.atmosphere.jersey.JerseyBroadcaster;
 import org.atmosphere.jersey.SuspendResponse;
+import org.atmosphere.jersey.util.JerseyBroadcasterUtil;
 
-import com.google.gson.Gson;
 import com.sun.jersey.spi.container.servlet.PerSession;
 
 /**
@@ -67,11 +70,6 @@ public class CometBindingHandler {
 	private Broadcaster broadcaster;
 
 	/**
-	 * JSON converter.
-	 */
-	private Gson gson = new Gson();
-
-	/**
 	 * The underlying servlet context.
 	 */
 	@Context
@@ -82,6 +80,15 @@ public class CometBindingHandler {
 
 	private CometComponentContext context;
 
+	@GET
+	@Path("/sessionId")
+	@Produces(MediaType.TEXT_PLAIN)
+	public String establishSessionId() {
+		request.getSession().invalidate();
+		request.getSession(true);
+		return "OK";
+	}
+
 	/**
 	 * Method called at comet connect time. This suspends the response and keeps
 	 * the connection opened.
@@ -89,12 +96,14 @@ public class CometBindingHandler {
 	 * @return the suspended response
 	 */
 	@GET
+	@Path("/connect")
 	public SuspendResponse<String> connect() {
 		System.out.println("-- connect -- Session Id: " + request.getSession().getId());
 		if (broadcaster == null) {
-			broadcaster = new DefaultBroadcaster();
+			broadcaster = new JerseyBroadcaster();
 			context = (CometComponentContext) sc.getAttribute(ServletFactory.COMET_COMPONENT_CONTEXT_KEY);
 		}
+		CometComponentContext.broadcasters.put(request.getSession().getId(), broadcaster);
 		return new SuspendResponse.SuspendResponseBuilder<String>().broadcaster(this.broadcaster).outputComments(true)
 				.build();
 	}
@@ -117,7 +126,6 @@ public class CometBindingHandler {
 	 */
 	@POST
 	@Path("/{service}/{method}")
-	@Broadcast
 	public void handleRequest(@PathParam("service") final String service, @PathParam("method") final String method,
 			@FormParam("callback") final String callbackMethod, @FormParam("params") final String jsonData)
 			throws InvocationTargetException {
@@ -128,6 +136,7 @@ public class CometBindingHandler {
 
 		final Object[] args = decodeJsonDataForOperation(jsonData, operation);
 		Message msg = createMessageWithMockedCometReference(args, callbackMethod);
+		System.out.println("CometBindingHandler thread id: " + Thread.currentThread().getId());
 		wire.invoke(operation, msg);
 	}
 
@@ -144,7 +153,7 @@ public class CometBindingHandler {
 		int index = 0;
 		// convert each argument to the corresponding class
 		for (final DataType<?> dataType : operation.getInputType().getLogical()) {
-			args[index] = this.gson.fromJson(json[index], dataType.getPhysical());
+			args[index] = CometComponentContext.gson.fromJson(json[index], dataType.getPhysical());
 			index++;
 		}
 		return args;
@@ -161,11 +170,12 @@ public class CometBindingHandler {
 	 */
 	private Message createMessageWithMockedCometReference(Object[] args, String callbackMethod) {
 		Message msg = new MessageImpl();
+		msg.getHeaders().put(Constants.MESSAGE_ID, request.getSession().getId());
 		msg.setBody(args);
-		CometMessageContext messageContext = new CometMessageContext(this, callbackMethod);
-		msg.setBindingContext(messageContext);
 		EndpointReference re = new RuntimeEndpointReferenceImpl();
-		re.setCallbackEndpoint(new RuntimeEndpointImpl());
+		RuntimeEndpointImpl callbackEndpoint = new RuntimeEndpointImpl();
+		callbackEndpoint.setURI(callbackMethod);
+		re.setCallbackEndpoint(callbackEndpoint);
 		msg.setFrom(re);
 		return msg;
 	}
@@ -211,12 +221,4 @@ public class CometBindingHandler {
 		return objects.toArray(new String[] {});
 	}
 
-	public void respondToClient(String callbackMethod, Object response) {
-		broadcaster.broadcast(callbackMethod + "($.secureEvalJSON('" + this.gson.toJson(response) + "'))");
-	}
-
-	public boolean isClientConnected() {
-		return !broadcaster.getAtmosphereResources().isEmpty();
-	}
-
 }