Package org.ow2.joram.mom.amqp

Class AMQPConnectionListener

java.lang.Object

fr.dyade.aaa.common.Daemon

org.ow2.joram.mom.amqp.AMQPConnectionListener

All Implemented Interfaces:

Runnable

public class AMQPConnectionListener
extends Daemon

Listens to the TCP connections.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
(package private) class	AMQPConnectionListener.NetServerOut

Field Summary

Fields

Modifier and Type	Field	Description
static int	AMQP_FRAME_EXTRA_SIZE	The number of bytes that is not used for payload in an AMQP frame: all frames consist of a header (7 octets), a payload of arbitrary size, and a 'frame-end' octet
private int	channelMax
private int	cHeartbeat
(package private) boolean	closing
static Locale	JORAM_AMQP_LOCALE	The message locale that the server supports.
static int	JORAM_AMQP_MAX_CHANNELS	Specifies highest channel number that the server permits.
static int	JORAM_AMQP_MAX_FRAME_SIZE	The largest frame size that the server proposes for the connection, including frame header and endbyte.
static String	JORAM_AMQP_SECURITY	The security mechanism that the server supports.
static String	JORAM_AMQP_VERSION	The implementation version of the broker.
private static Logger	logger
private int	maxBodySize
static Map<String,Object>	MOM_PROPERTIES	This table provides a set of peer properties, used for identification, debugging, and general information.
private AMQPConnectionListener.NetServerOut	netServerOut
private static int	NO_CHANNEL
private Map<Integer,PublishRequest>	openChannel	Contains the opened channels.
private Queue	queueIn
private BlockingQueue	queueOut
private ServerSocket	serverSocket	The server socket listening to connections from the AMQP peer.
private int	sHeartbeat	Heartbeats requested by client and server, in seconds.
private Socket	sock	The socket used to listen.

Fields inherited from class fr.dyade.aaa.common.Daemon

canStop, group, logmon, priority, running, thread

Constructor Summary

Constructors

Constructor	Description
AMQPConnectionListener(ServerSocket serverSocket, int heartbeat)	Creates a new connection listener.

Method Summary

Modifier and Type	Method	Description
private void	acceptConnection()
protected void	close()	Releases any resources attached to this daemon.
private void	closeChannel(int channel)
private void	closeProxy()
private void	closeSocket()
private void	connectionException(int errorNumber, String message, int classId, int methodId)	Release channel resources and close it by sending a notification to the client.
private PublishRequest	createPublishRequest(int channel)
private void	doProcessBody(byte[] body, PublishRequest publishRequest, int channelNumber)	Process the content body.
private void	doProcessHeader(MarshallingHeader header, PublishRequest publishRequest, int channelNumber)	Process the content header.
private void	doProcessMethod(AbstractMarshallingMethod method, int channelNumber)
private static AMQP.Connection.Start	getConnectionStartMethod()	Creates a AMQP.Connection.Start method object.
private boolean	isChannelOpen(int channel)
private void	openChannel(int channel)
private void	process(Frame frame)	Proceed this frame.
private static void	readProtocolHeader(InputStream in)
private void	removePublishRequest(int channel)
void	run()
private void	sendMethodToPeer(AbstractMarshallingMethod method, int channelNumber)
private void	sendToProxy(AbstractMarshallingMethod method)
private void	sendToProxy(PublishRequest publishRequest)
protected void	shutdown()	Interrupts a thread that waits for long periods.
private void	tuneConnectionParameters(AMQP.Connection.TuneOk tuneOk)

Methods inherited from class fr.dyade.aaa.common.Daemon

finish, getName, interrupt, isCurrentThread, isRunning, setDaemon, setName, setPriority, setThreadGroup, start, stop, toString

Methods inherited from class java.lang.Object

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

Field Detail

logger

private static final Logger logger

AMQP_FRAME_EXTRA_SIZE

public static final int AMQP_FRAME_EXTRA_SIZE

The number of bytes that is not used for payload in an AMQP frame: all frames consist of a header (7 octets), a payload of arbitrary size, and a 'frame-end' octet

See Also:

Constant Field Values

JORAM_AMQP_VERSION

public static final String JORAM_AMQP_VERSION

The implementation version of the broker.

JORAM_AMQP_LOCALE

public static final Locale JORAM_AMQP_LOCALE

The message locale that the server supports. The locale defines the language in which the server will send reply texts.

JORAM_AMQP_SECURITY

public static final String JORAM_AMQP_SECURITY

The security mechanism that the server supports.

