org.jumpi.impl.connector.tcp
Class TcpConnector

java.lang.Object
  |
  +--org.jumpi.impl.AbstractConnector
        |
        +--org.jumpi.impl.connector.tcp.TcpConnector

All Implemented Interfaces:

Component, Configurable, Connector, Manageable, java.lang.Runnable, Task, TcpAcceptListener, TimerListener

public class TcpConnector

extends AbstractConnector

implements TcpAcceptListener, Task, TimerListener

The TcpConnector provides unreliable messaging guarrantees over TCP. A TcpConnection is established per TcpDestination on send. TcpConnections are maintained in a connection cache and reused for subsequent sends. TcpConnections may be closed when the connection cache is full or when the connection timeout has expired without any send or receive activity taking place.

Each TcpConnection has a dedicated Thread which reads data continuously from the remote peer and dispatches the data to the TcpConnector. If data is read and there is no receive operation registered at the TcpConnector, then the data is discarded. Therefore the TcpConnector is not strictly reliable, eventhough TCP provides a certain amount of "reliability". True reliability can anyway only be provided by an acknowledgement functionality at a higher layer. This non blocking, lossy, reception property on the otherhand avoids distributed deadlock. If send or receive failure occurs, then the TcpConnection is closed and the matching operation is failed. Receives on wildcard destinations are used to match incoming data on any host, any port or the combination of both.

The syntax of a TcpDestination is:

<protocolname>://[<ipaddress>|*]:[<port>|*]

Wildcard character is used to match any host or port. There are a configurable maximum number of TcpConnections allowed at any one time. No new TcpConnections are established or accepted if this maximum is exceeded.

A TcpConnectionAcceptor accepts incoming connection requests continuously and notifies them to the TcpConnector. The TcpConnector will not accept new connections if the connection cache is full.

The TcpConnection cache is cleaned up after a configurable time interval has expired. Firstly timed-out TcpConnections are closed. These are connections which have seen no send or receive activity for a configurable interval, and are using thread resources. If the cache size is still exceeded after removing timed-out connections, then the least recently used connections are closed, until the number of active connections are less than the cache size.

Field Summary

Fields inherited from class org.jumpi.impl.AbstractConnector

cancelledRecvs_, cancelledSends_, controller_, joinFailInterval_, joinTestInterval_, jumpi_, maxRecvBacklog_, maxSendBacklog_, name_, protocol_, recvQueue_, recvTransformer_, sendQueue_, sendTransformer_, started_, stopping_, taskId_, url_format_, url_prefix_

Constructor Summary

TcpConnector()

Method Summary

boolean	acceptSocket(java.net.Socket sock) Notify the TcpConnector that a new client Socket connection has been established with the TcpConnectionAcceptor.
void	cancelRecv(Envelope env) Cancel the receive operation.
protected boolean	checkJoinCondition() The clean shutdown condition is reached when there are no pending send operations and both wildcard and specific receive Destination filters are empty, and when there are no TcpConnections active.
void	configure(java.lang.String name, Properties props) Configure the TcpConnector.
Destination	getDestination(java.lang.String url, java.util.Hashtable clientProps) Return a Destination corresponding to the Url if the Url matches the TcpDestination syntax.
void	handleTimerInterrupt() Schedule the TcpConnection cache cleanup task.
boolean	isSchedulable() The TcpConnector's Task is schedulable if there are pending send operations to be performed, or if there is a TcpConnection cache cleanup pending.
void	manage(Component root, Component parent, java.lang.String operation, java.util.Hashtable parameters) Perform a recursive management operation through the Jumpi component tree, using the parameters provided.
protected boolean	onRecv(Envelope env) Does nothing since there is no synchronous receiving.
protected boolean	onSend(Envelope env) Send a message to a TcpDestination over a TcpConnection.
void	recv(Envelope env) Receive a message from a TcpDestination.
void	run() Perform cache management if it has been scheduled since the last time this method has been called, and perform the pending send operations which have been buffered.
protected void	shutdownState() Clears the pending send operation queue and receive filters and closes all TcpConnections.

Methods inherited from class org.jumpi.impl.AbstractConnector

cancelSend, getName, getTaskId, isLongRunning, send

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.jumpi.spi.component.Task

getTaskId, isLongRunning

Constructor Detail

TcpConnector

public TcpConnector()

Method Detail

configure

public void configure(java.lang.String name,
                      Properties props)
               throws java.lang.Exception

Configure the TcpConnector.

The following configuration attributes are defined in addition to those of the superclass:

