Thread.currentThread().join()

Will continues to post articles about MINA, to be updated, Subscribe in a reader

Apache MINA has wonderful concept of ProtocolDecoder to process Decoding protocol specific messages. XML is one of the most widely used format for EDA. Lets see how can we implement a Protocol Decoder for Apache MINA.

Algorithm

The picture below describes the basic algorithm that we need to use to construct an XML message from bytes.

The logic is simple, keep reading the bytes till the XML message is balanced. Balanced here means, that end of the root element has been achieved. For eg. If the xml document has root element as , we have to read the bytes till we received .

Its very particular to note that large XML packets when sent over TCP, may get fragmented and we shall received the same amount of read events while using Apache MINA low level API’s.

This type of situations where, we need to wait to data to completely arrive, calls for the use of CumulativeProtocolDecoder. As the name signifies, the decoder waits till, we get the balanced xml. Once the balanced XML is found, we write the parsed object to the output, to be processed further.

Lets see the code. My apologies for the unformatted code

public abstract class XMLDecoder extends CumulativeProtocolDecoder  {
 
/*
* As per XML specification 1.0, http://www.w3.org/TR/REC-xml
*/
private static final char XML_START_TAG = '<';
private static final char XML_END_TAG = '>';
private static final char XML_PI_TAG = '?';
private static final char XML_COMMENT_TAG = '!';
 
protected static enum ParseState {ELEMENT_START, ELEMENT_END, COMMENTS, ENDELEMENT, PI, UNDEFINED};
 
protected static final int ELEMENT_START = 1;
protected static final int ELEMENT_END = 2;
 
private static Logger logger = LoggerFactory.getLogger(XMLDecoder.class);
 
@Override
protected boolean doDecode(IoSession session, IoBuffer ioBuffer,
ProtocolDecoderOutput decoderOutput) throws Exception {
 
int startPosition = ioBuffer.position();
 
if(!ioBuffer.hasRemaining()) {
logger.debug("NO bytes to read keep waiting...");
return false;
}
 
// Continue to read the bytes and keep parsing
char currentChar = '0', previousChar = '0';
 
boolean rootElementStarted = false;
boolean rootElementPresent = false;
boolean isBalanced = false;
 
int rootStartPosition, rootEndPosition;
 
ParseState parsingState = ParseState.UNDEFINED;
logger.debug("Lets start decoding the XML");
 
String root = null;
 
boolean markedForEndElement = false;
 
while(ioBuffer.hasRemaining()) {
previousChar = currentChar;
currentChar = (char)ioBuffer.get();
 
switch (parsingState) {
case ELEMENT_START:
if(currentChar == XML_PI_TAG){
logger.debug("Got PI Element");
parsingState = ParseState.PI;
} else if(currentChar == XML_COMMENT_TAG) {
logger.debug("Got Comment Element");
parsingState = ParseState.COMMENTS;
} else if((currentChar == ' ' || currentChar == XML_END_TAG)
&& rootElementStarted && !rootElementPresent) {
rootEndPosition = ioBuffer.position();
rootElementPresent = true;
 
// Copy the Root Element
int cPos = ioBuffer.position();
int mPos = ioBuffer.markValue();
 
char[] rootChar = new char[cPos - mPos];
for(int i = mPos - 1, j =0; i < cPos - 1; i++) {
rootChar[j++] = (char)ioBuffer.get(i);
}
 
root = new String(rootChar);
logger.debug("Root Element = "+ root);
parsingState = ParseState.ELEMENT_END;
logger.debug("Root Element detection completed "+rootEndPosition);
} else if(currentChar == XML_END_TAG) {
parsingState = ParseState.ELEMENT_END;
} else if(!rootElementStarted && !rootElementPresent) {
rootStartPosition = ioBuffer.position();
ioBuffer.mark();
rootElementStarted = true;
logger.debug("Got the root element at "+rootStartPosition);
} else if (currentChar == '/') {
// Change state
if(previousChar == XML_START_TAG) {
parsingState = ParseState.ENDELEMENT;
}
}
break;
 
case ENDELEMENT:
if(currentChar == XML_END_TAG) {
parsingState = ParseState.ELEMENT_END;
 
int cPos = ioBuffer.position();
int mPos = ioBuffer.markValue();
 
char[] el = new char[cPos - mPos];
for(int i = mPos - 1, j =0; i < cPos - 1; i++) {
el[j++] = (char)ioBuffer.get(i);
}
markedForEndElement = false;
if(root.equalsIgnoreCase(new String(el))) {
logger.debug("XML is balanced."+root);
isBalanced = true;
}
 
break;
} else if (currentChar == ' ') {
continue;
} else {
 
// mark the position, we need to compare the it to see that if its the end element
if(!markedForEndElement) {
ioBuffer.mark();
markedForEndElement = true;
}
}
break;
 
case ELEMENT_END:
if(currentChar == XML_START_TAG) {
parsingState = ParseState.ELEMENT_START;
}
break;
 
case UNDEFINED:
if(currentChar == XML_START_TAG) {
parsingState = ParseState.ELEMENT_START;
}
break;
 
case COMMENTS:
if (currentChar == '-') {
previousChar = currentChar;
} else if (previousChar == '-' && currentChar == '>') {
parsingState = ParseState.ELEMENT_END;
}
break;
 
case PI:
if (currentChar == '?') {
previousChar = currentChar;
} else if (previousChar == '?' && currentChar == XML_END_TAG) {
parsingState = ParseState.ELEMENT_END;
}
break;
 
default:
break;
}
}
 
if(isBalanced) {
decoderOutput.write(parserXML(ioBuffer));
}
 
if(isBalanced && !ioBuffer.hasRemaining()) {
logger.debug("No more bytes to process");
return true;
}
 
ioBuffer.position(startPosition);
return false;
}
 
/**
* Extending classes can implement their custom XML parsing to create Objects
* from XML and use them appropriately in Handler
*
* @param xmlBuffer
* @return
*/
public abstract Object parserXML(IoBuffer xmlBuffer);
}

