streaming hessian/comet servlet

Resin's Comet servlet API enables streaming communication that makes pushing data to the rich clients possible. The API encapsulates of the threading and communications issues between the request threads and the rest of the application. Using Hessian as the communication protocol, we show how to push data to a Flash RIA.

Demo

Resin's Comet API lets server applications push new data to the client as it becomes available. Flash clients can be used to display continually updated information to the user.

The architecture in the picture uses two HTTP streams to the server application, one for normal client-server requests, and a second unidirectional stream to send updates from the server to the client.

The example updates the Flash application every two seconds with new data. In this case, lines of a Latin text.

The components of the Comet/AJAX application look like:

• Protocol: Hessian objects.
• Client:
  • View: Flash updated by Hessian.
  • Controller: call server with a <mx:HessianStreamingServer>
• Server:
  • Service: TimerService manages the comet connections and wakes them with new data.
  • Servlet: TestCometServlet generates Hessian streaming packets from new data from the TimerService on each resume.
  • State: CometState encapsulates both the item's state (the current text), and the CometController needed to wake the servlet and pass updated data.

The comet HTTP stream is a sequence of Hessian streaming packets. Because the Flex implementation of Hessian is able to read data progressively, the user will see the updates immediately without waiting for the entire HTTP request to complete.

The Flash client receives the objects from the server and executes a method on those objects to get the next line in the text and update them in a List control. The client connects to the service using the HessianStreamingService tag. This tag requires a destination attribute which specifies the URL of the service and a responder attribute which handles data received from the server. In this case, our service is at "comet", which is a relative URL to the location of the Flash SWF file. The responder is the application itself. The responder must implement the mx.rpc.IResponder interface. In our example, we also specify the policyPort attribute. We'll discuss this later in the section about sandboxing.

When new data arrives from the service, the responder's result() method is called with the data. In our example, the data is an AddString object on which we call the "execute()" method. This method adds a line to the client state and the state is then reflected in the List control.

<?xml version="1.0"?>
<mx:Application 
  xmlns:mx="http://www.adobe.com/2006/mxml" 
  implements="mx.rpc.IResponder">

  <hessian:HessianStreamingService xmlns:hessian="hessian.mxml.*" 
    destination="comet" responder="{this}" policyPort="80"/>

  <mx:Panel 
    title="Caucho Hessian Comet Demo"
    paddingTop="10" paddingBottom="10" paddingLeft="10" paddingRight="10">

    <mx:Script>
    <![CDATA[
      [Bindable]
      private var _state:ClientState = new ClientState();

      public function result(data:Object):void
      {
        var updater:Updater = Updater(data);

        updater.update(_state);
      }

      public function fault(info:Object):void
      {
      }
    ]]>
    </mx:Script>

    <mx:Label text="The following messages have been received from the server:"/>
    <mx:List dataProvider="{_state}" width="100%"/>

  </mx:Panel>    
</mx:Application>

package example {
  public interface Updater {
    function update(state:ClientState):void;
  }
}

package example {
  public class AddString implements Updater {
    private var _value:String;

    public function get value():String
    {
      return _value;
    }

    public function set value(st:String):void
    {
      _value = st;
    }

    public function update(state:ClientState):void
    {
      state.update(value);
    }
  }
}

package example {
  import mx.collections.ArrayCollection;

  public class ClientState extends ArrayCollection {
    private var _lines:Array = new Array();

    public function update(value:String):void
    {
      if (value == null)
        return;

      if (lines.length > 6)
        lines.shift();

      lines.push(value);

      source = lines;
    }

    public function get lines():Array
    {
      return _lines;
    }
  }
}

<protocol port="80" type="com.caucho.protocols.flash.SocketPolicyProtocol">
  <init>
    <socket-policy-file>/<path-to>/socketpolicy.xml</socket-policy-file>
  </init>
</protocol>

<cross-domain-policy>
  <allow-access-from domain="hessian.caucho.com" to-ports="80"/>
</cross-domain-policy>

Remember when we specified the policyPort on the HessianStreamingService tag in the CometClient.mxml file above? That port corresponds to the port of the socket policy server we configured. Hessian/Flash takes care of contacting the server when this port is specified.