• cleanupinterval - integer[1000..MAX-INT] - optional[60000].

  The interval in milliseconds between TcpConnection cache cleanups.

• connectiontimeout - integer[1000..MAX-INT] - optional[30000].

  The TcpConnection timeout interval in milliseconds. A timed-out connection is always closed during cache cleanup.

• flowcontroltimeout - integer[-1..MAX-INT] - optional[60000].

  The maximum time in milliseconds that a TcpConnection can delay receiving subsequent messages from it's Socket because a matching receive is not posted. If a TcpConnection cannot pass its received message on to the TcpConnector within this time period, then the message is lost with a warning. The timeout should be greater than the maximum processing time for messages received to avoid message loss. -1 means infinite timeout, 0 - no timeout, positive is finite timeout.

• servermode - boolean - optional[false].

  If servermode is true then send operations do not automatically create TcpConnections to the TcpDestination addressed. Send operations are assumed to be replies to previous received messages, which should have already a TcpConnection established. Sends which do not have active TcpConnections associated to the Destination are failed immediately. If servermode is true, then the acceptor must be set also.

• cachesize - integer[1..MAX-INT] - optional[100].

  The size of the TcpConnection cache. The number of active connections is reduced to the cache size at each cleanup interval.

• acceptor - instance - optional.

  The acceptor subcomponent's name suffix. If servermode is true, then this must be set.

• maxmessagesize - integer[0..MAX-INT] - optional[10000000].

  The maximum message size allowed on send.

• maxconnections - integer[0..MAX-INT] - optional[200].

  The maximum number of active connections allowed.

• socketreadtimeout - integer[0..MAX-INT] - optional[30000].

  The socket read timeout value in milliseconds.

• socketlinger - integer[0..MAX-INT] - optional[0].

  The socket linger timeout value in milliseconds.

• socketnodelay - boolean - optional[false].

  The socket no delay value.

• socketkeepalive - boolean - optional[false].

  The socket keep alive value.

Specified by:

configure in interface Configurable

Overrides:

configure in class AbstractConnector

Parameters:

name - the fully qualified instance name.

props - the read-only properties to configure the instance with.

Throws:

java.lang.Exception - when configuration is unsuccessful.

checkJoinCondition

protected boolean checkJoinCondition()

The clean shutdown condition is reached when there are no pending send operations and both wildcard and specific receive Destination filters are empty, and when there are no TcpConnections active.

Specified by:

checkJoinCondition in class AbstractConnector

Returns:

true if there are no pending sends or receives or TcpConnections.

shutdownState

protected void shutdownState()

Clears the pending send operation queue and receive filters and closes all TcpConnections.

Specified by:

shutdownState in class AbstractConnector

manage

public void manage(Component root,
                   Component parent,
                   java.lang.String operation,
                   java.util.Hashtable parameters)
            throws java.lang.Exception

Perform a recursive management operation through the Jumpi component tree, using the parameters provided. Management operations are delegated also to the TcpConnectionAcceptor child component.

The following management actions are handled in addition to those of the superclass:

• STARTUP. Start the cleanup timer.
• SHUTDOWN. Stop the cleanup timer.

Specified by:

manage in interface Manageable

Overrides:

manage in class AbstractConnector

Parameters:

root - the Jumpi instance at the root of the component tree.

parent - the immediate parent of the component.

operation - the operation.

parameters - the operation parameters.

Throws:

java.lang.Exception - if any failure to perform management occurs.

isSchedulable

public boolean isSchedulable()

The TcpConnector's Task is schedulable if there are pending send operations to be performed, or if there is a TcpConnection cache cleanup pending.

Specified by:

isSchedulable in interface Task

Overrides:

isSchedulable in class AbstractConnector

Returns:

true if cache cleanup or send operations are pending, otherwise false.

handleTimerInterrupt

public void handleTimerInterrupt()

Schedule the TcpConnection cache cleanup task.

Specified by:

handleTimerInterrupt in interface TimerListener

See Also:

Timer.interruptAt(long, org.jumpi.spi.component.TimerListener), Timer.interruptAfterDelay(long, org.jumpi.spi.component.TimerListener), Timer.interruptPeriodically(long, org.jumpi.spi.component.TimerListener, boolean)

getDestination

public Destination getDestination(java.lang.String url,
                                  java.util.Hashtable clientProps)

Return a Destination corresponding to the Url if the Url matches the TcpDestination syntax. The empty Uri, i.e. <protocol>:// represents the TcpConnector's TcpConnectionAcceptor.

