From e03b437442335d33ad52d31a750701933fb021d0 Mon Sep 17 00:00:00 2001 From: fmoga Date: Tue, 24 May 2011 14:27:45 +0000 Subject: 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 --- .../comet/runtime/CometComponentContext.java | 5 ++ .../sca/binding/comet/runtime/CometInvoker.java | 36 +++++--- .../binding/comet/runtime/CometMessageContext.java | 51 ----------- .../comet/runtime/CometServiceBindingProvider.java | 100 +++++++++++---------- .../comet/runtime/callback/CometCallback.java | 4 +- .../sca/binding/comet/runtime/callback/Status.java | 5 ++ .../comet/runtime/handler/CometBindingHandler.java | 46 +++++----- 7 files changed, 111 insertions(+), 136 deletions(-) delete mode 100644 sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometMessageContext.java create mode 100644 sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/callback/Status.java (limited to 'sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org') 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 broadcasters = new ConcurrentHashMap(); + public static Gson gson = new Gson(); private Map endpoints; private Map 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; /** @@ -66,11 +69,6 @@ public class CometBindingHandler { */ private Broadcaster broadcaster; - /** - * JSON converter. - */ - private Gson gson = new Gson(); - /** * The underlying servlet 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 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().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(); - } - } -- cgit v1.2.3