casa.agentCom

Class SocketServerTCPIP

java.lang.Object

java.lang.Thread

casa.agentCom.SocketServerTCPIP

All Implemented Interfaces:

SocketServerInterface, Runnable

public class SocketServerTCPIP
extends Thread
implements SocketServerInterface

This class is designed to listen on a socket on behalf of one or more agents. For each connection it receives, it creates a TCPChannel object representing the channel. The constructor automatically runs itself as a thread. Objects of this class will Accept, Read, and Write MLMessage objects, and maintain the socket connections until they are broken at the far end.

Copyright: Copyright 2003-2014, Knowledge Science Group, University of Calgary. Permission to use, copy, modify, distribute and sell this software and its documentation for any purpose is hereby granted without fee, provided that the above copyright notice appear in all copies and that both that copyright notice and this permission notice appear in supporting documentation. The Knowledge Science Group makes no representations about the suitability of this software for any purpose. It is provided "as is" without express or implied warranty.

Author:

Rob Kremer, Roberto Flores

Nested Class Summary

Nested classes/interfaces inherited from class java.lang.Thread

Thread.State, Thread.UncaughtExceptionHandler

Field Summary

Fields

Modifier and Type	Field and Description
private static Vector<SocketServerTCPIP>	all All the socket servers in the process
private static Charset	ENCODING Used in #readAMessage(SocketChannel) and writeAMessage(SelectionKey) to decode/encode the incoming message buffer.
(package private) static int	HEADER_SIZE The size of header.
private InetAddress	IPaddress the local IP address of socket
private int	IPport The port number the socket is listening on
private static int	numberOfPortsToTry When the IP port given in negative, it means to try ports n++ until an open port is found, this limits the number of tries
private LimitedSet<MLMessage>	recentMessages A buffer to remember recently received messages to detect possible endless loops of resending messages.
private boolean	recoveryMode True if the we are lost in the channel and scanning for a valid header.
private ConcurrentLinkedQueue<Pair<SocketChannel,TCPChannel>>	registerQueue Used by registerChannel(SocketChannel, TCPChannel) and run() to safely do new registrations without generating an interrupt on the socket listener thread.
private Selector	selector The selector object used in the run() method
private static int	SIGNATURE This is a 4-byte signature to be prepended to a message send/receive to identify the start of a message.
private ServerSocket	socket The server socket listing on port IPport
private static boolean	useLoopbackAsFallback If we fail to get a local host address, setting this to true will use the loopback address instead.

Fields inherited from class java.lang.Thread

MAX_PRIORITY, MIN_PRIORITY, NORM_PRIORITY

Constructor Summary

Constructors

Constructor and Description
SocketServerTCPIP(int IPport) Constructs a new SocketServer listening on the specified port.

Method Summary

Modifier and Type	Method and Description
private TCPChannel	addConnection(URLDescriptor url, SelectionKey key, InfiniteReadWriteByteBuffer initBuffer) Creates a new TCPChannel object, and links it to url.
void	closePort() Method to close the socket server's open port and return it to the system.
void	exit() This method is ignored.
static SocketServerTCPIP[]	get()
String	getHostAddress() Retrieves the address of the listening socket
int	getLocalPort() Retrieves the port of the listening socket
String	getLocalPortAsString() Retrieves the port of the listening socket
void	queueAWrite(SelectionKey key) External threads may call this method to queue a write on the channel of key.
private boolean	queueMessage(MLMessage message) Queue a message that has been read to a (probably) local agent.
private int	readABlockHeader(byte[] buffer) Read a header, checking the signature (first 4bytes) and returning the size (in bytes) of the body that follows the header.
private int	readData(SelectionKey key) Attempt to read an arbitrary number of bytes from key.channel(), and places it in key.attachment().buffer, where key.attachment() is a TCPChannel object or a InfiniteReadWriteByteBuffer.
private void	readMessages(SelectionKey key, Vector<SocketChannel> orphanedChannels) Does a read from the key's TCP/IP channel and tries to interpret messages and queues them to the relevant agent's event queue.
private static String	readyOpsToString(int code) Utility method to print out the OP codes in a more legible manner in the log messages.
private void	recover(InfiniteReadWriteByteBufferInterface buffer) Attempt to recover from corrupt data by scanning for a legal header in the byte buffer.
(package private) void	registerChannel(SocketChannel channel, TCPChannel tcpChannel) External threads may call this method to register the channel.
private void	removeConnection(SelectionKey key) Removed the URL as an attachment from the key, cancels the key, and removes the key/channel from the URL.
void	run() The runnable obligation for this thread.
private void	setPort(int IPport) Set the port to listen on.
private void	writeAMessage(SelectionKey key) Called when the key.channel is ready for writing, but there may be nothing there to write.