See Also:

Constant Field Values

JORAM_AMQP_MAX_CHANNELS

public static int JORAM_AMQP_MAX_CHANNELS

Specifies highest channel number that the server permits. Usable channel numbers are in the range 1..channelmax. Zero indicates no specified limit.

JORAM_AMQP_MAX_FRAME_SIZE

public static int JORAM_AMQP_MAX_FRAME_SIZE

The largest frame size that the server proposes for the connection, including frame header and endbyte. The client can negotiate a lower value. Zero means that the server does not impose any specific limit but may reject very large frames if it cannot allocate resources for them.

MOM_PROPERTIES

public static final Map<String,Object> MOM_PROPERTIES

This table provides a set of peer properties, used for identification, debugging, and general information.

NO_CHANNEL

private static final int NO_CHANNEL

See Also:

Constant Field Values

openChannel

private Map<Integer,PublishRequest> openChannel

Contains the opened channels.

serverSocket

private volatile ServerSocket serverSocket

The server socket listening to connections from the AMQP peer.

sock

private Socket sock

The socket used to listen.

sHeartbeat

private int sHeartbeat

Heartbeats requested by client and server, in seconds.

cHeartbeat

private int cHeartbeat

queueIn

private Queue queueIn

queueOut

private BlockingQueue queueOut

netServerOut

private AMQPConnectionListener.NetServerOut netServerOut

maxBodySize

private int maxBodySize

channelMax

private int channelMax

closing

volatile boolean closing

Constructor Detail

AMQPConnectionListener

public AMQPConnectionListener(ServerSocket serverSocket,
                              int heartbeat)
                       throws IOException

Creates a new connection listener.

Parameters:

serverSocket - the server socket to listen to

heartbeat - the socket timeout delay.

Throws:

IOException

Exception

Method Detail

run

public void run()

process

private void process(Frame frame)
              throws IOException,
                     ConnectionException

Proceed this frame.

Parameters:

frame -

Throws:

IOException

ConnectionException

connectionException

private void connectionException(int errorNumber,
                                 String message,
                                 int classId,
                                 int methodId)

Release channel resources and close it by sending a notification to the client.

closeProxy

private void closeProxy()

sendMethodToPeer

private void sendMethodToPeer(AbstractMarshallingMethod method,
                              int channelNumber)

doProcessMethod

private void doProcessMethod(AbstractMarshallingMethod method,
                             int channelNumber)
                      throws ConnectionException

Throws:

ConnectionException

tuneConnectionParameters

private void tuneConnectionParameters(AMQP.Connection.TuneOk tuneOk)
                               throws SyntaxErrorException

Throws:

SyntaxErrorException

doProcessHeader

private void doProcessHeader(MarshallingHeader header,
                             PublishRequest publishRequest,
                             int channelNumber)

Process the content header.

Parameters:

header -

channelNumber -

doProcessBody

private void doProcessBody(byte[] body,
                           PublishRequest publishRequest,
                           int channelNumber)
                    throws FrameErrorException

Process the content body.

Parameters:

body -

channelNumber -

Throws:

FrameErrorException

sendToProxy

private void sendToProxy(PublishRequest publishRequest)

sendToProxy

private void sendToProxy(AbstractMarshallingMethod method)

acceptConnection

private void acceptConnection()
                       throws AMQPException,
                              IOException

Throws:

AMQPException

IOException

readProtocolHeader

private static void readProtocolHeader(InputStream in)
                                throws IOException,
                                       FrameErrorException

Throws:

IOException

FrameErrorException

getConnectionStartMethod

private static AMQP.Connection.Start getConnectionStartMethod()
                                                       throws IOException

Creates a AMQP.Connection.Start method object.

Throws:

IOException

isChannelOpen

private boolean isChannelOpen(int channel)

openChannel

private void openChannel(int channel)

closeChannel

private void closeChannel(int channel)

createPublishRequest

private PublishRequest createPublishRequest(int channel)

removePublishRequest

private void removePublishRequest(int channel)

closeSocket

private void closeSocket()

shutdown

protected void shutdown()

Description copied from class: Daemon

Interrupts a thread that waits for long periods. In some cases, we must use application specific tricks. For example, if a thread is waiting on a known socket, we have to close the socket to cause the thread to return immediately. Unfortunately, there really isn't any technique that works in general.

Specified by:

shutdown in class Daemon

close

protected void close()

Description copied from class: Daemon

Releases any resources attached to this daemon. Be careful, its method should be called more than one time.

Specified by:

close in class Daemon