Use the BroadcasterLifeCyclePolicy and new callback mechanism from Atmosphere to improve resource cleanup. Move to Atmosphere 0.8-SNAPSHOT.

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

3 files changed, 49 insertions, 12 deletions

diff --git a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometCallbackInvoker.java b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometCallbackInvoker.java
index c3733e6dfe..a9d91d2f48 100644
--- a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometCallbackInvoker.java
+++ b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/CometCallbackInvoker.java
@@ -65,9 +65,6 @@ public class CometCallbackInvoker implements Invoker {
         Message response = new MessageImpl();
         if (broadcaster == null) {
             response.setBody(Status.CLIENT_DISCONNECTED);
-        } else if (broadcaster.getAtmosphereResources().isEmpty()) {
-            CometSessionManager.remove(sessionId);
-            response.setBody(Status.CLIENT_DISCONNECTED);
         } else {
             String callbackMethod = msg.getTo().getURI();
             Object[] body = msg.getBody();
diff --git a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/ServletFactory.java b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/ServletFactory.java
index a334eaf006..f40d1f62ce 100644
--- a/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/ServletFactory.java
+++ b/sca-java-2.x/trunk/modules/binding-comet-runtime/src/main/java/org/apache/tuscany/sca/binding/comet/runtime/ServletFactory.java
@@ -56,6 +56,16 @@ public final class ServletFactory {
     private static final String JS_PACKAGE = "org.apache.tuscany.sca.binding.comet.runtime.javascript";
 
     /**
+     * Init-param key for Atmosphere filters.
+     */
+    private static final String FILTERS_KEY = "org.atmosphere.cpr.broadcastFilterClasses";
+
+    /**
+     * Defined filters.
+     */
+    private static final String FILTERS = "org.atmosphere.client.JavascriptClientFilter";
+
+    /**
      * Path where services will be exposed.
      */
     public static final String PATH = "/tuscany-comet/*";
@@ -99,8 +109,9 @@ public final class ServletFactory {
     private static String registerCometServlet(ServletHost servletHost) {
         if (ServletFactory.cometServlet == null) {
             ServletFactory.cometServlet = new AtmosphereServlet();
-            ServletFactory.cometServlet.addInitParameter(ServletFactory.PACKAGE_KEY, ServletFactory.HANDLER_PACKAGE);
-            String uri = servletHost.addServletMapping(ServletFactory.PATH, ServletFactory.cometServlet);
+            ServletFactory.cometServlet.addInitParameter(PACKAGE_KEY, HANDLER_PACKAGE);
+//            ServletFactory.cometServlet.addInitParameter(FILTERS_KEY, FILTERS);
+            String uri = servletHost.addServletMapping(PATH, cometServlet);
             return uri;
         }
         return null;
@@ -109,8 +120,8 @@ public final class ServletFactory {
     private static void registerJavascriptServlet(ServletHost servletHost) {
         if (ServletFactory.javascriptServlet == null) {
             ServletFactory.javascriptServlet = new AtmosphereServlet();
-            ServletFactory.javascriptServlet.addInitParameter(ServletFactory.PACKAGE_KEY, ServletFactory.JS_PACKAGE);
-            servletHost.addServletMapping(ServletFactory.JS_PATH, ServletFactory.javascriptServlet);
+            ServletFactory.javascriptServlet.addInitParameter(PACKAGE_KEY, JS_PACKAGE);
+            servletHost.addServletMapping(JS_PATH, javascriptServlet);
         }
     }
 
@@ -121,8 +132,8 @@ public final class ServletFactory {
      *            the underlying servlet host.
      */
     public static void unregisterServlet(final ServletHost servletHost) {
-        servletHost.removeServletMapping(ServletFactory.PATH);
-        servletHost.removeServletMapping(ServletFactory.JS_PATH);
+        servletHost.removeServletMapping(PATH);
+        servletHost.removeServletMapping(JS_PATH);
     }
 
 }
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 9035c2b599..7c23a4e73b 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
@@ -43,6 +43,9 @@ 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.BroadcasterLifeCyclePolicyListener;
+import org.atmosphere.cpr.BroadcasterLifeCyclePolicy.ATMOSPHERE_RESOURCE_POLICY;
 import org.atmosphere.jersey.JerseyBroadcaster;
 import org.atmosphere.jersey.SuspendResponse;
 
@@ -73,6 +76,10 @@ public class CometBindingHandler {
         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<String>().broadcaster(broadcaster).outputComments(true)
@@ -110,9 +117,8 @@ public class CometBindingHandler {
         if (!isVoidReturnType) {
             Object response = wire.invoke(operation, args);
             Broadcaster broadcaster = CometSessionManager.get(sessionId);
-            broadcaster.broadcast(callbackMethod + "($.secureEvalJSON('" + gson.toJson(response) + "'))");
-            if (broadcaster.getAtmosphereResources().isEmpty()) {
-                CometSessionManager.remove(sessionId);
+            if (broadcaster != null) {
+                broadcaster.broadcast(callbackMethod + "($.secureEvalJSON('" + gson.toJson(response) + "'))");
             }
         } else {
             wire.invoke(operation, msg);
@@ -205,4 +211,27 @@ public class CometBindingHandler {
         return objects.toArray(new String[] {});
     }
 
+    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() {
+        }
+
+    }
+
 }