Methods inherited from class java.lang.Thread

activeCount, checkAccess, clone, countStackFrames, currentThread, destroy, dumpStack, enumerate, getAllStackTraces, getContextClassLoader, getDefaultUncaughtExceptionHandler, getId, getName, getPriority, getStackTrace, getState, getThreadGroup, getUncaughtExceptionHandler, holdsLock, interrupt, interrupted, isAlive, isDaemon, isInterrupted, join, join, join, resume, setContextClassLoader, setDaemon, setDefaultUncaughtExceptionHandler, setName, setPriority, setUncaughtExceptionHandler, sleep, sleep, start, stop, stop, suspend, toString, yield

Methods inherited from class java.lang.Object

equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface casa.agentCom.SocketServerInterface

interrupt, isAlive

Field Detail

numberOfPortsToTry

private static final int numberOfPortsToTry

When the IP port given in negative, it means to try ports n++ until an open port is found, this limits the number of tries

See Also:

Constant Field Values

useLoopbackAsFallback

private static final boolean useLoopbackAsFallback

If we fail to get a local host address, setting this to true will use the loopback address instead. Otherwise we will fail.

See Also:

Constant Field Values

ENCODING

private static final Charset ENCODING

Used in #readAMessage(SocketChannel) and writeAMessage(SelectionKey) to decode/encode the incoming message buffer.

SIGNATURE

private static final int SIGNATURE

This is a 4-byte signature to be prepended to a message send/receive to identify the start of a message.

See Also:

writeAMessage(SelectionKey), #readABlockHeader(SocketChannel), Constant Field Values

all

private static Vector<SocketServerTCPIP> all

All the socket servers in the process

HEADER_SIZE

static final int HEADER_SIZE

The size of header.

See Also:

readABlockHeader(byte[]), Constant Field Values

socket

private ServerSocket socket

The server socket listing on port IPport

IPaddress

private InetAddress IPaddress

the local IP address of socket

IPport

private int IPport

The port number the socket is listening on

selector

private Selector selector

The selector object used in the run() method

recoveryMode

private boolean recoveryMode

True if the we are lost in the channel and scanning for a valid header.

See Also:

recover(InfiniteReadWriteByteBufferInterface)

registerQueue

private ConcurrentLinkedQueue<Pair<SocketChannel,TCPChannel>> registerQueue

Used by registerChannel(SocketChannel, TCPChannel) and run() to safely do new registrations without generating an interrupt on the socket listener thread.

recentMessages

private LimitedSet<MLMessage> recentMessages

A buffer to remember recently received messages to detect possible endless loops of resending messages.

See Also:

queueMessage(MLMessage)

Constructor Detail

SocketServerTCPIP

public SocketServerTCPIP(int IPport)
                  throws IPSocketException

Constructs a new SocketServer listening on the specified port. The port number is negative, this constructor will look for a empty port between IPport and IPport+numberOfPortsToTry. When using this constructor the thread is started.

Parameters:

IPport - the port to listen on. A negative number indicates to try other ports.

Throws:

IPSocketException - if (IPport is negative) a valid port in the range IPport -> IPport+numberOfPortsToTry is not found, or if (IPport is positive) IPport is not available.

Method Detail

get

public static SocketServerTCPIP[] get()

Returns:

All the SocketServers in this process.

setPort

private void setPort(int IPport)
              throws IOException

Set the port to listen on.

Parameters:

IPport - the port to listen on

Throws:

IOException - when either 'e' is an instance of IOException or when getLocalHost() fails. <-- In any of these cases, the port will fail to listen.

getHostAddress

public String getHostAddress()

Retrieves the address of the listening socket

Specified by:

getHostAddress in interface SocketServerInterface

getLocalPortAsString

public String getLocalPortAsString()

Retrieves the port of the listening socket

Specified by:

getLocalPortAsString in interface SocketServerInterface

getLocalPort

public int getLocalPort()

Retrieves the port of the listening socket

Specified by:

getLocalPort in interface SocketServerInterface

run

public void run()

The runnable obligation for this thread. This method uses selector to ACCEPT connections to channels, as well as to READ and WRITE through the selected channels.

Specified by:

run in interface Runnable

Overrides:

run in class Thread

registerChannel

void registerChannel(SocketChannel channel,
                     TCPChannel tcpChannel)

External threads may call this method to register the channel. Note that the channel is NOT necessarily registered during this method's invocation, but is queued for registration when it's safe to interrupt the socket listener thread. This method is synchronized with the i/o methods because it generates an interrupt in the socket listener thread, which would cause the channel to be closed if it happens during i/o operations.

Parameters:

channel -

tcpChannel -

queueAWrite

public void queueAWrite(SelectionKey key)

