Source code: MessageProcessor.java
|
[ index | 101 lines ]
|
package nl.west.aaa;
import java.util.*;
import java.io.*;
/**
* Class to put processing of messages in a seporate thread.
*/
class MessageProcessor
extends MessageHandlerContainer
implements MessageHandler
{
class MessageTuple
{
Message msg;
AAAUnit unit;
MessageTuple(Message msg,AAAUnit unit)
{
this.msg=msg;
this.unit=unit;
}
}
/**
* Queue.
*/
private LinkedList messageQueue;
/**
* Create a MessageProcessor that creates it's own queue.
*/
MessageProcessor()
{
this.messageQueue=new LinkedList();
// start 1 handling thread
new Processor().start();
}
/**
* Handle the given message.
* This method returns after storing the message to the
* queue.
* This method allways returns true. If the message cannot
* be handled it is droped.
*/
public boolean handleMessage(Message msg,AAAUnit unit)
{
MessageTuple dat=new MessageTuple(msg,unit);
synchronized (messageQueue)
{
messageQueue.addLast(dat);
messageQueue.notify();
}
return true;
}
/**
* Get the next message from the queue.
* This method blocks until a message can be read from
* the queue.
*/
private MessageTuple nextMessageTuple()
{
synchronized (messageQueue)
{
// wait until a message is present
while(messageQueue.isEmpty())
{
try
{
messageQueue.wait();
}
catch (InterruptedException e)
{
}
}
return (MessageTuple)messageQueue.removeFirst();
}
}
class Processor
extends Thread
{
public void run()
{
while (true)
{
MessageTuple nxt=nextMessageTuple();
if(!MessageProcessor.super.handleMessage(nxt.msg,nxt.unit))
{
System.err.println("MessageProcessor: Warning: incoming message not handled");
}
}
}
}
}
|