Atmosphere almost integrated, ClassLoader issue.

bin/start-akka-server.sh

@@ -15,6 +15,7 @@ mkdir $BASE_DIR/storage/commitlog
 mkdir $BASE_DIR/storage/data
 mkdir $BASE_DIR/storage/system
 
+CLASSPATH=$BASE_DIR/config:$BASE_DIR/deploy/root:$BASE_DIR/deploy/root/META-INF:$BASE_DIR/deploy/root/WEB-INF/classes
 CLASSPATH=$CLASSPATH:$BASE_DIR/config
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/akka-kernel-0.5.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/akka-util-java-0.5.jar
@@ -35,11 +36,12 @@ CLASSPATH=$CLASSPATH:$BASE_DIR/lib/commons-logging-1.0.4.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/commons-math-1.1.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/configgy-1.3.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/google-collect-snapshot-20090211.jar
-CLASSPATH=$CLASSPATH:$BASE_DIR/lib/grizzly-framework-1.8.6.3.jar
+#CLASSPATH=$CLASSPATH:$BASE_DIR/lib/grizzly-framework-1.8.6.3.jar
-CLASSPATH=$CLASSPATH:$BASE_DIR/lib/grizzly-http-1.8.6.3.jar
+#CLASSPATH=$CLASSPATH:$BASE_DIR/lib/grizzly-http-1.8.6.3.jar
-CLASSPATH=$CLASSPATH:$BASE_DIR/lib/grizzly-http-servlet-1.8.6.3.jar
+#CLASSPATH=$CLASSPATH:$BASE_DIR/lib/grizzly-http-servlet-1.8.6.3.jar
-CLASSPATH=$CLASSPATH:$BASE_DIR/lib/grizzly-http-utils-1.8.6.3.jar
+#CLASSPATH=$CLASSPATH:$BASE_DIR/lib/grizzly-http-utils-1.8.6.3.jar
-CLASSPATH=$CLASSPATH:$BASE_DIR/lib/grizzly-servlet-webserver-1.8.6.3.jar
+#CLASSPATH=$CLASSPATH:$BASE_DIR/lib/grizzly-servlet-webserver-1.8.6.3.jar
+CLASSPATH=$CLASSPATH:$BASE_DIR/lib/grizzly-comet-webserver-1.9.15b.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/guice-core-2.0-SNAPSHOT.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/guice-jsr250-2.0-SNAPSHOT.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/high-scale-lib.jar
@@ -50,7 +52,10 @@ CLASSPATH=$CLASSPATH:$BASE_DIR/lib/jersey-core-1.0.3.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/jersey-json-1.0.3.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/jersey-server-1.0.3.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/jersey-scala-1.1.2-ea-SNAPSHOT.jar
-CLASSPATH=$CLASSPATH:$BASE_DIR/lib/jersey-lift-1.1.2-ea-SNAPSHOT.jar
+CLASSPATH=$CLASSPATH:$BASE_DIR/lib/atmosphere-core-0.3-SNAPSHOT.jar
+CLASSPATH=$CLASSPATH:$BASE_DIR/lib/atmosphere-portable-runtime-0.3-SNAPSHOT.jar
+CLASSPATH=$CLASSPATH:$BASE_DIR/lib/atmosphere-compat-0.3-SNAPSHOT.jar
+#CLASSPATH=$CLASSPATH:$BASE_DIR/lib/atmosphere-grizzly-adapter-0.3-SNAPSHOT.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/JSAP-2.1.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/jsr250-api-1.0.jar
 CLASSPATH=$CLASSPATH:$BASE_DIR/lib/jsr311-api-1.0.jar

config/web.xml

