Hitchiker Guide to the Atmosphere Framework using WebSocket, Long-Polling and Http Streaming

The Atmosphere Framework easily allow the writing of web application that support, transparently, SSE (Server Side Events), JSONP, WebSocket, Long-Polling and Http Streaming. The Atmosphere Framework also hide the complexity of the current asynchronous API, which differ from Server to Server and make your application portable among them. More important, it is much more easy to write an Atmosphere application than using the Servlet 3.0 API.

There are several APIs available in Atmosphere to write an asynchronous application: AtmosphereHandler, Meteor or using Jersey‘s Atmosphere extension. In this blog I will take the famous JQuery PubSub sample to demonstrate those APIs. Note that I will not discuss the JQuery Atmosphere Plugin as it is the same for all APIs. Important, all code snippet below support WebSocket, SSE, JSONP, Long-Polling and Streaming by default. Only the last section only support WebSocket.

The JQuery PubSub Application is quite simple. You enter a topic to subscribe, you select a transport to use (WebSocket, JSONP, SSE, Streaming or Long-Polling) or let the plug in decide for you, and then you are ready to publish message. You can use Redis on the server side to cluster your application among servers. The subscribe operation is done using a GET, the publish using a POST (in the form of message=”something”). If the WebSocket transport is used, the message is wrapped as a POST and delivered as a normal HTTP request. Note that this feature is configurable in Atmosphere

PubSub using AtmosphereHandler

The AtmosphereHandler  is a low level API that can be used to write an asynchronous application. An application just have to implement that interface. This API is usually used by other framework in order to integrate with Atmosphere (GWT, Jersey, Vaading, etc.) but it can also be used if you want to write Servlet style code.  So, with an AtmosphereHandler, the PubSub implementation will take the form of:

public class AtmosphereHandlerPubSub
      extends AbstractReflectorAtmosphereHandler {

    @Override
    public void onRequest
       (AtmosphereResource r)
          throws IOException {

        HttpServletRequest req = r.getRequest();
        HttpServletResponse res = r.getResponse();
        String method = req.getMethod();

        // Suspend the response.
        if ("GET".equalsIgnoreCase(method)) {
            String trackingId = trackingId(req);

            // Log all events on the console, including WebSocket events.
            r.addEventListener(new WebSocketEventListenerAdapter());

            res.setContentType("text/html;charset=ISO-8859-1");

            Broadcaster b = lookupBroadcaster(req.getPathInfo());
            r.setBroadcaster(b);

            if (req.getHeader(X_ATMOSPHERE_TRANSPORT)
                    .equalsIgnoreCase(LONG_POLLING_TRANSPORT)) {
                req.setAttribute(RESUME_ON_BROADCAST, Boolean.TRUE);
                r.suspend(-1, false);
            } else {
                r.suspend(-1);
            }
        } else if ("POST".equalsIgnoreCase(method)) {
            Broadcaster b = lookupBroadcaster(req.getPathInfo());

            String message = req.getReader().readLine();

            if (message != null && message.indexOf("message") != -1) {
                b.broadcast(message.substring("message=".length()));
            }
        }
    }

    @Override
    public void destroy() {
    }

    Broadcaster lookupBroadcaster(String pathInfo) {
        String[] decodedPath = pathInfo.split("/");
        Broadcaster b = BroadcasterFactory.getDefault()
              .lookup(decodedPath[decodedPath.length - 1], true);
        return b;
    }

}

When a GET is received, we lookup a Broadcaster and then suspend the response based on the path info (REST style). Here we need a make sure we aren’t sending padding data (required for WebKit browser) when long polling is used. That’s the only required conditional evaluation needed in terms of transport. With the POST we just look up the Broadcaster (which represent a pubsub topic) and broadcast the request’s body. That;s it.

PubSub using Meteor

The Meteor is another low level API that can be used with existing Servlet application. As an example, the ADF framework use Meteor in order to integrate Atmosphere support.

public class MeteorPubSub extends HttpServlet {