External threads may call this method to queue a write on the channel of key. Note that the write is NOT necessarily done during this method's invocation, but is queued for when it's safe to interrupt the socket listener thread. This method is synchronized with the i/o methods because it generates an interrupt in the socket listener thread, which would cause the channel to be closed if it happens during i/o operations.

Parameters:

key -

writeAMessage

private void writeAMessage(SelectionKey key)
                    throws ClosedChannelException

Called when the key.channel is ready for writing, but there may be nothing there to write. All the message queued for writing in the key's TCPChannel are written, and the key's interest is set back to OP_READ. The message actually send on the channel is in the following format:

1. a 4-byte "signature" equal to the SIGNATURE constant of value -1498131167.
2. a 4-byte length indicating the number of bytes in the body message
3. the actual message data encoded according to ENCODING of

This method is synchronized with to avoid interrupts (which would close the channel) from registerChannel(SocketChannel, TCPChannel) and queueAWrite(SelectionKey).

Parameters:

key -

Throws:

ClosedChannelException

removeConnection

private void removeConnection(SelectionKey key)

Removed the URL as an attachment from the key, cancels the key, and removes the key/channel from the URL. This method is synchronized with to avoid interrupts (which would close the channel) from registerChannel(SocketChannel, TCPChannel) and queueAWrite(SelectionKey).

Parameters:

key -

recover

private void recover(InfiniteReadWriteByteBufferInterface buffer)

Attempt to recover from corrupt data by scanning for a legal header in the byte buffer. When this method terminates, the buffer will either begin with a valid header or there will be less than HEADER_SIZE (8) bytes left in the buffer. This behaviour allows a recovery to span multiple channel reads.

This is a VERY inefficient implementation of recover(), but fairly robust.

Parameters:

buffer -

readMessages

private void readMessages(SelectionKey key,
                          Vector<SocketChannel> orphanedChannels)
                   throws ClosedChannelException

Does a read from the key's TCP/IP channel and tries to interpret messages and queues them to the relevant agent's event queue. Actually, if the message is NOT directed at a local agent, the message is forwarded to the agent the information is available in its URLDescriptor.

Parameters:

key -

orphanedChannels -

Throws:

ClosedChannelException

readData

private int readData(SelectionKey key)
              throws IOException

Attempt to read an arbitrary number of bytes from key.channel(), and places it in key.attachment().buffer, where key.attachment() is a TCPChannel object or a InfiniteReadWriteByteBuffer.

Parameters:

the - key containing the channel and attachment objects.

Returns:

the number of bytes read.

Throws:

IOException

readABlockHeader

private int readABlockHeader(byte[] buffer)
                      throws SignatureException

Read a header, checking the signature (first 4bytes) and returning the size (in bytes) of the body that follows the header.

Parameters:

buffer - The bytes to read; must be HEADER_SIZE (8).

Returns:

the size of the data in the body of this block (excluding the header).

Throws:

SignatureException - if the signature in the header doesn't match.

queueMessage

private boolean queueMessage(MLMessage message)
                      throws ClosedChannelException

Queue a message that has been read to a (probably) local agent. The "(probably)" is in the previous sentence because it's possible that we might be forwarding on the message, in which case this event will be logging as a warning tag.

Parameters:

toURL -

message -

Returns:

true if the message could be sent (the toURL had a channel); otherwise false.

Throws:

ClosedChannelException

addConnection

private TCPChannel addConnection(URLDescriptor url,
                                 SelectionKey key,
                                 InfiniteReadWriteByteBuffer initBuffer)
                          throws IOException

Creates a new TCPChannel object, and links it to url. Also logs the creation through Trace.log(String, String) using the "msg" tag. This method is synchronized with to avoid interrupts (which would close the channel) from registerChannel(SocketChannel, TCPChannel) and queueAWrite(SelectionKey).

Parameters:

url - The URL under which to register the new connection. (The URL holds the connection object, since it is unique to the agent, whether local or remote.)

channel - The channel.

Returns:

the newly-created TCPChannel object.

Throws:

IOException - if TCPChannel#TCPChannel(URLDescriptor, SelectionKey, SocketServerInterface) throws an exception.

closePort

public void closePort()

Method to close the socket server's open port and return it to the system.

Specified by:

closePort in interface SocketServerInterface

exit

public void exit()

This method is ignored. This socket server is meant to be global to the process, so it will continue until the process shuts down.

Specified by:

exit in interface SocketServerInterface

See Also:

SocketServerInterface.exit()

readyOpsToString

private static String readyOpsToString(int code)

Utility method to print out the OP codes in a more legible manner in the log messages.

Parameters:

code - The OP code

Returns:

A string version of the OP code set ("OP_ACCEPT ...").