Pages

Thursday, June 18, 2009

I have worked on couple of push technologies namely blazeDS and Lightstreamer and always wondered how does it work. I am just concerned about streaming in Flex and not anything else. You can think I am little inclined towards flex, but the truth is Flex rocks, and after I entered the world of Flex, other UI technologies - I don't care anymore! Anyways, this logic of streaming data to Flex from a backend came to me just by chance! It took 2 days effort to get the below crappy thing going! I read a blog on the Internet that used URLStream to load image data in Flex progressively. Now, if you can load image progressively, then I should be able to do the same with any data. So, here I present Flex - XSocket combination to subscribe/load data from backend for a watchlist. There are a few things, that I did not resolve, as I did not have time to dedicate for it. I will explain those at the end.

JAVA

Lets start from the java backend. I use XScoket Xlightweb library to create a NIO server that can take any symbol as subscription and pushes randomly generated quotes to the client.

Lets write a request handler. In XLightweb, the request handler implements the IHttpRequestHandler.

public class ERequestHandler implements IHttpRequestHandler{ //Timer that generates the quotes at the specified interval private final Timer timer = new Timer("timer", true); public static Map<String, SymbolSender> state = new HashMap<String, SymbolSender>();

Lets implement the onRequest method. The same handler will be used to receive commands, like add symbol, remove symbol and also to receive data request, we use the URI to differentiate between the two request.

Every ipAddress is given its own handler. So, obtain the handler ie SymbolSender and delegate the action to it. The SymbolSender is a private class that is given little below. Before any client sends an action, it has to connect to this server.

The action has been delegated, so close the channel with a 200 Response header as in the below snippet.

IHttpResponseHeader respHdr = new HttpResponseHeader(200, "text/plain"); BodyDataSink outChannel = exchange.send(respHdr); outChannel.close(); }else if(req.getRequestURI().endsWith("/link2")){//The initial Data link, that is used to send data back. String command = req.getParameter("action"); Initialize a new SymbolSender if none exist for the given ipAddress and start the handler using the timer. Here, it is scheduled at a fixed rate of every 500ms.

The stream listener for the data is given below. Very simple logic, just read the multi byte which is already written in form of utf-8 by the server and append it to the data string. Now, this stream listener is called by Flex when there is some specific amount of data is present. And, it does not make sense for me to parse the data here to create objects at that rate. Instead as you will see next code, the timer will invoke the parse method every 60ms and then parse the data in the data string to an appropriate VO.

Now, the timepass listener is here. Comment the alerts if you do not want, or use trace instead.

private function listener(event:Event):void{ Alert.show("Here at other side of the world :("); }

Now, here is the parse method for the data I stream from the server. Actually, I do not like to call it streaming, as it is just a normal read and write to streams. Heck! that's why we call it streaming. He he..

Find out if the data is written partially. That means if its not ending with "|". If that is the case then keep the last piece of string in the data string for appending with the next data arrived from the backend. This will be parsed after the data has arrived and 60ms time lag.

Parse all the data other than "STAY ALIVE" pings. I am not sure if you can stop sending STAY ALIVE or delay its frequency. And, I am not doing that R & D either. My purpose remains just of stream data to Flex. Only the positive use case :) The rest of the for loop is just parsing data, creating or updating the StreamerVO object.

Next lets write the 4 actions supported namely add, delete, start and stop. The url's are in sync with the actions supported by respective links given in the backend code. The get parameters are constructed manually as shown below.

And, that's all folks. The backed is writing the data to the channel, and front end flex is reading it and displaying. Well, there are few bugs in this code I presented.

The data does not start to flow to the front end for initial 20 sec or so. Try to connect and subscribe say 3-4 symbols IBM, GOOG, MSFT, CSCO soon. I guess the problem is with the Operating system buffer because even when I said flush manually in the XSocket code (Application buffer) no effect could be seen or the problem could exist in Flex also, I am not sure. As I said earlier, my goal is not to solve these problems but to rather get streaming up, which gets up after a initial delay which is fine with me right now.

I had to use two links so that I could add symbols to be subscribed with out re-connecting to get the data that is being pushed. I could not figure out a way to do it using a single link. You guys can give a shot and let me know. I leave this here for now, as I am in no mood of writing a complete infrastructure for push when several exists.

Below is a small video of the Flex UI using the code above showing streaming data.

Wednesday, June 17, 2009

Writing an MBean in java is very simple. But, when you want the return type to a table of data(say, some metrics) then simple Mbeans are not of much use. One way to get table of data as a return type is using a Open Mbeans. This tutorial gives a simple example of how to write an Open Mbean that returns Table of data.

All the variables can be made private and initialized either in a static block or in the constructor. The CompositeType just defined that each row will contain the columns "symbol", "localTimeStamp", "latestResponse". Now lets define the table type it self.

OpenMBeanParameterInfo[] params = new OpenMBeanParameterInfoSupport[0]; operations[0] = new OpenMBeanOperationInfoSupport( "resetAndGetQuoteSnapshot", "Reset and get the latest available data for the Quotes", params, quoteSnapshotType, MBeanOperationInfo.INFO); //Build the info

This is all is needed to build an Open MBean that supports a Tabular data as output. You can use jconsole and view the output. Obviously, you have to deploy the MBean in mbean server. I use spring to do that job. The spring XML is given below.