Now when XMPP subsystem is designed and implementation is ready and polished I’d like to share some my ideas.
Lets start from class diagram.
Connection class represents object which can communicate with remote XMPP server. All network packets that are produced by server come to Connection class through recieveMessage() method. Connection can send network packets back to XMPP server using send() method implementation of ITransport interface.
When Connection receives packet from server it performs some basic transformation of packet (e.g. extracts sender’s jabber identifier, packet text contents, packet type and so on) and creates Message object. This object comes to Processor that does all actual work — message processing. If message answer should be created, Processor creates it and puts answer to TransportQueue using it’s add() method.
TransportQueue is active object. It periodically polls it’s internal thread-safe queue and if the internal queue isn’t empty it’s contents are sent to ITransport.
Here is the picture:
The core idea is that Processor can be facade over multiple services and active objects. It’s processMessage() method should’n block and return immediately. We’ll talk about this later. Now let’s examine why TransportQueue is active object and why it is needed.
As it was written above Processor can contain multiple services and active objects. They can process the same messages differently but simultaneously. Here is the picture:
As you can see Processor follows Composite Pattern. It stores and manages internal pluginList collection. There are shown only two plugins on the picture above for simplicity. In real system there can be hundreds of plugins. So when processMessage() method is called on Processor it actually redirects this method call to all it’s plugins. What would be if two or more plugins will produce answer simultaneously? Nothing good — all of them will send answer packet to Connection simultaneously.
Yes, we can synchronize send() method of Connection class. But this way processMessage() method of composite Processor will block. Lets illustrate this with small example:
There are is the StopwatchPlugin which performs time tracking. If activated using text command this stopwatch plugin produces small text messages with current time text. In order to stop stopwatch another command should be received. LoggerPlugin logs every text message that Processor receives and sends report with logging status back to server.
So lets imagine that Processor has received message with StopwatchPlugin start command. Processor redirects copy of this message to StopwatchPlugin. Plguin starts it’s execution. Let’s say every 5 seconds it generates text message with current timestamp and puts this message to TransportQueue.
After StopwatchPlugin has been started Processor sends the copy of start command to LoggerPlugin. It logs the command text into some external data storage and creates message with logging status which should be send back to xmpp server. This message is put into TransportQueue.
As it was written above TransportQueue has thread-safe add() method. So if we would have even hundred of plugins and all of them will be generating answers for remote commands simultaneously the access to Connection will be synchronized using TransportQueue. And the processMessage() method of Processor won’t block.
Finally lets examine the order of creation of components of XMPP subsystem.
We use XmppManager notation here to illustrate facade over all components of subsystem. This facade stores Processor, Connection and TransportQueue. First of all Processor instance is created and all it’s plugins are initialized. Then Connection instance is created using already existing Processor. Then TransportQueue instance is created using already existing Connection.
Finally TransportQueue is assigned to Processor using it’s addTransport() method.
Next time we will talk about messages hierarchy and take a closer look to Connection