The implementation is pretty straight forward. We take each character and try to match the characters as specified in XML specification.

Some keys things in the implementation:
1. The Decode function just collects the bytes till we get the balanced XML document
2. Once we get the balanced XML document, we shall call the abstract function parseXML(). The function has been kept abstract, so that its easy to implement custom parsing using desired XML library like JAXB, JIBX etc
3. We have to return true from doDecode(), the moment we have balanced XML. Return type true indicates to the framework that we are not waiting for any more data. A false, forces the framework to keep accumulating the data, till we write it to the output. Now it must be clear why, its called Cumulative decoder.

Still have Queries, please leave a comment and I shall revert back to you.

In this post, we shall implement a Trap Sender using SNMP4J. We may choose to use Apache MINA for sending Traps or can resort to using DatagramSocket class directly.

This shall be the logical flow of the implementation

• Get the encoded Trap Data
• Send the Trap 

The code snippet above shows a simple way of creating and encoding a Trap PDU. Essentially, we create an instance of PDU class and sets the type as Trap. This is important, else SNMP4J shall throw an exception. Thereafter, we can set the trap parameters. Here, we have hardcoded the parameters, there can be custom implementations that can take these from config files or from UI. After setting the parameters, we just call the encode function passing the Output stream and collect the byte array to be sent.

Sending part is even simpler     

The send code is preety straight forward. Here we have used Datagram Socket, we can use Apache MINA UDP Client implementation to send the trap as well.

References

• http://www.snmp4j.org/
• Implementing UDP Client using Apache MINA 

Will continues to post articles about MINA, to be updated,  Subscribe in a reader

Yes, we are going to implement a SNMP trap receiver in less than 30 minutes. If you have been following my post on Apache MINA, this would be a natural extension to it. In this article, we shall bring together all our components to build the system.

 Pre-requisites:

 Please read through these articles. You can find all these posts on Apache MINA

• What is Apache MINA
• Apache MINA based Server Application Architecture
• Implementing UDP Server using Apache MINA
• Implementing SNMP4J Decoder for Apache MINA

 To execute the code, you need following jars

• mina-core-2.0.0-M1.jar
• slf4j-api-1.5.0.jar
• slf4j-log4j12-1.5.0.jar
• log4j-1.2.15.jar
• SNMP4J.jar

 I am assuming that you have read my previous posts and jumping straight to the implementations. The high level architecture is explained in the post Apache MINA based Server Application Architecture

Architecture 

Let’s understand the flow here and see what all we need to do to process a trap

• Receive the trap over UDP
• Decode the Trap
• Dump the Trap

Lets see how each of the these maps to the code we have already written

It's a now simple to understand that we have reused all the components to create the Server. Let me not write too much, and jump straight to the code J

 The Server Code