    @Override
    public void doGet(HttpServletRequest req, HttpServletResponse res)
       throws IOException {
        // Create a Meteor
        Meteor m = Meteor.build(req);

        // Log all events on the console, including WebSocket events.
        m.addListener(new WebSocketEventListenerAdapter());

        res.setContentType("text/html;charset=ISO-8859-1");

        Broadcaster b = lookupBroadcaster(req.getPathInfo());
        m.setBroadcaster(b);

        if (req.getHeader(X_ATMOSPHERE_TRANSPORT)
                .equalsIgnoreCase(LONG_POLLING_TRANSPORT)) {
            req.setAttribute(RESUME_ON_BROADCAST, Boolean.TRUE);
            m.suspend(-1, false);
        } else {
            m.suspend(-1);
        }
    }

    public void doPost(HttpServletRequest req, HttpServletResponse res)
        throws IOException {

        Broadcaster b = lookupBroadcaster(req.getPathInfo());
        String message = req.getReader().readLine();

        if (message != null && message.indexOf("message") != -1) {
            b.broadcast(message.substring("message=".length()));
        }
    }

    Broadcaster lookupBroadcaster(String pathInfo) {
        String[] decodedPath = pathInfo.split("/");
        Broadcaster b = BroadcasterFactory.getDefault()
              .lookup(decodedPath[decodedPath.length - 1], true);
        return b;
    }
}

When a GET is received, we create a Meteor and use that Meteor to suspend the response, again using the path info. For post, we do the same as with AtmosphereHandler, e.g retrieve the Broadcaster and broadcast the message.

PubSub using Jersey’s Atmosphere Extension

With the Jersey extension, we can either use annotations or the programmatic API. As simple as”

@Path("/pubsub/{topic}")
@Produces("text/html;charset=ISO-8859-1")
public class JQueryPubSub {

    private
    @PathParam("topic")
    Broadcaster topic;

    @GET
    public SuspendResponse subscribe() {
        return new SuspendResponse.SuspendResponseBuilder()
                .broadcaster(topic)
                .outputComments(true)
                .addListener(new EventsLogger())
                .build();
    }

    @POST
    @Broadcast
    public Broadcastable publish(@FormParam("message") String message) {
        return new Broadcastable(message, "", topic);
    }
}

The GET could have been handled using the @Suspend annotation:

    @GET
    @Suspend(listeners = EventsLogger.class, outputComments = true)
    public Broadcastable subscribe(){
        return new Broadcastable(topic);
    }

As you can see, it is quite simpler that with Meteor and AtmosphereHandler.
PubSub.

PubSub using WebSocket only

If you are planning to write pure WebSocket application and don’t plan to support normal HTTP, you can also write your own WebSocket sub protocol. It is quote important to note here that only WebSocket will be supported.

public class WebSocketPubSub implements WebSocketProtocol {

    private AtmosphereResource r;

    @Override
    public HttpServletRequest onMessage(WebSocket webSocket, String message) {
        Broadcaster b = lookupBroadcaster(r.getRequest().getPathInfo());

        if (message != null && message.indexOf("message") != -1) {
            b.broadcast(message.substring("message=".length()));
        }

        //Do not dispatch to another Container like Jersey
        return null;
    }

    @Override
    public void onOpen(WebSocket webSocket) {
        // Accept the handshake by suspending the response.
        r = (AtmosphereResource)
              webSocket.atmosphereResource();

        Broadcaster b = lookupBroadcaster(r.getRequest().getPathInfo());
        r.setBroadcaster(b);
        r.addEventListener(new WebSocketEventListenerAdapter());

        r.suspend(-1);
    }

    @Override
    public void onClose(WebSocket webSocket) {
        webSocket.atmosphereResource().resume();
    }

    Broadcaster lookupBroadcaster(String pathInfo) {
        String[] decodedPath = pathInfo.split("/");
        Broadcaster b = BroadcasterFactory.getDefault().
                 lookup(decodedPath[decodedPath.length - 1], true);
        return b;
    }

The important method here is onOpen (for accepting the handshake) and the onMessage, which is were the messages are received.

Conclusion

It is quite important to pick the best API when writing Atmosphere application as it can save you a lot of time! You can download all samples from here.

For any questions or to download Atmosphere Client and Server Framework, go to our main site, use our Google Group forum, follow the team or myself and tweet your questions there! .

About these ads