In my previous blog posting, I wrote about what I would expect node.js to do in order to become mature. I introduced the idea of having a framework which lets you define a protocol and some handlers in order to let the developer concentrate on writing useful business software, rather than technical code, in a very similar manner to which Java EE does. And through that posting, I came to learn about a thing called Comet. I had stated that using a non-blocking server wouldn't really be any more useful than a blocking one in a typical web application (HTTP), and so I created an example based on my own protocol and a VOIP server, for streaming binary data to many concurrently connected clients.



I have now read up on Comet and come to realise there is indeed a good case for having a non-blocking server in the web. That case is pushing data back to the client, like for continuously publishing latest stock prices. While this example could be solved using polling, true Comet uses long-polling or even better, full on push. A great introduction I read was here. The idea is that the client makes a call to the server and instead of the server returning data immediately, it keeps the connection open and returns data at some time in the future, potentially many times. This is not a new idea - the term Comet seems to have been invented in about 2006 and the article I refer to above is from 2009. I think I've arrived at this party very late :-)



My new found case of server push for non-blocking HTTP, and a strong curiosity drove me to knuckle down and start coding. Before long, I had a rough implementation of the HTTP protocol for my little framework, and I was able to write an app for my server using handlers that subclassed the HttpHandler , which was for all intents and purposes, a Servlet.



To get Comet push to work properly, you get the client to "log in" and register itself with the server. In my demo, I didn't check authorisations against a database, like you might do for a real app, but I had the concept of a channel, to which any browser client could subscribe. During this login, the client says which channel it wants to subscribe to, and the server adds the clients non-blocking connection to its model. The server responds using chunked transfer encoding, because that way, the connection stays open, and you don't need to state up front how much data you will send back. At some time in the future, when someone publishes something, the server can use the connection which is still open contained in its model, to push that published data back to the subscribed client, by sending another chunk of data.



The server implementation wasn't too hard, but the client posed a few problems, until I realised that the data was arriving in the ajax client with a ready state of 3, rather than the more usual 4. The ajax client's onreadystatechange callback function was also given every byte of data in it's responseText , rather than just the new stuff, so I had some fiddling around until I could get the browser to just append the new stuff to the innerHTML attribute of a div on my page. Anyway, after just a few hours, I had an app that worked quite well. But it wasn't entirely satisfactory, partly because as I stated in the previous posting, the server is still a little buggy, especially when the client drops a connection, because for example the browser page is closed. I had also ended up implementing the HTTP protocol for my framework, which seemed to be reinventing the wheel - servlet technology does all this stuff already, and much better than I can hope to do it. One of the reasons that I don't like node.js is that everything is being reinvented.



So, like the article which I referenced above indicated, version 3.0 of Servlets should be able to handle Comet. I downloaded Tomcat 7.0, which has a Servlet 3.0 container, and I ported my app code to proper Servlets. It took a while to work out exactly how to use the new asynchronous parts of servlets because there aren't all that many accurate tutorials out there. The servlet specs (JSR 315) helped a lot. Once I cracked how to use the async stuff properly, I had a really really satisfying solution to my push requirements.



The first step, was to reconfigure Tomcat so that it uses non-blocking (NIO) for its connector protocol. The point of this is that I want to keep a connection open to the client in order to push data to it. I can't rely on the one-thread-per-request paradigm, because context switching and thread memory requirements are likely to be a performance killer. In Tomcat's server.xml file, I configured the connector's protocol:

<!-- NIO HTTP/1.1 connector --> <Connector port="8080" protocol="org.apache.coyote.http11.Http11NioProtocol" connectionTimeout="20000" redirectPort="8443" />



rather than the normal:

<Connector port="8080" protocol="HTTP/1.1" connectionTimeout="20000" redirectPort="8443" />



All you need to do to get Tomcat to turn into an NIO server is change the protocol attribute to the slightly longer class name.



The second step, was to create two servlets. The first LoginServlet handles the client "logging in" and subscribing to a channel. That servlet looks like this:

/* * Copyright (c) 2011 Ant Kutschera * * This file is part of Ant Kutschera's blog, * http://blog.maxant.co.uk * * This is free software: you can redistribute * it and/or modify it under the terms of the * Lesser GNU General Public License as published by * the Free Software Foundation, either version 3 of * the License, or (at your option) any later version. * * It is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR * PURPOSE. See the Lesser GNU General Public License for * more details. * * You should have received a copy of the * Lesser GNU General Public License along with this software. * If not, see http://www.gnu.org/licenses/. */ package ch.maxant.blog.nio.servlet3; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; import javax.servlet.AsyncContext; import javax.servlet.ServletContext; import javax.servlet.annotation.WebServlet; import javax.servlet.http.HttpServlet; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; import ch.maxant.blog.nio.servlet3.model.Subscriber; @WebServlet(name = "loginServlet", urlPatterns = { "/login" }, asyncSupported = true) public class LoginServlet extends HttpServlet { public static final String CLIENTS = "ch.maxant.blog.nio.servlet3.clients"; private static final long serialVersionUID = 1L; public void doGet(HttpServletRequest request, HttpServletResponse response) throws IOException { // dont set the content length in the response, and we will end up with chunked // encoding so that a) we can keep the connection open to the client, and b) send // updates to the client as chunks. // ********************* // we use asyncSupported=true on the annotation for two reasons. first of all, // it means the connection to the client isn't closed by the container. second // it means that we can pass the asyncContext to another thread (eg the publisher) // which can then send data back to that open connection. // so that we dont require a thread per client, we also use NIO, configured in the // connector of our app server (eg tomcat) // ********************* // what channel does the user want to subscribe to? // for production we would need to check authorisations here! String channel = request.getParameter("channel"); // ok, get an async context which we can pass to another thread final AsyncContext aCtx = request.startAsync(request, response); // a little longer than default, to give us time to test. // TODO if we use a heartbeat, then time that to pulse at a similar rate aCtx.setTimeout(20000L); // create a data object for this new subscription Subscriber subscriber = new Subscriber(aCtx, channel); // get the application scope so that we can add our data to the model ServletContext appScope = request.getServletContext(); // fetch the model from the app scope @SuppressWarnings("unchecked") Map



The inline comments describe most of the choices I made. I added a listener to the context so that we get an event in the case of a disconnected client, so that we can tidy up our model - the full code is in the ZIP at the end of this article. Importantly, this servlet does no async processing itself. It simply prepares the request and response objects for access at some time in the future. We stick them (via the async context) into a model which is in application scope. That model can then be used by the servlet which receives a request to publish something to a given channel:

/* * Copyright (c) 2011 Ant Kutschera * * This file is part of Ant Kutschera's blog, * http://blog.maxant.co.uk * * This is free software: you can redistribute * it and/or modify it under the terms of the * Lesser GNU General Public License as published by * the Free Software Foundation, either version 3 of * the License, or (at your option) any later version. * * It is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR * PURPOSE. See the Lesser GNU General Public License for * more details. * * You should have received a copy of the * Lesser GNU General Public License along with this software. * If not, see http://www.gnu.org/licenses/. */ package ch.maxant.blog.nio.servlet3; import java.io.IOException; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.Map; import javax.servlet.AsyncContext; import javax.servlet.ServletContext; import javax.servlet.annotation.WebServlet; import javax.servlet.http.HttpServlet; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; import ch.maxant.blog.nio.servlet3.model.Subscriber; @WebServlet(name = "publishServlet", urlPatterns = { "/publish" }, asyncSupported = true) public class PublishServlet extends HttpServlet { private static final long serialVersionUID = 1L; public void doGet(HttpServletRequest request, HttpServletResponse response) throws IOException { // ************************* // this servlet simply spawns a thread to send its message to all subscribers. // this servlet keeps the connection to its client open long enough to tell it // that it has published to all subscribers. // ************************* // add a pipe character, so that the client knows from where the newest model has started. // if messages are published really quick, its possible that the client gets two at // once, and we dont want it to be confused! these messages also arrive at the // ajax client in readyState 3, where the responseText contains everything since login, // rather than just the latest chunk. so, the client needs a way to work out the // latest part of the message, containing the newest version of the model it should // work with. might be better to return XML or JSON here! final String msg = "|" + request.getParameter("message") + " " + new Date(); // to which channel should it publish? in prod, we would check authorisations here too! final String channel = request.getParameter("channel"); // get the application scoped model, and copy the list of subscribers, so that the // long running task of publishing doesnt interfere with new logins ServletContext appScope = request.getServletContext(); @SuppressWarnings("unchecked") final Map



Again, there are plenty of comments in the code. In this servlet, we actually do some (potentially) long running task. In many online examples of Servlet 3.0 Async Support, they show handing off work to an Executor. The async context provides the ideal way to do this via the container though, using its start(Runnable) method. The container implementation then descides how to handle the task, rather than the developer having to worry about spawning threads, which on app servers like WebSphere is illegal and leads to errors. In true Java EE fashion, the developer can concentrate on business code, rather than technicalities.



Something else which might be important in the above code, is that the publishing is done on a different thread. Imagine publishing data to ten thousand clients. In order to finish within a second, each push it going to have to complete in less than a tenth of a millisecond. That means doing something useful like a database lookup is not going to be feasible. On a non-blocking server, we can't afford to take a second to do something, and many would argue a second is eternity in such an environment. The ability to hand off such work to a different thread is invaluable, and sadly, something node.js cannot currently do, although you can hand off the task to a different process, albeit potentially messier than that shown here.



Now, we just need to create a client to subscribe to the server. This client is an ajax request object which is created when the HTML is loaded which runs some JavaScript in a library I have written. The HTML looks like this:

<html> <head> <script language="Javascript" type="text/javascript" src="push_client.js"></script> </head> <body> <p>Boo!</p> <div id="myDiv"></div> </body> <script language="Javascript" type="text/javascript"> function callback(model){ //simply append the model to a div, for demo purposes var myDiv = document.getElementById("myDiv"); myDiv.innerHTML = myDiv.innerHTML + "<br>" + model; } new PushClient("myChannel", callback).login(); </script> </html>



As you can see, it simply needs to define a callback function which will handle each message published from the server. The published message could be text, XML or JSON - the publisher chooses. The JavaScript in that library is a little more complicated, but basically creates an XHR requester which sends a request to the login servlet. Any data it receives from the server, it parses, and returns the newest part of the data back to the callback.

/* * Copyright (c) 2011 Ant Kutschera * * This file is part of Ant Kutschera's blog, * http://blog.maxant.co.uk * * This is free software: you can redistribute * it and/or modify it under the terms of the * Lesser GNU General Public License as published by * the Free Software Foundation, either version 3 of * the License, or (at your option) any later version. * * It is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR * PURPOSE. See the Lesser GNU General Public License for * more details. * * You should have received a copy of the * Lesser GNU General Public License along with this software. * If not, see http://www.gnu.org/licenses/. */ function PushClient(ch, m){ this.channel = ch; this.ajax = getAjaxClient(); this.onMessage = m; // stick a reference to "this" into the ajax client, so that the handleMessage // function can access the push client - its "this" is an XMLHttpRequest object // rather than the push client, coz thats how javascript works! this.ajax.pushClient = this; function getAjaxClient(){ /* * Gets the ajax client * http://en.wikipedia.org/wiki/XMLHttpRequest * http://www.w3.org/TR/XMLHttpRequest/#responsetext */ var client = null; try{ // Firefox, Opera 8.0+, Safari client = new XMLHttpRequest(); }catch (e){ // Internet Explorer try{ client = new ActiveXObject("Msxml2.XMLHTTP"); }catch (e){ client = new ActiveXObject("Microsoft.XMLHTTP"); } } return client; }; /** * pass in a callback and a channel. * the callback should take a string, * which is the latest version of the model */ PushClient.prototype.login = function(){ try{ var params = escape("channel") + "=" + escape(this.channel); var url = "login?" + params; this.ajax.onreadystatechange = handleMessage; this.ajax.open("GET",url,true); //true means async, which is the safest way to do it // hint to the browser and server, that we are doing something long running // initial tests only seemed to work with this - dont know, perhaps now it // works without it? this.ajax.setRequestHeader("Connection", "Keep-Alive"); this.ajax.setRequestHeader("Keep-Alive", "timeout=999, max=99"); this.ajax.setRequestHeader("Transfer-Encoding", "chunked"); //send the GET request to the server this.ajax.send(null); }catch(e){ alert(e); } }; function handleMessage() { //states are: // 0 (Uninitialized) The object has been created, but not initialized (the open method has not been called). // 1 (Open) The object has been created, but the send method has not been called. // 2 (Sent) The send method has been called. responseText is not available. responseBody is not available. // 3 (Receiving) Some data has been received. responseText is not available. responseBody is not available. // 4 (Loaded) try{ if(this.readyState == 0){ //this.pushClient.onMessage("0/-/-"); }else if (this.readyState == 1){ //this.pushClient.onMessage("1/-/-"); }else if (this.readyState == 2){ //this.pushClient.onMessage("2/-/-"); }else if (this.readyState == 3){ //for chunked encoding, we get the newest version of the entire response here, //rather than in readyState 4, which is more usual. if (this.status == 200){ this.pushClient.onMessage("3/200/" + this.responseText.substring(this.responseText.lastIndexOf("|"))); }else{ this.pushClient.onMessage("3/" + this.status + "/-"); } }else if (this.readyState == 4){ if (this.status == 200){ //the connection is now closed. this.pushClient.onMessage("4/200/" + this.responseText.substring(this.responseText.lastIndexOf("|"))); //start again - we were just disconnected! this.pushClient.login(); }else{ this.pushClient.onMessage("4/" + this.status + "/-"); } } }catch(e){ alert(e); } }; }



The important parts here are that we need to listen for events on both ready state 3 and 4, rather than the usual 4. While the connection is kept open, the client only receives data in ready state 3, and everytime it receives a chunk, the ajax.responseText attribute contains all chunks since login, rather than just the newest chunk. This could be bad, if the connection receives tons of data - eventually the browser will run out of memory! You could measure the number of bytes sent to any client on the server, and when its above a given threshold, force the client to disconnect by ending the stream (call the complete() method on the async context of the relevant client just after publishing a message to it). The client above automatically logs itself back into the server when the server disconnects.



Instead of the reconnect solution for dropped / timed-out connections (which is very similar to long polling) we could add a heartbeat which the server sends to each subscribed client. The heartbeat period would need to be slightly less than the timeout. The exact details could get messy - do you consider sending a heartbeat every second, but only do it to clients who need it? Or do you send it to every client, say every 25 seconds, if the timeout is say 30 seconds? You could use performance tuning to determine if that is better than the reconnecting I have shown above. Then again, a heartbeat is good for culling closed connections, because it tests the connection every so often, and gets an exception if the push fails. And then again, the container informs the listener that we added to the login async context if a disconnect occurs too, so perhaps we don't need a heartbeat - you decide :-)



Now, we need a way to publish data - that's easy - I just type the following URL into the browser, and it sends a GET request to the publish servlet:

http://localhost:8080/nio-servlet3/publish?channel=myChannel&message=javaIsAwesome



Keep refreshing the browser window with that protocol, and the other window almost instantly gets updates, showing the latest message at the bottom. I tested it using Firefox as the subscriber, and Chrome as the publisher.



I haven't checked out scalability, because I have assumed that Tomcat's NIO connector has been well tested and performs well. I'll let someone else play with the scalability of Servlet 3.0 for a push solution. This article is about showing how easy it is to implement Comet push, using Java EE Servlets. Note, it used to be easy too, because servers like Jetty and Tomcat and others provided bespoke Comet interfaces, but now, with the advent of Servlet 3.0, there is a standardised way to do it.



There are plenty of profesional and open source solutions which do what I have done in this article. See this article, which lists many, whackiest of all, APE - another project which puts JavaScript on the server!? Well, better than PHP I guess :-)



The complete code for this demo is downloadable here.



Copyright © 2011 Ant Kutschera