Specified by:

getDestination in interface Connector

Overrides:

getDestination in class AbstractConnector

Parameters:

url - specific or wildcard TcpDestination Url.

clientProps - not used.

Returns:

a Destination if the url matches the TcpDestination syntax, else null.

Throws:

java.lang.IllegalArgumentException - when url is null.

onSend

protected boolean onSend(Envelope env)
                  throws java.lang.Exception

Send a message to a TcpDestination over a TcpConnection. If a TcpConnection exists in the connection cache, then this is used to send the data. If there is no TcpConnection, and the servermode is not set, a TcpConnection is established to the remote peer.

The message format required for sending via the TcpConnection is a byte[]. If a send transformer is configured then this must map the input data to a byte[], otherwise the byte[] must be supplied by the Controller. The TcpConnection wraps the data to be sent into a packet. The data length must be less than the configured maximum for sending.

Allows multiple concurrent send operations to any Destinations. The sends are processes sequentially by the instance's task working off the buffered send queue.

Non blocking until the buffered send queue has filled, at which time send flow control prohibits further sending until there is space in the queue.

Specified by:

onSend in class AbstractConnector

Parameters:

env - the Envelope containing the Destination to send to and the message.

Returns:

true if successful, otherwise false on a managed failure.

Throws:

java.lang.Exception - if a failure condition occurs.

run

public void run()

Perform cache management if it has been scheduled since the last time this method has been called, and perform the pending send operations which have been buffered.

Specified by:

run in interface java.lang.Runnable

Overrides:

run in class AbstractConnector

recv

public void recv(Envelope env)
          throws java.lang.Exception

Receive a message from a TcpDestination. The TcpDestination to receive from is included in a filter. When a TcpConnection receives data and the TcpConnection's remote peer TcpDestination matches this entry, then the data will be passed up as the response to the receive operation. Wildcard receives may only be used to match the first message from a remote peer, since until a single message is received, the client layer does not know the sender's address. This is since the local port number of the sender is determined dynamically when establishing the TcpConnection when sending. A receive from a specific TcpDestination will fail immediately if there is no TcpConnection established to the remote peer. When a TcpConnection breaks and there is a receive operation pending for that remote peer, then the receive operation fails.

The format of the data provided by the TcpConnection is a byte[]. This is given directly as the received data to the Controller in absence of a receive transformer. If a receive transformer is configured, then this is given the byte[] and it gives the transformed message to the Controller layer.

Allows multiple concurrent receive operations to different Destinations but not the same Destination. There is no limit to the number of different Destinations which can simultaneously be received from.

Non blocking always.

Specified by:

recv in interface Connector

Overrides:

recv in class AbstractConnector

Parameters:

env - the Envelope containing the Destination to receive from.

Throws:

java.lang.Exception - if a failure condition occurs.

java.lang.IllegalArgumentException - if env is null.

java.lang.IllegalStateException - if not started or stopping.

See Also:

Controller.callbackRecv(org.jumpi.spi.Envelope, boolean, java.lang.String)

onRecv

protected boolean onRecv(Envelope env)
                  throws java.lang.Exception

Does nothing since there is no synchronous receiving.

Specified by:

onRecv in class AbstractConnector

Parameters:

env - the receive operation's envelope.

Returns:

true if successful, otherwise false on a managed failure.

Throws:

java.lang.Exception - never.

cancelRecv

public void cancelRecv(Envelope env)

Cancel the receive operation.

Specified by:

cancelRecv in interface Connector

Overrides:

cancelRecv in class AbstractConnector

Parameters:

env - the receive operation's envelope.

Throws:

java.lang.IllegalArgumentException - when env is null.

acceptSocket

public boolean acceptSocket(java.net.Socket sock)

Notify the TcpConnector that a new client Socket connection has been established with the TcpConnectionAcceptor. The TcpConnector instance must either take responsibility for the new socket, or tell the TcpConnectionAcceptor to close it.

The Socket cannot be accepted if there already is a TcpConnection which is connected to the same Destination as the Socket. This can only happen if the Socket reading thread of the TcpConnection has yet to return from a blocking read, by the connection has broken in reality and the remote peer reconnects. Assuming that no read then takes place, the TcpConnection will eventually be timed-out.

Specified by:

acceptSocket in interface TcpAcceptListener

Parameters:

sock - the accepted client Socket connection.

Returns:

true if all is ok - else false for the acceptor to close it.

Throws:

java.lang.IllegalArgumentException - if sock is null.