@@ -1,10 +0,0 @@
-<web-app version="2.5" xmlns="http://java.sun.com/xml/ns/javaee">
-  <servlet>
-    <servlet-name>Akka Servlet</servlet-name>
-    <servlet-class>se.scalablesolutions.akka.kernel.jersey.AkkaServlet</servlet-class>
-  </servlet>
-  <servlet-mapping>
-    <servlet-name>Akka Servlet</servlet-name>
-    <url-pattern>/*</url-pattern>
-  </servlet-mapping>
-</web-app>

deploy/root/META-INF/atmosphere.xml

@@ -0,0 +1,5 @@
+<atmosphere-handlers>
+    <atmosphere-handler context-root="" class-name="org.atmosphere.handler.ReflectorServletProcessor">
+        <property name="servletClass" value="se.scalablesolutions.akka.kernel.jersey.AkkaServlet"/>
+    </atmosphere-handler>
+</atmosphere-handlers>

deploy/root/META-INF/context.xml

@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Context>
+    <!-- <Loader className="org.atmosphere.util.AtmosphereClassloader"/> -->
+    <Loader delegate="true"/>
+</Context>

kernel/pom.xml

@@ -105,10 +105,15 @@
         </dependency>
 
         <!-- For Jersey -->
-        <dependency>
+        <!--<dependency>
             <groupId>com.sun.grizzly</groupId>
             <artifactId>grizzly-servlet-webserver</artifactId>
             <version>1.9.9</version>
+        </dependency>-->
+        <dependency>
+            <groupId>com.sun.grizzly</groupId>
+            <artifactId>grizzly-comet-webserver</artifactId>
+            <version>1.9.15b</version>
         </dependency>
         <dependency>
             <groupId>com.sun.jersey</groupId>
@@ -130,11 +135,32 @@
             <artifactId>jersey-scala</artifactId>
             <version>1.1.2-ea-SNAPSHOT</version>
         </dependency>
-        <dependency>
+        <!--<dependency>
             <groupId>com.sun.jersey.contribs</groupId>
             <artifactId>jersey-lift</artifactId>
             <version>1.1.2-ea-SNAPSHOT</version>
+        </dependency>-->
+        
+        <dependency>
+            <groupId>org.atmosphere</groupId>
+            <artifactId>atmosphere-core</artifactId>
+            <version>0.3-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.atmosphere</groupId>
+            <artifactId>atmosphere-portable-runtime</artifactId>
+            <version>0.3-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.atmosphere</groupId>
+            <artifactId>atmosphere-compat</artifactId>
+            <version>0.3-SNAPSHOT</version>
+        </dependency>
+        <!--<dependency>
+            <groupId>org.atmosphere</groupId>
+            <artifactId>atmosphere-grizzly-adapter</artifactId>
+            <version>0.3-SNAPSHOT<</version>
+        </dependency>-->
 
         <!-- For third-party logging -->
         <dependency>
@@ -195,7 +221,7 @@
                     <execution>
                         <goals>
                             <goal>compile</goal>
-                            <goal>testCompile</goal>
+                            <!--<goal>testCompile</goal>-->
                         </goals>
                     </execution>
                 </executions>

kernel/src/main/scala/Kernel.scala

@@ -14,7 +14,7 @@ import java.net.URLClassLoader
 
 import net.lag.configgy.{Config, Configgy, RuntimeEnvironment}
 
-import kernel.jersey.AkkaServlet
+import kernel.jersey.{AkkaServlet,AkkaCometServlet}
 import kernel.nio.RemoteServer
 import kernel.state.CassandraStorage
 import kernel.util.Logging
@@ -36,7 +36,7 @@ object Kernel extends Logging {
   val REST_HOSTNAME = kernel.Kernel.config.getString("akka.rest.hostname", "localhost")
   val REST_URL = "http://" + REST_HOSTNAME
   val REST_PORT = kernel.Kernel.config.getInt("akka.rest.port", 9998)
-
+  val REST_ROOT = System.getenv("AKKA_HOME") + "/deploy/root/"
 
   // FIXME add API to shut server down gracefully
   private var remoteServer: RemoteServer = _
@@ -91,8 +91,8 @@ object Kernel extends Logging {
 
   private[akka] def runApplicationBootClasses = {
     val HOME = try { System.getenv("AKKA_HOME") } catch { case e: NullPointerException => throw new IllegalStateException("AKKA_HOME system variable needs to be set. Should point to the root of the Akka distribution.") }
-    val CLASSES = HOME + "/kernel/target/classes" // FIXME remove for dist
+    //val CLASSES = HOME + "/kernel/target/classes" // FIXME remove for dist
-    val LIB = HOME + "/lib"
+    //val LIB = HOME + "/lib"
     val CONFIG = HOME + "/config"
     val DEPLOY = HOME + "/deploy"
     val DEPLOY_DIR = new File(DEPLOY)
@@ -123,22 +123,25 @@ object Kernel extends Logging {
 
   private[akka] def startJersey = {
     val uri = UriBuilder.fromUri(REST_URL).port(REST_PORT).build()
-    val adapter = new ServletAdapter
-    val servlet = new AkkaServlet
-
-    adapter.addInitParameter("bootloader", config.getString("akka.lift.bootloader").getOrElse(null))
-    adapter.addInitParameter("com.sun.jersey.config.feature.Redirect", "true")
-    adapter.addInitParameter("com.sun.jersey.config.feature.ImplicitViewables", "true")
-    adapter.setServletInstance(servlet)
-    adapter.setContextPath(uri.getPath)
-
-        //TODO add initialization of Lift (LiftFilter needs initialization)
-
-    log.debug("REST service context path: [" + uri.getPath + "]")
 
     val scheme = uri.getScheme
     if (!scheme.equalsIgnoreCase("http")) throw new IllegalArgumentException("The URI scheme, of the URI " + REST_URL + ", must be equal (ignoring case) to 'http'")
 
+
+    val adapter = new ServletAdapter()
+    //val servlet = org.atmosphere.cpr.AtmosphereServlet
+    //val servlet = new AkkaServlet()
+    //adapter.addInitParameter("bootloader", config.getString("akka.lift.bootloader").getOrElse(null))
+    adapter.addInitParameter("com.sun.jersey.spi.container.ResourceFilters","org.atmosphere.core.AtmosphereFilter")
+    adapter.addInitParameter("com.sun.jersey.config.feature.Redirect", "true")
+    adapter.addInitParameter("com.sun.jersey.config.feature.ImplicitViewables", "true")
+
+    adapter.setServletInstance(new AkkaCometServlet)
+    adapter.setContextPath(uri.getPath)
+    adapter.setRootFolder(REST_ROOT)
+    log.info("REST service root path: [" + adapter.getRootFolder + "] and context path [" + adapter.getContextPath + "] ")
+
+    
     jerseySelectorThread = new SelectorThread
     jerseySelectorThread.setAlgorithmClassName(classOf[StaticStreamAlgorithm].getName)
     jerseySelectorThread.setPort(REST_PORT)

kernel/src/main/scala/jersey/AkkaServlet.scala

@@ -6,12 +6,17 @@ package se.scalablesolutions.akka.kernel.jersey
 
 import kernel.Kernel
 import config.ConfiguratorRepository
+import util.Logging
 
 import com.sun.jersey.api.core.{DefaultResourceConfig, ResourceConfig}
 import com.sun.jersey.spi.container.servlet.ServletContainer
 import com.sun.jersey.spi.container.WebApplication
+
+import javax.servlet.{ServletConfig}
+import java.net.{URL,URLClassLoader}
 import java.util.HashSet
 
+
 class AkkaServlet extends ServletContainer{
 
   override def initiate(rc: ResourceConfig, wa: WebApplication) = {
@@ -29,4 +34,17 @@ class AkkaServlet extends ServletContainer{
       new DefaultResourceConfig(set),
       new ActorComponentProviderFactory(configurators))
   }
-}
+}
+
+class AkkaCometServlet extends org.atmosphere.cpr.AtmosphereServlet with Logging
+  {
+      override def init(sconf : ServletConfig) = {
+
+        val url = sconf.getServletContext().getResource("/WEB-INF/classes/")
+        val urlC = new URLClassLoader(Array(url),Thread.currentThread().getContextClassLoader())
+
+        loadAtmosphereDotXml(sconf.getServletContext.getResourceAsStream("META-INF/atmosphere.xml"), urlC);
+
+        super.init(sconf)
+      }
+  }

samples-scala/src/main/scala/SimpleService.scala

@@ -1,26 +1,31 @@
 package sample.scala
 
-import javax.ws.rs.{Path, GET, Produces,QueryParam,DefaultValue}
+import javax.ws.rs.{GET, POST, Path, Produces, WebApplicationException, Consumes}
 import se.scalablesolutions.akka.kernel.state.{TransactionalState, TransactionalMap, CassandraStorageConfig}
 import se.scalablesolutions.akka.kernel.actor.{Supervisor, SupervisorFactory, Actor, StartSupervisor}
 import se.scalablesolutions.akka.kernel.config.ScalaConfig._
+import javax.ws.rs.core.MultivaluedMap
+
 
 import _root_.scala.xml.{NodeSeq}
 import se.scalablesolutions.akka.kernel.util.{Logging}
+import org.atmosphere.core.annotation.{Broadcast, BroadcastFilter, Suspend}
+import org.atmosphere.util.{XSSHtmlFilter}
+
 
 class Boot {
-  object factory extends SupervisorFactory {
+    object factory extends SupervisorFactory {
-    override def getSupervisorConfig: SupervisorConfig = {
+        override def getSupervisorConfig: SupervisorConfig = {
-      SupervisorConfig(
+            SupervisorConfig(
-        RestartStrategy(OneForOne, 3, 100),
+                RestartStrategy(OneForOne, 3, 100),
-        Supervise(
+                  Supervise(
-          new SimpleService,
+                    new Chat,
-          LifeCycle(Permanent, 100))
+                      LifeCycle(Permanent, 100))
-        :: Nil)
+                :: Nil)
+        }
     }
-  }
+    val supervisor = factory.newSupervisor
-  val supervisor = factory.newSupervisor
+    supervisor.startSupervisor
-  supervisor.startSupervisor
 }
 
 /**
@@ -29,45 +34,79 @@ class Boot {
  * curl http://localhost:9998/scalacount
  * </pre>
  * Or browse to the URL from a web browser.
- */                         
+ */
 @Path("/scalacount")
 class SimpleService extends Actor {
-  uuid = "SimpleService"
+    uuid = "SimpleService"
-  makeTransactionRequired
+    makeTransactionRequired
 
-  case object Tick
+    case object Tick
-  private val KEY = "COUNTER";
+    private val KEY = "COUNTER";
-  private var hasStartedTicking = false;
+    private var hasStartedTicking = false;
-  private val storage = TransactionalState.newPersistentMap(CassandraStorageConfig())
+    private val storage = TransactionalState.newPersistentMap(CassandraStorageConfig())
 
-  @GET
+    @GET
-  @Produces(Array("text/html"))
+    @Produces(Array("text/html"))
-  def count(@DefaultValue("unknown") @QueryParam("who") who : String) = {
+    def count() = {
-      log.info(who)
+        (this !! Tick).getOrElse(<error>Error in counter</error>)
-
-    (this !! Tick).getOrElse(view(<error>Error in counter</error>))
-  }
-
-  override def receive: PartialFunction[Any, Unit] = {
-    case Tick => if (hasStartedTicking) {
-      val counter = storage.get(KEY).get.asInstanceOf[Integer].intValue
-      storage.put(KEY, new Integer(counter + 1))
-      reply(view(<success>Tick: {counter + 1}</success>))
-    } else {
-      storage.put(KEY, new Integer(0))
-      hasStartedTicking = true
-      reply(view(<lift:success>Tick: 0</lift:success>))
     }
-  }
-  
-  override protected def postRestart(reason: AnyRef, config: Option[AnyRef]) = {
-    println("Restarting due to: " + reason.asInstanceOf[Exception].getMessage)
-  }
 
-  def view(data : NodeSeq) : NodeSeq = <lift:surround with="default" at="content"> { data } <lift:Howdy.greet/></lift:surround>
+    override def receive: PartialFunction[Any, Unit] = {
+        case Tick => if (hasStartedTicking) {
+                val counter = storage.get(KEY).get.asInstanceOf[Integer].intValue
+                storage.put(KEY, new Integer(counter + 1))
+                reply(<success>Tick: {counter + 1}</success>)
+            } else {
+                storage.put(KEY, new Integer(0))
+                hasStartedTicking = true
+                reply(<success>Tick: 0</success>)
+            }
+    }
+
+    override protected def postRestart(reason: AnyRef, config: Option[AnyRef]) = {
+        println("Restarting due to: " + reason.asInstanceOf[Exception].getMessage)
+    }
 }
 
-class Howdy
+@Path("/chat")
-{
+class Chat extends Actor with Logging{
-    def greet = <h2>Hello mommy</h2>
+    uuid = "Chat"
-}
+    makeTransactionRequired
+
+    case class Chat(val who : String, val what : String,val msg : String)
+    case object Suspend
+
+    private var hasStarted = false;
+    private val storage = TransactionalState.newPersistentMap(CassandraStorageConfig())
+
+    override protected def postRestart(reason: AnyRef, config: Option[AnyRef]) = {
+        println("Restarting due to: " + reason.asInstanceOf[Exception].getMessage)
+    }
+
+    @Suspend
+    @GET
+    @Produces(Array("text/html"))
+    def suspend() = "<!-- Comet is a programming technique that enables web " +
+    "servers to send data to the client without having any need " +
+    "for the client to request it. -->\n"
+
+    override def receive: PartialFunction[Any, Unit] = {
+        case Chat(who,what,msg) => {
+
+             log.info("Chat(" + who + ", " + what + ", " + msg + ")")
+             
+             what match {
+                 case "login" => reply(<h3>System Message: {who} has joined.</h3>)
+                 case "post"  => reply(<p>{who} says: {msg}</p>)
+                 case _       => throw new WebApplicationException(422)
+             }
+        }
+       }
+
+        @Broadcast
+        @Consumes(Array("application/x-www-form-urlencoded"))
+        @POST
+        @Produces(Array("text/html"))
+        //@BroadcastFilter(Array(classOf[XSSHtmlFilter]))//,classOf[JsonpFilter]))
+        def publishMessage(form: MultivaluedMap[String, String]) = (this !! Chat(form.getFirst("name"),form.getFirst("action"),form.getFirst("message"))).getOrElse(<p>Error</p>)
+    }