Archive

Archive for November 6, 2009

Servlet 3.0 Asynchronous API or Atmosphere? Easy decision!

November 6, 2009 14 comments

One the comment I’m getting about Atmosphere is why should I use the framework instead of waiting for Servlet 3.0 Async API. Well, it simple: much simpler, works with any existing Java WebServer (including Google App Engine!), and will auto-detect the Servlet 3.0 Async API if you deploy your application on a WebServer that support it.

IMG_0159.JPG

To make a fair comparison, let’s write the hello world of Comet, a Chat application and compare the server side code. Without technical details, let’s just drop the entire server code. First, the Servlet 3.0 version (can probably be optimized a little):

  1 package web.servlet.async_request_war;
  2 
  3 import java.io.IOException;
  4 import java.io.PrintWriter;
  5 import java.util.Queue;
  6 import java.util.concurrent.ConcurrentLinkedQueue;
  7 import java.util.concurrent.BlockingQueue;
  8 import java.util.concurrent.LinkedBlockingQueue;
  9 
 10 import javax.servlet.AsyncContext;
 11 import javax.servlet.AsyncEvent;
 12 import javax.servlet.AsyncListener;
 13 import javax.servlet.ServletConfig;
 14 import javax.servlet.ServletException;
 15 import javax.servlet.annotation.WebServlet;
 16 import javax.servlet.http.HttpServlet;
 17 import javax.servlet.http.HttpServletRequest;
 18 import javax.servlet.http.HttpServletResponse;
 19 
 20 @WebServlet(urlPatterns = {"/chat"}, asyncSupported = true)
 21 public class AjaxCometServlet extends HttpServlet {
 22 
 23     private static final Queue<AsyncContext> queue = new ConcurrentLinkedQueue<AsyncContext>();
 24     private static final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<String>();
 25     private static final String BEGIN_SCRIPT_TAG = "<script type='text/javascript'>\n";
 26     private static final String END_SCRIPT_TAG = "</script>\n";
 27     private static final long serialVersionUID = -2919167206889576860L;
 28 
 29     private Thread notifierThread = null;
 30 
 31     @Override
 32     public void init(ServletConfig config) throws ServletException {
 33         Runnable notifierRunnable = new Runnable() {
 34             public void run() {
 35                 boolean done = false;
 36                 while (!done) {
 37                     String cMessage = null;
 38                     try {
 39                         cMessage = messageQueue.take();
 40                         for (AsyncContext ac : queue) {
 41                             try {
 42                                 PrintWriter acWriter = ac.getResponse().getWriter();
 43                                 acWriter.println(cMessage);
 44                                 acWriter.flush();
 45                             } catch(IOException ex) {
 46                                 System.out.println(ex);
 47                                 queue.remove(ac);
 48                             }
 49                         }
 50                     } catch(InterruptedException iex) {
 51                         done = true;
 52                         System.out.println(iex);
 53                     }
 54                 }
 55             }
 56         };
 57         notifierThread = new Thread(notifierRunnable);
 58         notifierThread.start();
 59     }
 60 
 61     @Override
 62     protected void doGet(HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException {
 63         res.setContentType("text/html");
 64         res.setHeader("Cache-Control", "private");
 65         res.setHeader("Pragma", "no-cache");
 66         
 67         PrintWriter writer = res.getWriter();
 68         // for IE
 69         writer.println("<!-- 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");
 70         writer.flush();
 71 
 72         req.setAsyncTimeout(10 * 60 * 1000);
 73         final AsyncContext ac = req.startAsync();
 74         queue.add(ac);
 75         req.addAsyncListener(new AsyncListener() {
 76             public void onComplete(AsyncEvent event) throws IOException {
 77                 queue.remove(ac);
 78             }
 79 
 80             public void onTimeout(AsyncEvent event) throws IOException {
 81                 queue.remove(ac);
 82             }
 83         });
 84     }
 85 
 86     @Override
 87     @SuppressWarnings("unchecked")
 88     protected void doPost(HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException {
 89         res.setContentType("text/plain");
 90         res.setHeader("Cache-Control", "private");
 91         res.setHeader("Pragma", "no-cache");
 92
 93         req.setCharacterEncoding("UTF-8");
 94         String action = req.getParameter("action");
 95         String name = req.getParameter("name");
 96 
 97         if ("login".equals(action)) {
 98             String cMessage = BEGIN_SCRIPT_TAG + toJsonp("System Message", name + " has joined.") + END_SCRIPT_TAG;
 99             notify(cMessage);
100 
101             res.getWriter().println("success");
102         } else if ("post".equals(action)) {
103             String message = req.getParameter("message");
104             String cMessage = BEGIN_SCRIPT_TAG + toJsonp(name, message) + END_SCRIPT_TAG;
105             notify(cMessage);
106 
107             res.getWriter().println("success");
108         } else {
109             res.sendError(422, "Unprocessable Entity");
110         }
111     }
112 
113     @Override
114     public void destroy() {
115         queue.clear();
116         notifierThread.interrupt();
117     }
118 
119     private void notify(String cMessage) throws IOException {
120         try {
121             messageQueue.put(cMessage);
122         } catch(Exception ex) {
123             throw new IOException(ex);
124         }
125     }
126 
127     private String escape(String orig) {
128         StringBuffer buffer = new StringBuffer(orig.length());
129 
130         for (int i = 0; i < orig.length(); i++) {
131             char c = orig.charAt(i);
132             switch (c) {
133             case '\b':
134                 buffer.append("\\b");
135                 break;
136             case '\f':
137                 buffer.append("\\f");
138                 break;
139             case '\n':
140                 buffer.append("<br />");
141                 break;
142             case '\r':
143                 // ignore
144                 break;
145             case '\t':
146                 buffer.append("\\t");
147                 break;
148             case '\'':
149                 buffer.append("\\'");
150                 break;
151             case '\"':
152                 buffer.append("\\\"");
153                 break;
154             case '\\':
155                 buffer.append("\\\\");
156                 break;
157             case '<':
158                 buffer.append("<");
159                 break;
160             case '>':
161                 buffer.append(">");
162                 break;
163             case '&':
164                 buffer.append("&");
165                 break;
166             default:
167                 buffer.append(c);
168             }
169         }
170 
171         return buffer.toString();
172     }
173 
174     private String toJsonp(String name, String message) {
175         return "window.parent.app.update({ name: \"" + escape(name) + "\", message: \"" + escape(message) + "\" });\n";
176     }
177 }

OK now with Atmosphere , the same code consist of:

  1 package org.atmosphere.samples.chat.resources;
  2 
  3 import javax.ws.rs.Consumes;
  4 import javax.ws.rs.GET;
  5 import javax.ws.rs.POST;
  6 import javax.ws.rs.Path;
  7 import javax.ws.rs.Produces;
  8 import javax.ws.rs.WebApplicationException;
  9 import javax.ws.rs.core.MultivaluedMap;
 10 import org.atmosphere.annotation.Broadcast;
 11 import org.atmosphere.annotation.Schedule;
 12 import org.atmosphere.annotation.Suspend;
 13 import org.atmosphere.util.XSSHtmlFilter;
 14 
 15 @Path("/")
 16 public class ResourceChat {
 17 
 18     @Suspend
 19     @GET
 20     @Produces("text/html;charset=ISO-8859-1")
 21     public String suspend() {
 22         return "";
 23     }
 24 
 25     @Broadcast({XSSHtmlFilter.class, JsonpFilter.class})
 26     @Consumes("application/x-www-form-urlencoded")
 27     @POST
 28     @Produces("text/html;charset=ISO-8859-1")
 29     public String publishMessage(MultivaluedMap form) {
 30         String action = form.getFirst("action");
 31         String name = form.getFirst("name");
 32 
 33         if ("login".equals(action)) {
 34             return ("System Message" + "__" + name + " has joined.");
 35         } else if ("post".equals(action)) {
 36             return name + "__" + form.getFirst("message");
 37         } else {
 38             throw new WebApplicationException(422);
 39         }
 40     }
 41 }

OK so what’s the deal? What’s makes Atmosphere so easy? The Servlet 3.0 new Async API offers:

  • Method to suspend a response, HttpServletRequest.startAsync()
  • Method to resume a response: AsyncContext.complete()

Atmosphere offers:

  • Annotation to suspend: @Suspend
  • Annotation or resume: @Resume
  • Annotation to broadcast (or push) events to the set of suspended responses: @Broadcast
  • Annotation to filter and serialize broadcasted events using BroadcasterFilter (XSSHtmlFilter.class, JsonpFilter.class)
  • Build it support for all browser implementation incompatible implementation (ex: no need to output comments like in the Servlet 3.0 sample (line 69)). Atmosphere will workaround all those issues for you.

With Servlet 3.0 Async API, the missing part is how you share information with suspended responses. In the current chat sample, you need to creates your own Thread/Queue in order to broadcast events to your set of suspended responses (line 32 to 56). This is not a big deal, but you will need to do something like that for all your Servlet 3.0 Async based applications…or use a Framework that do it for you!.

Still not convinced? Well, you can write your Atmosphere applications today and not have to wait for Servlet.3.0 implementation (OK easy plug for my other project: GlassFish v3 supports it pretty well!). Why? Atmosphere always auto-detected the best asynchronous API when you deploy your application. It always try first to look up the 3.0 Async API. If it fails, it will try to find WebServer’s native API like Grizzly Comet (GlassFish), CometProcessor (Tomcat), Continuation (Jetty), HttpEventServlet (JBossWeb), AsyncServlet (WebLogic), Google App Engine (Google). Finally, it will fallback to use a blocking I/O Thread to emulate support for asynchronous events.

But you don’t want to use Java? Fine, try the Atmosphere Grails Plug In, or Atmosphere in PrimesFaces if you like JSF, or use Scala:

  1 package org.atmosphere.samples.scala.chat
  2 
  3 import javax.ws.rs.{GET, POST, Path, Produces, WebApplicationException, Consumes}
  4 import javax.ws.rs.core.MultivaluedMap
  5 import org.atmosphere.annotation.{Broadcast, Suspend}
  6 import org.atmosphere.util.XSSHtmlFilter
  7 
  8 @Path("/chat")
  9 class Chat {
 10 
 11     @Suspend
 12     @GET
 13     @Produces(Array("text/html;charset=ISO-8859-1"))
 14     def suspend() = {
 15         ""
 16     }
 17 
 18     @Broadcast(Array(classOf[XSSHtmlFilter],classOf[JsonpFilter]))
 19     @Consumes(Array("application/x-www-form-urlencoded"))
 20     @POST
 21     @Produces(Array("text/html;charset=ISO-8859-1"))
 22     def publishMessage(form: MultivaluedMap[String, String]) = {
 23         val action = form.getFirst("action")
 24         val name = form.getFirst("name")
 25 
 26         val result: String = if ("login".equals(action)) "System Message" + "__" + name + " has joined."
 27              else if ("post".equals(action)) name + "__" + form.getFirst("message")
 28              else throw new WebApplicationException(422)
 29 
 30         result
 31     }
 32 
 33 
 34 }

Echec et Mat!

Now, I can understand you already have an existing application and just want to update it with suspend/resume/broadcast functionality, without having to re-write it completely. Fine, let’s just use the Atmosphere’s Meteor API:

  1 package org.atmosphere.samples.chat;
  2 
  3 import java.io.IOException;
  4 import java.util.LinkedList;
  5 import java.util.List;
  6 import javax.servlet.http.HttpServlet;
  7 import javax.servlet.http.HttpServletRequest;
  8 import javax.servlet.http.HttpServletResponse;
  9 import org.atmosphere.cpr.BroadcastFilter;
 10 import org.atmosphere.cpr.Meteor;
 11 import org.atmosphere.util.XSSHtmlFilter;
 12 
 13 public class MeteorChat extends HttpServlet {
 14 
 15     private final List list;
 16 
 17     public MeteorChat() {
 18         list = new LinkedList();
 19         list.add(new XSSHtmlFilter());
 20         list.add(new JsonpFilter());
 21     }
 22 
 23     @Override
 24     public void doGet(HttpServletRequest req, HttpServletResponse res) throws IOException {
 25         Meteor m = Meteor.build(req, list, null);
 26 
 27         req.getSession().setAttribute("meteor", m);
 28 
 29         res.setContentType("text/html;charset=ISO-8859-1");
 30         res.addHeader("Cache-Control", "private");
 31         res.addHeader("Pragma", "no-cache");
 32 
 33         m.suspend(-1);
 34         m.broadcast(req.getServerName() + "__has suspended a connection from " + req.getRemoteAddr());
 35     }
 36 
 37     public void doPost(HttpServletRequest req, HttpServletResponse res) throws IOException {
 38         Meteor m = (Meteor)req.getSession().getAttribute("meteor");
 39         res.setCharacterEncoding("UTF-8");
 40         String action = req.getParameterValues("action")[0];
 41         String name = req.getParameterValues("name")[0];
 42 
 43         if ("login".equals(action)) {
 44             req.getSession().setAttribute("name", name);
 45             m.broadcast("System Message from " + req.getServerName() + "__" + name + " has joined.");
 46             res.getWriter().write("success");
 47             res.getWriter().flush();
 48         } else if ("post".equals(action)) {
 49             String message = req.getParameterValues("message")[0];
 50             m.broadcast(name + "__" + message);
 51             res.getWriter().write("success");
 52             res.getWriter().flush();
 53         } else {
 54             res.setStatus(422);
 55 
 56             res.getWriter().write("success");
 57             res.getWriter().flush();
 58         }
 59     }
 60 }

Servlet 3.0 Async API is Game Over! Finally I must admit that Servlet 3.0 Async API have asynchronous dispatcher you can use to forward request asynchronously:

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

        final AsyncContext ac = req.startAsync();
        final String target = req.getParameter("target");

        Timer asyncTimer = new Timer("AsyncTimer", true);
        asyncTimer.schedule(
            new TimerTask() {
                @Override
                public void run() {
                    ac.dispatch(target);
                }
            },
        5000);
    }

With Atmosphere, the same code will works but your application will only works when deployed on Servlet 3.0 WebServer. Instead, you can implement the same functionality using Broadcast’s delayed broadcast API and still have a portable application without limiting you with Servlet 3.0 Async API…that’s something I will talk in my next blog!

For any questions or to download Atmosphere, go to our main site and use our Nabble forum (no subscription needed) or follow us on Twitter and tweet your questions there!

var gaJsHost = ((“https:” == document.location.protocol) ? “https://ssl.&#8221; : “http://www.&#8221;);
document.write(unescape(“%3Cscript src='” + gaJsHost + “google-analytics.com/ga.js’ type=’text/javascript’%3E%3C/script%3E”));

var pageTracker = _gat._getTracker(“UA-3111670-3″);
pageTracker._initData();
pageTracker._trackPageview();

technorati:

Categories: Uncategorized
Follow

Get every new post delivered to your Inbox.

Join 51 other followers