|
package nl.west.aaa;
import java.util.*;
import java.io.*;
/**
* A AAAUnit is a class used for transporting AAA messages
* between different clients/servers. Messages are sent
* useing a RecordFormat and a TransportProtocol.
*/
public class AAAUnit
extends MessageHandlerContainer
{
/**
* The identifier indicating the
* address of this AAAUnit in the network.
*/
private Identifier myId;
/**
* The transport protocols that can be used in receiving and
* sending messages. The preferred one first.
*/
private Vector transportProtocols=new Vector();
/**
* The record formats that can be used in receiving and sending
* messages. The preferred one first.
*/
private Vector recordFormats=new Vector();
/**
* The thread that processes incoming messages.
*/
private MessageProcessor messageProcessor;
/**
* Construct a new AAAUnit with the given Identifier.
*/
public AAAUnit(Identifier id)
{
this.myId=id;
// create the messageprocessor messagehandling is delegated to
this.messageProcessor=new MessageProcessor();
}
/**
* Process incoming connections. This method is called by a
* TransportProtocol when incoming
* data is available from a connection.
* It reads messages from the stream and passes it to the
* registered handlers.
* This method only returns when the stream is closed.
*/
public void handleIncoming(InputStream in,Identifier from,TransportProtocol prot)
{
// synchronize on the stream: only one thread
// may read messages from a single stream
synchronized (in)
{
// add a buffer around the stream
if(!in.markSupported())
in=new BufferedInputStream(in);
// continue reading untill exception
try
{
while(true)
{
// trye recordformats
Enumeration e=recordFormats.elements();
Message msg=null;
RecordFormat rf=null;
while(e.hasMoreElements()&&(msg==null))
{
rf=(RecordFormat)e.nextElement();
msg=rf.decodeMessage(in,from);
}
// check if message was parsed
if(msg==null)
{
// check for end of stream
if(in.read()!=-1)
System.err.println("InputStream: record format not recogdnized");
in.close();
return; // to avaoid inf loop
}
// pass to handler
handleIncoming(msg,from,prot,rf);
}
}
catch (IOException e)
{
e.printStackTrace(System.err);
}
}
}
/**
* Process incoming connections. This method is called by a
* TransportProtocol when incoming
* data is available from a connection.
* Only one message may be present in the data.
*/
public void handleIncoming(byte[] data,Identifier from,TransportProtocol prot)
{
// check out record format
Enumeration e=recordFormats.elements();
Message msg=null;
RecordFormat rf=null;
while(e.hasMoreElements()&&(msg==null))
{
rf=(RecordFormat)e.nextElement();
msg=rf.decodeMessage(data,from);
}
// check if message was parsed
if(msg==null)
{
// error, no matching record format found
// could not process, so ignore
System.err.println("data: record format not recogdnized");
return;
}
// pass to handler
handleIncoming(msg,from,prot,rf);
}
/**
* Finish handling of incoming messages by passing the
* message to the messageProcessor. Allso does some
* small sanity checks.
*/
private void handleIncoming(Message msg,Identifier from,TransportProtocol prot,RecordFormat rf)
{
// set the sender if neccesary
if(msg.sender==null)
msg.sender=from;
// set the receiver if neccesary
if(msg.receiver==null)
msg.receiver=myId;
// set the recordformat
msg.recordFormat=rf;
// set the transportprotocol
msg.transportProtocol=prot;
// queue it for passing to the handlers
messageProcessor.handleMessage(msg,this);
}
/**
* Handle an incoming message. All incoming messages
* are passed to the registered MessageHandlers.
*/
public boolean handleMessage(Message msg,AAAUnit unit)
{
return messageProcessor.handleMessage(msg,unit);
}
/**
* Send a message.
* A Message may have a RecordFormat and/or
* TransportProtocol defined. If a
* RecordFormat or TransportProtocol is defined
* it is used (even if it is not registered).
* If no RecordFormat or TransportProtocol
* is defined the defaults are used.
* <br>
* Note that when sending a message with a non-registered
* TransportProtocol no replies sent by that
* TransportProtocol will be read.
*/
public void sendMessage(Message msg,Identifier to)
throws IOException
{
msg.receiver=to;
if(msg.recordFormat==null)
msg.recordFormat=(RecordFormat)recordFormats.elementAt(0);
if(msg.transportProtocol==null)
msg.transportProtocol=(TransportProtocol)transportProtocols.elementAt(0);
// send directly
msg.sender=this.myId;
// encode the message
byte []data=msg.recordFormat.encodeMessage(msg);
// check if encoding is successfull
if (data==null)
throw new IOException("encoding failed");
// save encoded data
msg.data=data;
// send it
msg.transportProtocol.sendMessage(data,msg.receiver);
}
/**
* Add the given TransportProtocol to the list
* of supported
* protocols. This unit will be registered for
* processing incoming connections for the given
* TransportProtocol.
* The first added protocol will be
* the default for sending.
*/
public void addTransportProtocol(TransportProtocol prot)
{
// add to list (for sending)
transportProtocols.add(prot);
// add listener (for receiving)
prot.startListening(this);
}
/**
* Remove the given TransportProtocol from the list
* of available TransportProtocols. Incoming
* connections on the TransportProtocol will no longer
* be handled from this protocol. Existing connections
* will not be terminated.
*/
public void removeTransportProtocol(TransportProtocol prot)
{
// remove from list (for sending)
transportProtocols.remove(prot);
// remove listener (for receiving)
prot.startListening(null);
}
/**
* Adds the given RecordFormat to the list of
* available RecordFormats. The first entry
* will be the default.
*/
public void addRecordFormat(RecordFormat rf)
{
recordFormats.add(rf);
}
/**
* Remove the given RecordFormat from the list
* of available RecordFormats.
*/
public void removeRecordFormat(RecordFormat rf)
{
recordFormats.remove(rf);
}
/**
* Register the MessageHandler as a handler of
* incoming messages. All incoming messages are fed through
* the registered MessageHandlers.
*/
public void addMessageHandler(MessageHandler handler)
{
messageProcessor.addMessageHandler(handler);
}
/**
* Remove the given MessageHandler from the list
* of registered MessageHandlers.
*/
public void removeMessageHandler(MessageHandler handler)
{
messageProcessor.removeMessageHandler(handler);
}
}
|