Source code: MessageProcessor.java

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");
                }
            }
        }
        
    }

}