The only change that we did to the UDP Server code was to add a Protocol Codec. At line 36, we added our custom ProtocolCodecFiler. The SNMPCodecFactory return the instance of our SNMP4J codec. There is not much code and this construct is fairly simple, and explained very well in MINA’s documentation. From the main method, we just need to call the method initialize() and our trap receiver starts.

References

• http://mina.apache.org/
• http://www.snmp4j.org/

In my last post we created a UDP client using Apache MINA. Lets turn the table and implement the Server side. Let's see how using Apache MINA reduces the effort to create a UDP Server.

Steps to create a UDP Server using java.net API's

1. Create a Socket and listen for incoming connection
2. Process each packet in a separate thread (I hate this, unfortunately need this to have high processing rate)
3. Parse and process the request and optionally send response (Lets omit this to keep things simple)

That's it

• Our protocol is carrying Strings in UDP packet
• We shall not do any transformation on the packets received. We shall just dump the content

The Handler

 

The Handler is in its basic form. Our only method of interest is messageReceived. Since we know that we are getting "String" message without any transformation, we could easily create a new string from the bytes received. A wonderful thing to note about this API is the parameter "message". This makes the API generic enough to cater to any kind of objects. If we had used a ProtocolCodec in the chain and had transformed the byte[] into a custom Object, we would have type-casted message to that object.

That's all in the handler.

 

The Server

Lets see the main Server code

 

 

The Server is even simpler than client

We create an instance of NioDatagramAcceptor and add our custom handler to it. We then bind to the port desired. Here I have made it bind to the default port, code can be customized to bind to any port desired.

So what does 31 does. There is wonderful description of this option at http://www.unixguide.net/network/socketfaq/4.5.shtml
The main function is pretty simple. The Server is ready. You can run it, using the Client from the post Implementing UDP Client using Apache MINA

References

• http://mina.apache.org/udp-tutorial.html

Its been a while that I wrote on Apache MINA. Please refer to my post collection on Apache MINA.

In this post, I have tried to capture briefly on how to implement a UDP Client using Apache MINA. We shall concentrate on keeping our scope limited to sending data over UDP. In Subsequent posts, we shall see how we can enhance this UDP Client to send SNMP Traps.

In brief, these are the steps we need to perform to send a UDP packet

1. Create a Datagram Socket
2. Create a Datagram Packet
3. Send the packet

1. Create a NioDatagramConnector instance
2. Add an IoHandler (we can use an IoHandlerAdapter, as for this example we don't need to use all the API's)
3. Connect the NioDatagramConnector
4. Get the session
5. Write the data onto the session

There is a slight difference in which the MINA API's work. The API's work asynchronously. A call to connect() on NioDatagramConnector, return a reference to ConnectFuture and the operation is kicked off in a new thread. Lets see how the Code looks like and see it in details.

 

 

Lets see the Code in detail now. We have maintained the Session at the class level, assuming that we may reuse the session for sending packets, if session is still active. At line 25, we check if the session is active or not. If inactive we initiate the connection. 

Line 28, creates an instance of NioDatagramConnector

Line 31, add an IoHandler to the NioDatagramConnector on Line 28. In MINA terminology, IoHandler contains the code where Business logic of an Application resides. In our case its a pure Adapter, with only log statements inside the functions.

Line 32, a call to connect(), returns an reference to ConnectFuture. The call, starts the connection operation, in a separate thread and returns to the caller. To cater to this asynchronous behaviour, we add an IoFutureListener, to returned ConnectFuture reference. Out there (Line 38-43), we just check if the connection is OK, we get the session from it. Line 39, signals that connection is connected and its safe to get the session for writing data.

The connect code is complete. Lets see how to send the data.

The send function is pretty simple. We just create an IoBuffer and call write on the session. We have called connect() on Line 59, just to ensure that the connection is alive. If it isn't, it shall create a connection and return a session back.

NOTE: This code is in its very simple form. There can be various mechanism to handle Session (based on destination etc). Also, please be aware of the asynchronous behaviour of the NioDatagramConnector connect() API. If the connection is not complete in time, you might see a null pointer at Line 67 . But this is usually the case. For production code, it is advisable to ensure that session is never null.

That's it, out UDP client is ready. To create an SNMP Trap sender, we just need to pass Trap byte array to send() and we are done. Will try to post that article soon.

References

• http://mina.apache.org/udp-tutorial.html