From 132aa8a77685ec92bc90c03f987650d275a7b639 Mon Sep 17 00:00:00 2001 From: lresende Date: Mon, 30 Sep 2013 06:59:11 +0000 Subject: 2.0.1 RC1 release tag git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@1527464 13f79535-47bb-0310-9956-ffa450edef68 --- .../comet/runtime/handler/CometBindingHandler.java | 168 +++++++++++++++++++++ 1 file changed, 168 insertions(+) create mode 100644 sca-java-2.x/tags/2.0.1-RC1/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/handler/CometBindingHandler.java (limited to 'sca-java-2.x/tags/2.0.1-RC1/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/handler/CometBindingHandler.java') diff --git a/sca-java-2.x/tags/2.0.1-RC1/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/handler/CometBindingHandler.java b/sca-java-2.x/tags/2.0.1-RC1/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/handler/CometBindingHandler.java new file mode 100644 index 0000000000..0c9902211c --- /dev/null +++ b/sca-java-2.x/tags/2.0.1-RC1/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/handler/CometBindingHandler.java @@ -0,0 +1,168 @@ +/* + * 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.handler; + +import java.lang.reflect.InvocationTargetException; + +import javax.ws.rs.FormParam; +import javax.ws.rs.GET; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.QueryParam; + +import org.apache.tuscany.sca.assembly.EndpointReference; +import org.apache.tuscany.sca.binding.comet.runtime.JSONUtil; +import org.apache.tuscany.sca.binding.comet.runtime.manager.CometEndpointManager; +import org.apache.tuscany.sca.binding.comet.runtime.manager.CometOperationManager; +import org.apache.tuscany.sca.binding.comet.runtime.manager.CometSessionManager; +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.Operation; +import org.apache.tuscany.sca.invocation.Message; +import org.apache.tuscany.sca.runtime.RuntimeEndpoint; +import org.atmosphere.cpr.Broadcaster; +import org.atmosphere.cpr.BroadcasterLifeCyclePolicy; +import org.atmosphere.cpr.BroadcasterLifeCyclePolicy.ATMOSPHERE_RESOURCE_POLICY; +import org.atmosphere.cpr.BroadcasterLifeCyclePolicyListener; +import org.atmosphere.jersey.JerseyBroadcaster; +import org.atmosphere.jersey.SuspendResponse; + +/** + * Handles requests for comet services and for creating a persistent connection. + */ +@Path("/") +public class CometBindingHandler { + + /** + * Suspends the current HTTP connection. + * + * @param sessionId + * session id to identify client + * @return a response that is not committed, just flushed + */ + @GET + @Path("/connect") + public SuspendResponse connect(@QueryParam("sessionId") String sessionId) { + Broadcaster broadcaster = CometSessionManager.get(sessionId); + if (broadcaster == null) { + broadcaster = new JerseyBroadcaster(sessionId); + BroadcasterLifeCyclePolicy policy = new BroadcasterLifeCyclePolicy.Builder().policy( + ATMOSPHERE_RESOURCE_POLICY.EMPTY_DESTROY).build(); + broadcaster.setBroadcasterLifeCyclePolicy(policy); + broadcaster.addBroadcasterLifeCyclePolicyListener(new CometBroadcasterLifeCyclePolicyListener(sessionId)); + CometSessionManager.add(sessionId, broadcaster); + } + return new SuspendResponse.SuspendResponseBuilder().broadcaster(broadcaster).outputComments(true) + .build(); + } + + /** + * Handles requests for service operations. + * + * @param service + * the service to invoke + * @param method + * the method to invoke + * @param sessionId + * the client session id + * @param callbackMethod + * the callbackMethod to invoke once a response is available + * @param jsonData + * method arguments sent by the client in JSON format + * @throws InvocationTargetException + * if a problem occurs while invoking the service implementation + */ + @POST + @Path("/{service}/{method}") + public void handleRequest(@PathParam("service") String service, @PathParam("method") String method, + @FormParam("sessionId") String sessionId, @FormParam("callbackMethod") String callbackMethod, + @FormParam("params") String jsonData) throws InvocationTargetException { + String url = "/" + service + "/" + method; + RuntimeEndpoint wire = CometEndpointManager.get(url); + Operation operation = CometOperationManager.get(url); + + final Object[] args = JSONUtil.decodeJsonParamsForOperation(jsonData, operation); + Message msg = createMessageWithMockedCometReference(args, sessionId, callbackMethod); + boolean isVoidReturnType = operation.getOutputType().getLogical().isEmpty(); + if (!isVoidReturnType) { + Object response = wire.invoke(operation, args); + Broadcaster broadcaster = CometSessionManager.get(sessionId); + if (broadcaster != null) { + broadcaster.broadcast(callbackMethod + "($.secureEvalJSON('" + JSONUtil.encodeResponse(response) + + "'))"); + } + } else { + wire.invoke(operation, msg); + } + } + + /** + * Creates a message with a mocked EndpointReference in the 'from' field to + * simulate a comet reference (because requests are coming from browsers). + * This is needed by the callback mechanism to have a source for the + * request. + * + * @param args + * arguments for the method invocation + * @param sessionId + * the session id of the client + * @param callbackMethod + * method to call once a response is available + * @return an invocation message + */ + private Message createMessageWithMockedCometReference(Object[] args, String sessionId, String callbackMethod) { + Message msg = new MessageImpl(); + msg.getHeaders().put(Constants.MESSAGE_ID, sessionId); + msg.setBody(args); + EndpointReference re = new RuntimeEndpointReferenceImpl(); + RuntimeEndpointImpl callbackEndpoint = new RuntimeEndpointImpl(); + callbackEndpoint.setURI(callbackMethod); + re.setCallbackEndpoint(callbackEndpoint); + msg.setFrom(re); + return msg; + } + + public class CometBroadcasterLifeCyclePolicyListener implements BroadcasterLifeCyclePolicyListener { + + private String sessionId; + + public CometBroadcasterLifeCyclePolicyListener(String sessionId) { + this.sessionId = sessionId; + } + + @Override + public void onDestroy() { + } + + @Override + public void onEmpty() { + CometSessionManager.remove(sessionId); + } + + @Override + public void onIdle() { + } + + } + +} -- cgit v1.2.3