The CometController is Resin's thread-safe encapsulation of control and communication from the application's service to the Comet servlet. Applications may safely pass the CometController to different threads, wake the servlet with wake(), and send data with setAttribute.

In the example, the TimerService passes the next line of text to the servlet by calling setAttribute("caucho.text", text), and wakes the servlet by calling wake(). When the servlet resumes, it will retrieve the line using request.getAttribute("caucho.text"). Note, applications must only use the thread-safe CometController in other threads. As with other servlets, the ServletRequest, ServletResponse, writers and output stream can only be used by the servlet thread itself, never by any other threads.

package com.caucho.servlet.comet;

public interface CometController
{
  public boolean wake();

  public Object getAttribute(String name);
  public void setAttribute(String name, Object value);
  public void removeAttribute(String name);

  public void close();
}

The comet servlet has three major responsibilities:

1. Process the initial request (service).
2. Register the CometController with the service (service).
3. Send streaming data as it becomes available (resume).

Like other servlets, only the comet servlet may use the ServletRequest, ServletResponse or any output writer or stream. No other thread may use these servlet objects, and the application must never store these objects in fields or objects accessible by other threads. Even in a comet servlet, the servlet objects are not thread-safe. Other services and threads must use the CometController to communicate with the servlet.

Process the initial request: our servlet just calls setContentType("text/html"), since it's a trivial example. A real application would do necessary database lookups and possibly send more complicated data to the client.

Register the CometController: our servlet registers the controller with the timer service by calling addCometState. In general, the application state object will contain the CometController as part of the registration process.

Send streaming data:. The TimerService will set new data in the "comet.text" attribute and wake() the controller. When the servlet executes the resume() method, it will retrieve the data, and send the next packet to the client.

package example;

import java.io.*;
import java.util.*;
import java.util.concurrent.*;

import javax.annotation.*;
import javax.servlet.*;
import javax.servlet.http.*;
import javax.webbeans.In;

import com.caucho.hessian.io.*;
import com.caucho.servlet.comet.*;

public class HessianCometServlet extends GenericCometServlet
{
  @In private TimerService _timerService;
  
  private ArrayList<CometState> _itemList
    = new ArrayList<CometState>();
  
  @Override
  public boolean service(ServletRequest request,
                         ServletResponse response,
                         CometController controller)
    throws IOException, ServletException
  {
    HttpServletRequest req = (HttpServletRequest) request;
    HttpServletResponse res = (HttpServletResponse) response;

    Hessian2StreamingOutput out = 
      new Hessian2StreamingOutput(res.getOutputStream());

    CometState state = new CometState(controller);
    
    // Add the comet state to the controller
    _timerService.addCometState(state);

    return true;
  }
  
  @Override
  public boolean resume(ServletRequest request,
                        ServletResponse response,
                        CometController controller)
    throws IOException, ServletException
  {
    HttpServletRequest req = (HttpServletRequest) request;
    HttpServletResponse res = (HttpServletResponse) response;
    
    Hessian2StreamingOutput out = 
      new Hessian2StreamingOutput(res.getOutputStream());

    Object command = controller.getAttribute("comet.command");
    out.writeObject(command);

    return true;
  }
}

The connection can close for a number of reasons. Either the service() or resume() methods may return false, telling Resin to close the connection. The service might call CometController.close() which will also close the connection. Finally, the client may close the connection itself.

The sequence of calls for the example looks like the following:

1. servlet.service() is called for the initial request
2. _service.addCometState() registers with the TimerService
3. after the service() completes, Resin suspends the servlet.
4. The TimerService detects an event, in this case the timer event.
5. The TimerService calls controller.setAttribute() to send new data.
6. The TimerService calls controller.wake() to wake the servlet.
7. servlet.resume() processes the data and sends the next packet.
8. After the resume() completes, Resin suspends the servlet again and we repeat as after step #3.
9. After the 10th data, the TimerService calls controller.close(), closing the servlet connection.

Comet Servlet State Machine

The sequence of comet servlet calls looks like the following state machine. After the initial request, the servlet spends most of its time suspended, waiting for the TimerService to call wake().

Demo