cvs commit: spice/sandbox/sca/src/java/org/realityforge/sca/connector ActivePinger.java AlwaysReconnectPolicy.java Connector.java ConnectorConnection.java ConnectorMonitor.java InactivityPingPolicy.java LimitingReconnectPolicy.java NeverPingPolicy.java NullMonitor.java PeriodicPingPolicy.java PingPolicy.java ReconnectionPolicy.java
Peter Donald <[email protected]>
| Newsgroups | gmane.comp.java.spice.cvs |
|---|---|
| Message-ID | <[email protected]> |
donaldp 03/11/04 17:11:59
Added: sandbox/sca/src/java/org/realityforge/sca/connector
ActivePinger.java AlwaysReconnectPolicy.java
Connector.java ConnectorConnection.java
ConnectorMonitor.java InactivityPingPolicy.java
LimitingReconnectPolicy.java NeverPingPolicy.java
NullMonitor.java PeriodicPingPolicy.java
PingPolicy.java ReconnectionPolicy.java
Log:
Migrate connector code into the sca toolkit
Revision Changes Path
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/ActivePinger.java
Index: ActivePinger.java
===================================================================
package org.realityforge.sca.connector;
/**
* A simple Runnable that checks ping for connector
* in a separate thread.
*/
public class ActivePinger
implements Runnable
{
/**
* The associated connector.
*/
private final Connector _connector;
/**
* The thread that pinger is running in.
*/
private Thread _thread;
/**
* Flag indicating whether the pinger has started.
*/
private boolean _started;
/**
* Flag indicating whether the pinger has ended.
*/
private boolean _ended;
/**
* Create pinger for specified connector.
*
* @param connector the connector
*/
public ActivePinger( final Connector connector )
{
if ( null == connector )
{
throw new NullPointerException( "connector" );
}
_connector = connector;
}
/**
* Deactivate the pinger and wait till
* it has stopped pinging.
*/
public synchronized void deactivate()
{
Thread thread = _thread;
_thread = null;
while ( !_ended )
{
thread.interrupt();
try
{
wait();
}
catch ( final InterruptedException ie )
{
//ignore
}
}
}
/**
* Return true if pinger has started.
*
* @return true if pinger has started.
*/
public synchronized boolean hasStarted()
{
return _started;
}
/**
* Main pinging loop.
*/
public void run()
{
synchronized ( this )
{
_started = true;
_thread = Thread.currentThread();
}
while ( null != _thread )
{
final long now = _connector.checkPing();
try
{
Thread.sleep( now );
}
catch ( final InterruptedException ie )
{
//Ignore and fall to through to isActive
}
}
synchronized ( this )
{
_ended = true;
notifyAll();
}
}
}
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/AlwaysReconnectPolicy.java
Index: AlwaysReconnectPolicy.java
===================================================================
package org.realityforge.sca.connector;
/**
* A Policy that will always attempt to reconnect
* regardless of how many failures or errors.
*/
public class AlwaysReconnectPolicy
implements ReconnectionPolicy
{
/**
* Constant containing instance of policy.
*/
public static final AlwaysReconnectPolicy POLICY = new AlwaysReconnectPolicy();
/**
* @see org.realityforge.sca.connector.ReconnectionPolicy#attemptConnection
*/
public boolean attemptConnection( final long lastConnectionAttempt,
final int connectionAttempts )
{
return true;
}
/**
* @see org.realityforge.sca.connector.ReconnectionPolicy#disconnectOnError
*/
public boolean disconnectOnError( final Throwable t )
{
return true;
}
/**
* @see org.realityforge.sca.connector.ReconnectionPolicy#reconnectOnDisconnect
*/
public boolean reconnectOnDisconnect()
{
return true;
}
}
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/Connector.java
Index: Connector.java
===================================================================
package org.realityforge.sca.connector;
/**
* The Connector is a base class for connectors.
* Connectors establish a connection to a resource
* and attempt to maintain the connection and reconnect
* when the connection fails.
*
* @mx.component
*/
public class Connector
{
/**
* The associated ping policy for connector.
*/
private PingPolicy _pingPolicy = NeverPingPolicy.POLICY;
/**
* The associated reconnection policy for connector.
*/
private ReconnectionPolicy _reconnectPolicy = AlwaysReconnectPolicy.POLICY;
/**
* The associated monitor that receives
* events about connector.
*/
private ConnectorMonitor _monitor = NullMonitor.MONITOR;
/**
* The underlying connection.
*/
private ConnectorConnection _connection;
/**
* A flag indicating whether the connection
* is "active".
*/
private boolean _active;
/**
* A flag indicating whether the connection
* is "connected".
*/
private boolean _connected;
/**
* Time at which last transmission occured.
*/
private long _lastTxTime;
/**
* The last message transmitted. May be null.
* Simply used to display status on the web page.
*/
private Object _lastTxMessage;
/**
* Time at which last receive occured.
*/
private long _lastRxTime;
/**
* The last message received. May be null.
* Simply used to display status on the web page.
*/
private Object _lastRxMessage;
/**
* The time the last connection attempt started.
*/
private long _lastConnectionTime;
/**
* Number of sequential failed connection
* attempts.
*/
private int _connectionAttempts;
/**
* The reason the last conenction attempt failed.
*/
private String _connectionError;
/**
* The time at which last ping occured.
*/
private long _lastPingTime;
/**
* Specify the ping policy that connector will use.
*
* @param pingPolicy the policy
*/
public void setPingPolicy( final PingPolicy pingPolicy )
{
if ( null == pingPolicy )
{
throw new NullPointerException( "pingPolicy" );
}
_pingPolicy = pingPolicy;
}
/**
* Specify the reconnection policy that connector will use.
*
* @param reconnectPolicy the policy
*/
public void setReconnectPolicy( final ReconnectionPolicy reconnectPolicy )
{
if ( null == reconnectPolicy )
{
throw new NullPointerException( "reconnectPolicy" );
}
_reconnectPolicy = reconnectPolicy;
}
/**
* Specify the connection that connector will manage.
*
* @param connection the connection
*/
public void setConnection( final ConnectorConnection connection )
{
if ( null == connection )
{
throw new NullPointerException( "connection" );
}
_connection = connection;
}
/**
* Specify the monitor to receive events from connector.
*
* @param monitor the monitor
*/
public void setMonitor( final ConnectorMonitor monitor )
{
if ( null == monitor )
{
throw new NullPointerException( "monitor" );
}
_monitor = monitor;
}
/**
* Method called to indicate transmission occured.
*
* @param message the message
*/
public void transmissionOccured( final Object message )
{
_lastTxTime = System.currentTimeMillis();
_lastTxMessage = message;
}
/**
* Method called to indicate receive occured.
*
* @param message the message
*/
public void receiveOccured( final Object message )
{
_lastRxTime = System.currentTimeMillis();
_lastRxMessage = message;
}
/**
* Method called to indicate bidirectional communication occured.
*
* @param message the message
*/
public void commOccured( final Object message )
{
final long now = System.currentTimeMillis();
_lastTxTime = now;
_lastRxTime = now;
_lastRxMessage = message;
_lastTxMessage = message;
}
/**
* Method called to failure to communicate.
*
* @param t the error
*/
public void commErrorOccured( final Throwable t )
{
if ( getReconnectPolicy().disconnectOnError( t ) )
{
disconnect();
if ( getReconnectPolicy().reconnectOnDisconnect() )
{
connect();
}
}
}
/**
* Return the time at which last ping occured.
*
* @return the time at which last ping occured.
* @mx.attribute
*/
public long getLastPingTime()
{
return _lastPingTime;
}
/**
* Return the time at which last transmission occured.
*
* @return the time at which last transmission occured.
* @mx.attribute
*/
public long getLastTxTime()
{
return _lastTxTime;
}
/**
* Return the last message transmitted.
*
* @return the last message transmitted.
* @mx.attribute
*/
public Object getLastTxMessage()
{
return _lastTxMessage;
}
/**
* Return the time at which last receive occured.
*
* @return the time at which last receive occured.
* @mx.attribute
*/
public long getLastRxTime()
{
return _lastRxTime;
}
/**
* Return the last message received.
*
* @return the last message received.
* @mx.attribute
*/
public Object getLastRxMessage()
{
return _lastRxMessage;
}
/**
* Return true if connector is active.
*
* @return true if connector is active.
* @mx.attribute
*/
public boolean isActive()
{
return _active;
}
/**
* Set the flag to indicate if the connector is active or inactive.
*
* @param active the flag to indicate if the connector is active or inactive.
* @mx.attribute
*/
public void setActive( final boolean active )
{
_active = active;
}
/**
* Return true if Connector connected.
*
* @return true if Connector connected.
* @mx.attribute
*/
public boolean isConnected()
{
return _connected;
}
/**
* Set the connected state.
*
* @param connected the connected state.
*/
protected void setConnected( final boolean connected )
{
_connected = connected;
}
/**
* Return the time the last connection attempt started.
*
* @return the time the last connection attempt started.
* @mx.attribute
*/
public long getLastConnectionTime()
{
return _lastConnectionTime;
}
/**
* Return the number of sequential failed connection attempts.
*
* @return the number of sequential failed connection attempts.
* @mx.attribute
*/
public int getConnectionAttempts()
{
return _connectionAttempts;
}
/**
* Return the last connection error.
* The error could be caused either by connection
* failure or failure during operation.
*
* @return the last connection error
* @mx.attribute
*/
public String getConnectionError()
{
return _connectionError;
}
/**
* Method to make connector establish connection.
*
* @mx.operation
*/
public void connect()
{
final long now = System.currentTimeMillis();
synchronized ( getSyncLock() )
{
if ( isConnected() )
{
disconnect();
}
while ( !isConnected() && isActive() )
{
if ( !getReconnectPolicy().attemptConnection( _lastConnectionTime,
_connectionAttempts ) )
{
getMonitor().skippingConnectionAttempt();
return;
}
getMonitor().attemptingConnection();
try
{
_lastConnectionTime = now;
getConnection().doConnect();
_lastPingTime = System.currentTimeMillis();
commOccured( null );
_connectionAttempts = 0;
_connectionError = null;
setConnected( true );
getMonitor().connectionEstablished();
}
catch ( final Throwable t )
{
_connectionAttempts++;
_connectionError = t.toString();
getMonitor().errorConnecting( t );
}
}
}
}
/**
* Method to disconect Connector.
*
* @mx.operation
*/
public void disconnect()
{
synchronized ( getSyncLock() )
{
if ( isConnected() )
{
getMonitor().attemptingDisconnection();
setConnected( false );
try
{
getConnection().doDisconnect();
}
catch ( final Throwable t )
{
getMonitor().errorDisconnecting( t );
}
}
}
}
/**
* Check to see if need to ping connection and
* if so then perform ping. Return the time
* that ping should be next checked at.
*
* @return the time that ping should be re-checked.
*/
public long checkPing()
{
final PingPolicy pingPolicy = getPingPolicy();
final boolean doPing = pingPolicy.checkPingConnection();
final long result = pingPolicy.nextPingCheck();
if ( doPing )
{
ping();
}
return result;
}
/**
* Attempt to verify Connector is connected.
* If not connected then the connector will attempt
* to establish a connection.
*
* @return true if connected
*/
public boolean verifyConnected()
{
synchronized ( getSyncLock() )
{
if ( !isConnected() )
{
connect();
}
return isConnected();
}
}
/**
* Attempt to ping connection.
* By default just calls {@link #validateConnection}.
*
* @return true if connected and ping successful.
*/
public boolean ping()
{
_lastPingTime = System.currentTimeMillis();
return validateConnection();
}
/**
* Attempt to verify Connector is connected.
* If not connected then the connector will attempt
* to establish a connection.
*
* @return true if connected
*/
public boolean validateConnection()
{
synchronized ( getSyncLock() )
{
getMonitor().attemptingValidation();
if ( !verifyConnected() )
{
return false;
}
else
{
doValidateConnection();
return isConnected();
}
}
}
/**
* Utility method that actully does the work
* of validating connection. If an error
* occurs the connection will be disconnected
* and the error recorded.
*/
void doValidateConnection()
{
try
{
getConnection().doValidateConnection();
}
catch ( final Throwable t )
{
_connectionError = t.toString();
getMonitor().errorValidatingConnection( t );
disconnect();
if ( getReconnectPolicy().reconnectOnDisconnect() )
{
connect();
}
}
}
/**
* Return the object that will be used to
* synchronization connection/disconnection.
*
* @return the sync lock
*/
protected Object getSyncLock()
{
return this;
}
/**
* Return the ping policy.
*
* @return the ping policy.
*/
protected PingPolicy getPingPolicy()
{
return _pingPolicy;
}
/**
* Return the reconnection policy.
*
* @return the reconnection policy.
*/
protected ReconnectionPolicy getReconnectPolicy()
{
return _reconnectPolicy;
}
/**
* Return the monitor.
*
* @return the monitor.
*/
protected ConnectorMonitor getMonitor()
{
return _monitor;
}
/**
* Return the connection.
*
* @return the connection
*/
protected ConnectorConnection getConnection()
{
if ( null == _connection )
{
throw new NullPointerException( "connection" );
}
return _connection;
}
}
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/ConnectorConnection.java
Index: ConnectorConnection.java
===================================================================
package org.realityforge.sca.connector;
/**
* Add in interface representing Connection
* managed by Connector.
*/
public interface ConnectorConnection
{
/**
* Establish the connection.
*
* @throws Exception if unable to connect
*/
void doConnect()
throws Exception;
/**
* Disconnect the connection.
*
* @throws Exception if unable to connect
*/
void doDisconnect()
throws Exception;
/**
* Validate the connection. The validation should
* involve explicitly testing that that the
* connection is valid.
*
* @throws Exception if unable to connection is not valid
*/
void doValidateConnection()
throws Exception;
}
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/ConnectorMonitor.java
Index: ConnectorMonitor.java
===================================================================
package org.realityforge.sca.connector;
/**
* The ConnectorMonitor gets notified of events
* in Connector.
*/
public interface ConnectorMonitor
{
/**
* Notify that connection attempt about to start.
*/
void attemptingConnection();
/**
* Notify monitor that connection has been established.
*/
void connectionEstablished();
/**
* Notify monitor that there was an error
* connecting.
*
* @param t the error
*/
void errorConnecting( Throwable t );
/**
* Notify monitor that there was an error
* disconnecting.
*
* @param t the error
*/
void errorDisconnecting( Throwable t );
/**
* Notify monito that attempting to
* validate connection.
*/
void attemptingValidation();
/**
* Notify monitor that there was an error
* validating connection. After this method
* is called the connection will be disconnected.
*/
void errorValidatingConnection( Throwable t );
/**
* Notify that the policy has indicated that
* a connection attempt should not be made at
* this point in time.
*/
void skippingConnectionAttempt();
/**
* Notify monitor that Connection is being
* disconnected.
*/
void attemptingDisconnection();
}
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/InactivityPingPolicy.java
Index: InactivityPingPolicy.java
===================================================================
package org.realityforge.sca.connector;
/**
* Ping policy that pings when Connector
* is innactive for a period of time.
*/
public class InactivityPingPolicy
implements PingPolicy
{
/**
* The period of transmission inactivity
* that will force a ping. -1 means
* that inactivitity not monitored for
* transmissions.
*/
private final long _txInactivity;
/**
* The period of retrieve inactivity
* that will force a ping. -1 means
* that inactivitity not monitored for
* receives.
*/
private final long _rxInactivity;
/**
* The associated connector.
*/
private final Connector _connector;
/**
* Create policy with specified periods.
*
* @param period the inactivity period for either receivees
* or transmissions
* @param connector the associated connector
*/
public InactivityPingPolicy( final long period,
final Connector connector )
{
this( period, period, connector );
}
/**
* Create policy with specified periods.
*
* @param txInactivity the inactivity period for transmissions
* @param rxInactivity the inactivity period for receives
* @param connector the associated connector
*/
public InactivityPingPolicy( final long txInactivity,
final long rxInactivity,
final Connector connector )
{
_txInactivity = txInactivity;
_rxInactivity = rxInactivity;
_connector = connector;
}
/**
* @see PingPolicy#checkPingConnection
*/
public boolean checkPingConnection()
{
final long now = System.currentTimeMillis();
if ( -1 != _txInactivity )
{
final long lastTxTime = _connector.getLastTxTime();
final long period = now - lastTxTime;
if ( period > _txInactivity )
{
return true;
}
}
if ( -1 != _rxInactivity )
{
final long lastRxTime = _connector.getLastRxTime();
final long period = now - lastRxTime;
if ( period > _rxInactivity )
{
return true;
}
}
return false;
}
/**
* @see PingPolicy#nextPingCheck
*/
public long nextPingCheck()
{
final long txPeriod;
if ( -1 != _txInactivity )
{
final long lastTxTime = _connector.getLastTxTime();
txPeriod = lastTxTime + _txInactivity;
}
else
{
txPeriod = Long.MAX_VALUE;
}
final long rxPeriod;
if ( -1 != _rxInactivity )
{
final long lastRxTime = _connector.getLastRxTime();
rxPeriod = lastRxTime + _rxInactivity;
}
else
{
rxPeriod = Long.MAX_VALUE;
}
return Math.min( txPeriod, rxPeriod );
}
}
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/LimitingReconnectPolicy.java
Index: LimitingReconnectPolicy.java
===================================================================
package org.realityforge.sca.connector;
/**
* A Policy that will always attempt to limit
* the number of connection attempts in a
* period of time.
*
* <p>After N connection attempts it will force a
* delay of T between successive connection
* attempts. This is an attempt not to overload
* the resource being connected to.</p>
*/
public class LimitingReconnectPolicy
implements ReconnectionPolicy
{
/**
* The number of attempts allowed before delay will kick in.
*/
private final int _attempts;
/**
* The delay between successive connection attempts.
*/
private final long _delay;
/**
* Create a policy instance.
*
* @param attempts the number of attempts before the delay is enabled.
* @param delay the delay
*/
public LimitingReconnectPolicy( final int attempts,
final long delay )
{
_attempts = attempts;
_delay = delay;
}
/**
* @see org.realityforge.sca.connector.ReconnectionPolicy#attemptConnection
*/
public boolean attemptConnection( final long lastConnectionAttempt,
final int connectionAttempts )
{
if( connectionAttempts >= _attempts )
{
final long now = System.currentTimeMillis();
final long nextAttempt = lastConnectionAttempt + _delay;
if( now < nextAttempt )
{
return false;
}
}
return true;
}
/**
* @see org.realityforge.sca.connector.ReconnectionPolicy#disconnectOnError
*/
public boolean disconnectOnError( final Throwable t )
{
return true;
}
/**
* @see org.realityforge.sca.connector.ReconnectionPolicy#reconnectOnDisconnect
*/
public boolean reconnectOnDisconnect()
{
return true;
}
}
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/NeverPingPolicy.java
Index: NeverPingPolicy.java
===================================================================
package org.realityforge.sca.connector;
/**
* Ping policy for never performing ping.
*/
public class NeverPingPolicy
implements PingPolicy
{
/**
* Constant containing instance of never ping policy.
*/
public static final NeverPingPolicy POLICY = new NeverPingPolicy();
/**
* @see PingPolicy#checkPingConnection
*/
public boolean checkPingConnection()
{
return false;
}
/**
* @see PingPolicy#nextPingCheck
*/
public long nextPingCheck()
{
return Long.MAX_VALUE;
}
}
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/NullMonitor.java
Index: NullMonitor.java
===================================================================
package org.realityforge.sca.connector;
/**
* A null monitor that consumes all messages.
*/
public class NullMonitor
implements ConnectorMonitor
{
/**
* A constant containing instance of null monitor.
*/
public static final NullMonitor MONITOR = new NullMonitor();
/**
* @see ConnectorMonitor#attemptingConnection
*/
public void attemptingConnection()
{
}
/**
* @see ConnectorMonitor#connectionEstablished
*/
public void connectionEstablished()
{
}
/**
* @see ConnectorMonitor#errorConnecting
*/
public void errorConnecting( final Throwable t )
{
}
/**
* @see ConnectorMonitor#errorDisconnecting
*/
public void errorDisconnecting( final Throwable t )
{
}
/**
* @see ConnectorMonitor#attemptingValidation
*/
public void attemptingValidation()
{
}
/**
* @see ConnectorMonitor#errorValidatingConnection
*/
public void errorValidatingConnection( final Throwable t )
{
}
/**
* @see ConnectorMonitor#skippingConnectionAttempt
*/
public void skippingConnectionAttempt()
{
}
/**
* @see ConnectorMonitor#attemptingDisconnection
*/
public void attemptingDisconnection()
{
}
}
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/PeriodicPingPolicy.java
Index: PeriodicPingPolicy.java
===================================================================
package org.realityforge.sca.connector;
/**
* Ping policy that pings on specified period.
*/
public class PeriodicPingPolicy
implements PingPolicy
{
/**
* The period between pings.
*/
private final long _period;
/**
* The associated connector.
*/
private final Connector _connector;
/**
* Create policy with specified period.
*
* @param period the period
* @param connector the associated connector
*/
public PeriodicPingPolicy( final long period,
final Connector connector )
{
_period = period;
_connector = connector;
}
/**
* @see PingPolicy#checkPingConnection
*/
public boolean checkPingConnection()
{
return true;
}
/**
* @see PingPolicy#nextPingCheck
*/
public long nextPingCheck()
{
return _connector.getLastPingTime() + _period;
}
}
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/PingPolicy.java
Index: PingPolicy.java
===================================================================
package org.realityforge.sca.connector;
/**
* Callback used to determine whether
* to ping a Connection.
*/
public interface PingPolicy
{
/**
* Return true if need to ping connection.
*
* @return true if need to ping connection.
*/
boolean checkPingConnection();
/**
* Return the time at which the next ping should occur.
* If the value {@link Long#MAX_VALUE} is returned
* then no pinging will ever occur.
*
* @return the time that ping should be checked
*/
long nextPingCheck();
}
1.1 spice/sandbox/sca/src/java/org/realityforge/sca/connector/ReconnectionPolicy.java
Index: ReconnectionPolicy.java
===================================================================
package org.realityforge.sca.connector;
/**
* Interface representing policy Connector uses to
* determine when to re-establish the connection.
*/
public interface ReconnectionPolicy
{
/**
* Return true to continue with connection attempt.
*
* @param connectionAttempts the number of sequential connection failures
* @param lastConnectionAttemptTime time at which connection was last attempted
* @return true to continue with connection attempt.
*/
boolean attemptConnection( long lastConnectionAttemptTime,
int connectionAttempts );
/**
* Return true to disconnect connection on specified error.
*
* @param t the error
* @return true to disconnect connection.
*/
boolean disconnectOnError( Throwable t );
/**
* Return true to reconnect on any forced disconnection.
*
* @return true to reconnect on any forced disconnection.
*/
boolean reconnectOnDisconnect();
}
-------------------------------------------------------
This SF.net email is sponsored by: SF.net Giveback Program.
Does SourceForge.net help you be more productive? Does it
help you create better code? SHARE THE LOVE, and help us help
YOU! Click Here: http://sourceforge.net/donate/