You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2012/11/25 18:15:46 UTC
svn commit: r1413376 - in /qpid/trunk/qpid/java:
broker/src/main/java/org/apache/qpid/server/protocol/
broker/src/main/java/org/apache/qpid/server/transport/
client/src/main/java/org/apache/qpid/client/
client/src/main/java/org/apache/qpid/client/proto...
Author: rgodfrey
Date: Sun Nov 25 17:15:43 2012
New Revision: 1413376
URL: http://svn.apache.org/viewvc?rev=1413376&view=rev
Log:
QPID-2796 : Implement hearbeating in Java Broker (all protocol versions) and Java Client (0-8/9/9-1 path)
Added:
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Ticker.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/TransportActivity.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java
qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/io/IdleTimeoutTickerTest.java
Modified:
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java Sun Nov 25 17:15:43 2012
@@ -33,6 +33,7 @@ import java.util.concurrent.ConcurrentHa
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -150,6 +151,8 @@ public class AMQProtocolEngine implement
private boolean _blocking;
private final Lock _receivedLock;
+ private AtomicLong _lastWriteTime = new AtomicLong(System.currentTimeMillis());
+
public AMQProtocolEngine(VirtualHostRegistry virtualHostRegistry, NetworkConnection network, final long connectionId)
{
@@ -541,7 +544,10 @@ public class AMQProtocolEngine implement
final ByteBuffer buf = asByteBuffer(frame);
_writtenBytes += buf.remaining();
_sender.send(buf);
- _lastIoTime = System.currentTimeMillis();
+ final long time = System.currentTimeMillis();
+ _lastIoTime = time;
+ _lastWriteTime.set(time);
+
if(!_deferFlush)
{
_sender.flush();
@@ -1058,12 +1064,12 @@ public class AMQProtocolEngine implement
public void readerIdle()
{
- // Nothing
+ // TODO - enforce disconnect on lack of inbound data
}
public synchronized void writerIdle()
{
- _sender.send(asByteBuffer(HeartbeatBody.FRAME));
+ writeFrame(HeartbeatBody.FRAME);
}
public void exception(Throwable throwable)
@@ -1461,4 +1467,16 @@ public class AMQProtocolEngine implement
{
return _receivedLock;
}
+
+ @Override
+ public long getLastReadTime()
+ {
+ return _lastReceivedTime;
+ }
+
+ @Override
+ public long getLastWriteTime()
+ {
+ return _lastWriteTime.get();
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java Sun Nov 25 17:15:43 2012
@@ -217,6 +217,18 @@ public class MultiVersionProtocolEngine
_sender = sender;
}
+ @Override
+ public long getLastReadTime()
+ {
+ return _delegate.getLastReadTime();
+ }
+
+ @Override
+ public long getLastWriteTime()
+ {
+ return _delegate.getLastWriteTime();
+ }
+
private static interface DelegateCreator
{
@@ -409,6 +421,18 @@ public class MultiVersionProtocolEngine
}
+ @Override
+ public long getLastReadTime()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getLastWriteTime()
+ {
+ return 0;
+ }
+
public long getConnectionId()
{
return _id;
@@ -566,5 +590,17 @@ public class MultiVersionProtocolEngine
{
}
+
+ @Override
+ public long getLastReadTime()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getLastWriteTime()
+ {
+ return 0;
+ }
}
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java Sun Nov 25 17:15:43 2012
@@ -44,6 +44,8 @@ public class ProtocolEngine_0_10 extend
private ServerConnection _connection;
private final IApplicationRegistry _appRegistry;
private long _createTime = System.currentTimeMillis();
+ private long _lastReadTime;
+ private long _lastWriteTime;
public ProtocolEngine_0_10(ServerConnection conn,
NetworkConnection network,
@@ -71,13 +73,61 @@ public class ProtocolEngine_0_10 extend
{
_network = network;
- _connection.setSender(new Disassembler(sender, MAX_FRAME_SIZE));
+ _connection.setNetworkConnection(network);
+ _connection.setSender(new Disassembler(wrapSender(sender), MAX_FRAME_SIZE));
_connection.setPeerPrincipal(_network.getPeerPrincipal());
// FIXME Two log messages to maintain compatibility with earlier protocol versions
_connection.getLogActor().message(ConnectionMessages.OPEN(null, null, null, false, false, false));
_connection.getLogActor().message(ConnectionMessages.OPEN(null, "0-10", null, false, true, false));
}
+ private Sender<ByteBuffer> wrapSender(final Sender<ByteBuffer> sender)
+ {
+ return new Sender<ByteBuffer>()
+ {
+ @Override
+ public void setIdleTimeout(int i)
+ {
+ sender.setIdleTimeout(i);
+
+ }
+
+ @Override
+ public void send(ByteBuffer msg)
+ {
+ _lastWriteTime = System.currentTimeMillis();
+ sender.send(msg);
+
+ }
+
+ @Override
+ public void flush()
+ {
+ sender.flush();
+
+ }
+
+ @Override
+ public void close()
+ {
+ sender.close();
+
+ }
+ };
+ }
+
+ @Override
+ public long getLastReadTime()
+ {
+ return _lastReadTime;
+ }
+
+ @Override
+ public long getLastWriteTime()
+ {
+ return _lastWriteTime;
+ }
+
public SocketAddress getRemoteAddress()
{
return _network.getRemoteAddress();
@@ -90,6 +140,7 @@ public class ProtocolEngine_0_10 extend
public void received(final ByteBuffer buf)
{
+ _lastReadTime = System.currentTimeMillis();
super.received(buf);
_connection.receivedComplete();
}
@@ -106,7 +157,7 @@ public class ProtocolEngine_0_10 extend
public void writerIdle()
{
- //Todo
+ _connection.doHeartbeat();
}
public void readerIdle()
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java Sun Nov 25 17:15:43 2012
@@ -53,6 +53,8 @@ public class ProtocolEngine_1_0_0 implem
//private NetworkConnection _networkDriver;
private long _readBytes;
private long _writtenBytes;
+ private long _lastReadTime;
+ private long _lastWriteTime;
private final IApplicationRegistry _appRegistry;
private long _createTime = System.currentTimeMillis();
private ConnectionEndpoint _conn;
@@ -178,6 +180,7 @@ public class ProtocolEngine_1_0_0 implem
public synchronized void received(ByteBuffer msg)
{
+ _lastReadTime = System.currentTimeMillis();
if(RAW_LOGGER.isLoggable(Level.FINE))
{
ByteBuffer dup = msg.duplicate();
@@ -320,6 +323,7 @@ public class ProtocolEngine_1_0_0 implem
synchronized(_sendLock)
{
+ _lastWriteTime = System.currentTimeMillis();
if(FRAME_LOGGER.isLoggable(Level.FINE))
{
FRAME_LOGGER.fine("SEND[" + getRemoteAddress() + "|" + amqFrame.getChannel() + "] : " + amqFrame.getFrameBody());
@@ -374,4 +378,13 @@ public class ProtocolEngine_1_0_0 implem
return _connectionId;
}
+ public long getLastReadTime()
+ {
+ return _lastReadTime;
+ }
+
+ public long getLastWriteTime()
+ {
+ return _lastWriteTime;
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java Sun Nov 25 17:15:43 2012
@@ -51,6 +51,9 @@ public class ProtocolEngine_1_0_0_SASL i
{
private long _readBytes;
private long _writtenBytes;
+
+ private long _lastReadTime;
+ private long _lastWriteTime;
private final IApplicationRegistry _appRegistry;
private long _createTime = System.currentTimeMillis();
private ConnectionEndpoint _conn;
@@ -222,6 +225,7 @@ public class ProtocolEngine_1_0_0_SASL i
public synchronized void received(ByteBuffer msg)
{
+ _lastReadTime = System.currentTimeMillis();
if(RAW_LOGGER.isLoggable(Level.FINE))
{
ByteBuffer dup = msg.duplicate();
@@ -364,7 +368,7 @@ public class ProtocolEngine_1_0_0_SASL i
synchronized(_sendLock)
{
-
+ _lastWriteTime = System.currentTimeMillis();
if(FRAME_LOGGER.isLoggable(Level.FINE))
{
FRAME_LOGGER.fine("SEND[" + getRemoteAddress() + "|" + amqFrame.getChannel() + "] : " + amqFrame.getFrameBody());
@@ -425,4 +429,13 @@ public class ProtocolEngine_1_0_0_SASL i
return _connectionId;
}
+ public long getLastReadTime()
+ {
+ return _lastReadTime;
+ }
+
+ public long getLastWriteTime()
+ {
+ return _lastWriteTime;
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java Sun Nov 25 17:15:43 2012
@@ -48,6 +48,7 @@ import org.apache.qpid.transport.Executi
import org.apache.qpid.transport.Method;
import org.apache.qpid.transport.ProtocolEvent;
import org.apache.qpid.transport.Session;
+import org.apache.qpid.transport.network.NetworkConnection;
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CONNECTION_FORMAT;
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SOCKET_FORMAT;
@@ -68,6 +69,7 @@ public class ServerConnection extends Co
private AtomicLong _lastIoTime = new AtomicLong();
private boolean _blocking;
private Principal _peerPrincipal;
+ private NetworkConnection _networkConnection;
public ServerConnection(final long connectionId)
{
@@ -490,4 +492,20 @@ public class ServerConnection extends Co
{
super.setLocalAddress(localAddress);
}
+
+ public void setNetworkConnection(NetworkConnection network)
+ {
+ _networkConnection = network;
+ }
+
+ public NetworkConnection getNetworkConnection()
+ {
+ return _networkConnection;
+ }
+
+ public void doHeartbeat()
+ {
+ super.doHeartBeat();
+
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java Sun Nov 25 17:15:43 2012
@@ -43,6 +43,8 @@ import org.apache.qpid.server.subscripti
import org.apache.qpid.server.virtualhost.State;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.transport.*;
+import org.apache.qpid.transport.network.NetworkConnection;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -226,14 +228,18 @@ public class ServerConnectionDelegate ex
return;
}
- setConnectionTuneOkChannelMax(sconn, okChannelMax);
- }
+ if(ok.hasHeartbeat())
+ {
+ final int heartbeat = ok.getHeartbeat();
+ if(heartbeat > 0)
+ {
+ final NetworkConnection networkConnection = sconn.getNetworkConnection();
+ networkConnection.setMaxReadIdle(2 * heartbeat);
+ networkConnection.setMaxWriteIdle(heartbeat);
+ }
+ }
- @Override
- protected int getHeartbeatMax()
- {
- //TODO: implement broker support for actually sending heartbeats
- return 0;
+ setConnectionTuneOkChannelMax(sconn, okChannelMax);
}
@Override
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java Sun Nov 25 17:15:43 2012
@@ -484,10 +484,14 @@ public class AMQConnectionDelegate_0_10
heartbeat = Integer.getInteger(ClientProperties.IDLE_TIMEOUT_PROP_NAME)/1000;
_logger.warn("JVM arg -Didle_timeout=<mili_secs> is deprecated, please use -Dqpid.heartbeat=<secs>");
}
- else
+ else if(Integer.getInteger(ClientProperties.HEARTBEAT) != null)
{
heartbeat = Integer.getInteger(ClientProperties.HEARTBEAT,ClientProperties.HEARTBEAT_DEFAULT);
}
+ else
+ {
+ heartbeat = Integer.getInteger("amqj.heartbeat.delay", ClientProperties.HEARTBEAT_DEFAULT);
+ }
return heartbeat;
}
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java Sun Nov 25 17:15:43 2012
@@ -122,7 +122,9 @@ public class AMQConnectionDelegate_8_0 i
SecurityLayer securityLayer = SecurityLayerFactory.newInstance(settings);
OutgoingNetworkTransport transport = Transport.getOutgoingTransportInstance(getProtocolVersion());
- NetworkConnection network = transport.connect(settings, securityLayer.receiver(_conn.getProtocolHandler()));
+
+ NetworkConnection network = transport.connect(settings, securityLayer.receiver(_conn.getProtocolHandler()),
+ _conn.getProtocolHandler());
_conn.getProtocolHandler().setNetworkConnection(network, securityLayer.sender(network.getSender()));
StateWaiter waiter = _conn.getProtocolHandler().createWaiter(openOrClosedStates);
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java Sun Nov 25 17:15:43 2012
@@ -178,6 +178,8 @@ public class AMQProtocolHandler implemen
private NetworkConnection _network;
private Sender<ByteBuffer> _sender;
+ private long _lastReadTime = System.currentTimeMillis();
+ private long _lastWriteTime = System.currentTimeMillis();
/**
* Creates a new protocol handler, associated with the specified client connection instance.
@@ -442,6 +444,7 @@ public class AMQProtocolHandler implemen
public void received(ByteBuffer msg)
{
_readBytes += msg.remaining();
+ _lastReadTime = System.currentTimeMillis();
try
{
final ArrayList<AMQDataBlock> dataBlocks = _codecFactory.getDecoder().decodeBuffer(msg);
@@ -560,6 +563,7 @@ public class AMQProtocolHandler implemen
public synchronized void writeFrame(AMQDataBlock frame, boolean flush)
{
final ByteBuffer buf = asByteBuffer(frame);
+ _lastWriteTime = System.currentTimeMillis();
_writtenBytes += buf.remaining();
_sender.send(buf);
if(flush)
@@ -882,6 +886,18 @@ public class AMQProtocolHandler implemen
_sender = sender;
}
+ @Override
+ public long getLastReadTime()
+ {
+ return _lastReadTime;
+ }
+
+ @Override
+ public long getLastWriteTime()
+ {
+ return _lastWriteTime;
+ }
+
protected Sender<ByteBuffer> getSender()
{
return _sender;
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java Sun Nov 25 17:15:43 2012
@@ -23,6 +23,7 @@ package org.apache.qpid.protocol;
import org.apache.qpid.transport.Receiver;
import org.apache.qpid.transport.Sender;
import org.apache.qpid.transport.network.NetworkConnection;
+import org.apache.qpid.transport.network.TransportActivity;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
@@ -31,7 +32,7 @@ import java.nio.ByteBuffer;
* A ProtocolEngine is a Receiver for java.nio.ByteBuffers. It takes the data passed to it in the received
* decodes it and then process the result.
*/
-public interface ProtocolEngine extends Receiver<java.nio.ByteBuffer>
+public interface ProtocolEngine extends Receiver<java.nio.ByteBuffer>, TransportActivity
{
// Returns the remote address of the NetworkDriver
SocketAddress getRemoteAddress();
@@ -56,6 +57,6 @@ public interface ProtocolEngine extends
void readerIdle();
- public void setNetworkConnection(NetworkConnection network, Sender<ByteBuffer> sender);
+ public void setNetworkConnection(NetworkConnection network, Sender<ByteBuffer> sender);
}
\ No newline at end of file
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java Sun Nov 25 17:15:43 2012
@@ -21,12 +21,7 @@
package org.apache.qpid.transport;
import org.apache.qpid.framing.ProtocolVersion;
-import org.apache.qpid.transport.network.Assembler;
-import org.apache.qpid.transport.network.Disassembler;
-import org.apache.qpid.transport.network.InputHandler;
-import org.apache.qpid.transport.network.NetworkConnection;
-import org.apache.qpid.transport.network.OutgoingNetworkTransport;
-import org.apache.qpid.transport.network.Transport;
+import org.apache.qpid.transport.network.*;
import org.apache.qpid.transport.network.security.SecurityLayer;
import org.apache.qpid.transport.network.security.SecurityLayerFactory;
import org.apache.qpid.transport.util.Logger;
@@ -73,6 +68,9 @@ public class Connection extends Connecti
//Usable channels are numbered 0 to <ChannelMax> - 1
public static final int MAX_CHANNEL_MAX = 0xFFFF;
public static final int MIN_USABLE_CHANNEL_NUM = 0;
+ private long _lastSendTime;
+ private long _lastReadTime;
+
public enum State { NEW, CLOSED, OPENING, OPEN, CLOSING, CLOSE_RCVD, RESUMING }
@@ -231,7 +229,8 @@ public class Connection extends Connecti
addConnectionListener((ConnectionListener)secureReceiver);
}
- NetworkConnection network = transport.connect(settings, secureReceiver);
+ NetworkConnection network = transport.connect(settings, secureReceiver, null);
+
setRemoteAddress(network.getRemoteAddress());
setLocalAddress(network.getLocalAddress());
@@ -368,6 +367,7 @@ public class Connection extends Connecti
public void received(ProtocolEvent event)
{
+ _lastReadTime = System.currentTimeMillis();
if(log.isDebugEnabled())
{
log.debug("RECV: [%s] %s", this, event);
@@ -377,6 +377,7 @@ public class Connection extends Connecti
public void send(ProtocolEvent event)
{
+ _lastSendTime = System.currentTimeMillis();
if(log.isDebugEnabled())
{
log.debug("SEND: [%s] %s", this, event);
@@ -745,4 +746,38 @@ public class Connection extends Connecti
sessionDetached.setCode(sessionDetachCode);
invoke(sessionDetached);
}
+
+
+ protected void doHeartBeat()
+ {
+ connectionHeartbeat();
+ }
+
+ private class ConnectionActivity implements TransportActivity
+ {
+ @Override
+ public long getLastReadTime()
+ {
+ return _lastReadTime;
+ }
+
+ @Override
+ public long getLastWriteTime()
+ {
+ return _lastSendTime;
+ }
+
+ @Override
+ public void writerIdle()
+ {
+ connectionHeartbeat();
+ }
+
+ @Override
+ public void readerIdle()
+ {
+ // TODO
+
+ }
+ }
}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java Sun Nov 25 17:15:43 2012
@@ -27,5 +27,7 @@ import javax.net.ssl.SSLContext;
public interface IncomingNetworkTransport extends NetworkTransport
{
- public void accept(NetworkTransportConfiguration config, ProtocolEngineFactory factory, SSLContext sslContext);
+ public void accept(NetworkTransportConfiguration config,
+ ProtocolEngineFactory factory,
+ SSLContext sslContext);
}
\ No newline at end of file
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java Sun Nov 25 17:15:43 2012
@@ -50,4 +50,8 @@ public interface NetworkConnection
void setPeerPrincipal(Principal principal);
Principal getPeerPrincipal();
+
+ int getMaxReadIdle();
+
+ int getMaxWriteIdle();
}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java Sun Nov 25 17:15:43 2012
@@ -29,5 +29,7 @@ public interface OutgoingNetworkTranspor
{
public NetworkConnection getConnection();
- public NetworkConnection connect(ConnectionSettings settings, Receiver<ByteBuffer> delegate);
+ public NetworkConnection connect(ConnectionSettings settings,
+ Receiver<ByteBuffer> delegate,
+ TransportActivity transportActivity);
}
\ No newline at end of file
Added: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Ticker.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Ticker.java?rev=1413376&view=auto
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Ticker.java (added)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Ticker.java Sun Nov 25 17:15:43 2012
@@ -0,0 +1,29 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.transport.network;
+
+public interface Ticker
+{
+ int getTimeToNextTick(long currentTime);
+
+ int tick(long currentTime);
+}
Added: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/TransportActivity.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/TransportActivity.java?rev=1413376&view=auto
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/TransportActivity.java (added)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/TransportActivity.java Sun Nov 25 17:15:43 2012
@@ -0,0 +1,33 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.transport.network;
+
+public interface TransportActivity
+{
+ long getLastReadTime();
+
+ long getLastWriteTime();
+
+ void writerIdle();
+
+ void readerIdle();
+}
Added: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java?rev=1413376&view=auto
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java (added)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java Sun Nov 25 17:15:43 2012
@@ -0,0 +1,87 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.transport.network.io;
+
+import org.apache.qpid.transport.network.NetworkConnection;
+import org.apache.qpid.transport.network.Ticker;
+import org.apache.qpid.transport.network.TransportActivity;
+
+class IdleTimeoutTicker implements Ticker
+{
+ private final TransportActivity _transport;
+ private final int _defaultTimeout;
+ private NetworkConnection _connection;
+
+ public IdleTimeoutTicker(TransportActivity transport, int defaultTimeout)
+ {
+ _transport = transport;
+ _defaultTimeout = defaultTimeout;
+ }
+
+ @Override
+ public int getTimeToNextTick(long currentTime)
+ {
+ long nextTime = -1;
+ final long maxReadIdle = 1000l * _connection.getMaxReadIdle();
+
+ if(maxReadIdle != 0)
+ {
+ nextTime = _transport.getLastReadTime() + maxReadIdle;
+ }
+
+ long maxWriteIdle = 1000l * _connection.getMaxWriteIdle();
+
+ if(maxWriteIdle != 0)
+ {
+ long writeTime = _transport.getLastWriteTime() + maxWriteIdle;
+ if(nextTime == -1l || writeTime < nextTime)
+ {
+ nextTime = writeTime;
+ }
+ }
+ return nextTime == -1 ? _defaultTimeout : (int) (nextTime - currentTime);
+ }
+
+ @Override
+ public int tick(long currentTime)
+ {
+ // writer Idle
+ long maxWriteIdle = 1000l * _connection.getMaxWriteIdle();
+ if(maxWriteIdle != 0 && maxWriteIdle+ _transport.getLastWriteTime() <= currentTime)
+ {
+ _transport.writerIdle();
+ }
+ // reader Idle
+ final long maxReadIdle = 1000l * _connection.getMaxReadIdle();
+ if(maxReadIdle != 0 && maxReadIdle+ _transport.getLastReadTime() <= currentTime)
+ {
+
+ _transport.readerIdle();
+ }
+ return getTimeToNextTick(currentTime);
+ }
+
+ public void setConnection(NetworkConnection connection)
+ {
+ _connection = connection;
+ }
+}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java Sun Nov 25 17:15:43 2012
@@ -26,7 +26,9 @@ import java.nio.ByteBuffer;
import java.security.Principal;
import org.apache.qpid.transport.Receiver;
import org.apache.qpid.transport.Sender;
+import org.apache.qpid.transport.network.Ticker;
import org.apache.qpid.transport.network.NetworkConnection;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,14 +40,23 @@ public class IoNetworkConnection impleme
private final IoSender _ioSender;
private final IoReceiver _ioReceiver;
private Principal _principal;
+ private int _maxReadIdle;
+ private int _maxWriteIdle;
public IoNetworkConnection(Socket socket, Receiver<ByteBuffer> delegate,
- int sendBufferSize, int receiveBufferSize, long timeout)
+ int sendBufferSize, int receiveBufferSize, long timeout)
+ {
+ this(socket,delegate,sendBufferSize,receiveBufferSize,timeout,null);
+ }
+
+ public IoNetworkConnection(Socket socket, Receiver<ByteBuffer> delegate,
+ int sendBufferSize, int receiveBufferSize, long timeout, Ticker ticker)
{
_socket = socket;
_timeout = timeout;
_ioReceiver = new IoReceiver(_socket, delegate, receiveBufferSize,_timeout);
+ _ioReceiver.setTicker(ticker);
_ioSender = new IoSender(_socket, 2 * sendBufferSize, _timeout);
@@ -88,14 +99,12 @@ public class IoNetworkConnection impleme
public void setMaxWriteIdle(int sec)
{
- // TODO implement support for setting heartbeating config in this way
- // Currently a socket timeout is used in IoSender
+ _maxWriteIdle = sec;
}
public void setMaxReadIdle(int sec)
{
- // TODO implement support for setting heartbeating config in this way
- // Currently a socket timeout is used in IoSender
+ _maxReadIdle = sec;
}
@Override
@@ -109,4 +118,16 @@ public class IoNetworkConnection impleme
{
return _principal;
}
+
+ @Override
+ public int getMaxReadIdle()
+ {
+ return _maxReadIdle;
+ }
+
+ @Override
+ public int getMaxWriteIdle()
+ {
+ return _maxWriteIdle;
+ }
}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java Sun Nov 25 17:15:43 2012
@@ -41,9 +41,8 @@ import org.apache.qpid.transport.Connect
import org.apache.qpid.transport.NetworkTransportConfiguration;
import org.apache.qpid.transport.Receiver;
import org.apache.qpid.transport.TransportException;
-import org.apache.qpid.transport.network.IncomingNetworkTransport;
-import org.apache.qpid.transport.network.NetworkConnection;
-import org.apache.qpid.transport.network.OutgoingNetworkTransport;
+import org.apache.qpid.transport.network.*;
+
import org.slf4j.LoggerFactory;
public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNetworkTransport
@@ -56,7 +55,9 @@ public class IoNetworkTransport implemen
private IoNetworkConnection _connection;
private AcceptingThread _acceptor;
- public NetworkConnection connect(ConnectionSettings settings, Receiver<ByteBuffer> delegate)
+ public NetworkConnection connect(ConnectionSettings settings,
+ Receiver<ByteBuffer> delegate,
+ TransportActivity transportActivity)
{
int sendBufferSize = settings.getWriteBufferSize();
int receiveBufferSize = settings.getReadBufferSize();
@@ -91,7 +92,9 @@ public class IoNetworkTransport implemen
try
{
- _connection = new IoNetworkConnection(_socket, delegate, sendBufferSize, receiveBufferSize, TIMEOUT);
+ IdleTimeoutTicker ticker = new IdleTimeoutTicker(transportActivity, TIMEOUT);
+ _connection = new IoNetworkConnection(_socket, delegate, sendBufferSize, receiveBufferSize, TIMEOUT, ticker);
+ ticker.setConnection(_connection);
_connection.start();
}
catch(Exception e)
@@ -128,7 +131,9 @@ public class IoNetworkTransport implemen
return _connection;
}
- public void accept(NetworkTransportConfiguration config, ProtocolEngineFactory factory, SSLContext sslContext)
+ public void accept(NetworkTransportConfiguration config,
+ ProtocolEngineFactory factory,
+ SSLContext sslContext)
{
try
{
@@ -149,6 +154,7 @@ public class IoNetworkTransport implemen
private ProtocolEngineFactory _factory;
private SSLContext _sslContext;
private ServerSocket _serverSocket;
+ private int _timeout;
private AcceptingThread(NetworkTransportConfiguration config,
ProtocolEngineFactory factory,
@@ -157,6 +163,7 @@ public class IoNetworkTransport implemen
_config = config;
_factory = factory;
_sslContext = sslContext;
+ _timeout = TIMEOUT;
InetSocketAddress address = config.getAddress();
@@ -217,6 +224,7 @@ public class IoNetworkTransport implemen
{
socket = _serverSocket.accept();
socket.setTcpNoDelay(_config.getTcpNoDelay());
+ socket.setSoTimeout(_timeout);
final Integer sendBufferSize = _config.getSendBufferSize();
final Integer receiveBufferSize = _config.getReceiveBufferSize();
@@ -226,7 +234,10 @@ public class IoNetworkTransport implemen
ProtocolEngine engine = _factory.newProtocolEngine();
- NetworkConnection connection = new IoNetworkConnection(socket, engine, sendBufferSize, receiveBufferSize, TIMEOUT);
+ final IdleTimeoutTicker ticker = new IdleTimeoutTicker(engine, TIMEOUT);
+ NetworkConnection connection = new IoNetworkConnection(socket, engine, sendBufferSize, receiveBufferSize, _timeout,
+ ticker);
+ ticker.setConnection(connection);
if(_sslContext != null)
{
@@ -293,6 +304,7 @@ public class IoNetworkTransport implemen
}
}
}
+
}
}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java Sun Nov 25 17:15:43 2012
@@ -24,6 +24,7 @@ import org.apache.qpid.common.Closeable;
import org.apache.qpid.thread.Threading;
import org.apache.qpid.transport.Receiver;
import org.apache.qpid.transport.TransportException;
+import org.apache.qpid.transport.network.Ticker;
import org.apache.qpid.transport.util.Logger;
import javax.net.ssl.SSLSocket;
@@ -31,6 +32,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.net.SocketException;
+import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -51,6 +53,8 @@ final class IoReceiver implements Runnab
private final AtomicBoolean closed = new AtomicBoolean(false);
private final Thread receiverThread;
private static final boolean shutdownBroken;
+
+ private Ticker _ticker;
static
{
String osName = System.getProperty("os.name");
@@ -136,7 +140,7 @@ final class IoReceiver implements Runnab
{
final int threshold = bufferSize / 2;
- // I set the read buffer size simillar to SO_RCVBUF
+ // I set the read buffer size similar to SO_RCVBUF
// Haven't tested with a lower value to see if it's better or worse
byte[] buffer = new byte[bufferSize];
try
@@ -144,17 +148,60 @@ final class IoReceiver implements Runnab
InputStream in = socket.getInputStream();
int read = 0;
int offset = 0;
- while ((read = in.read(buffer, offset, bufferSize-offset)) != -1)
+ long currentTime;
+ while(read != -1)
{
- if (read > 0)
+ try
{
- ByteBuffer b = ByteBuffer.wrap(buffer,offset,read);
- receiver.received(b);
- offset+=read;
- if (offset > threshold)
+ while ((read = in.read(buffer, offset, bufferSize-offset)) != -1)
{
- offset = 0;
- buffer = new byte[bufferSize];
+ if (read > 0)
+ {
+ ByteBuffer b = ByteBuffer.wrap(buffer,offset,read);
+ receiver.received(b);
+ offset+=read;
+ if (offset > threshold)
+ {
+ offset = 0;
+ buffer = new byte[bufferSize];
+ }
+ }
+ currentTime = System.currentTimeMillis();
+
+ if(_ticker != null)
+ {
+ final int tick = _ticker.getTimeToNextTick(currentTime);
+ try
+ {
+ if(!socket.isClosed())
+ {
+ socket.setSoTimeout(tick <= 0 ? 1 : tick);
+ }
+ }
+ catch(SocketException e)
+ {
+ // ignore - closed socket
+ }
+ }
+ }
+ }
+ catch (SocketTimeoutException e)
+ {
+ currentTime = System.currentTimeMillis();
+ if(_ticker != null)
+ {
+ final int tick = _ticker.tick(currentTime);
+ if(!socket.isClosed())
+ {
+ try
+ {
+ socket.setSoTimeout(tick <= 0 ? 1 : tick );
+ }
+ catch(SocketException ex)
+ {
+ // ignore - closed socket
+ }
+ }
}
}
}
@@ -195,4 +242,15 @@ final class IoReceiver implements Runnab
return !brokenClose && !sslSocketClosed;
}
+ public Ticker getTicker()
+ {
+ return _ticker;
+ }
+
+ public void setTicker(Ticker ticker)
+ {
+ _ticker = ticker;
+ }
+
+
}
Modified: qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java (original)
+++ qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java Sun Nov 25 17:15:43 2012
@@ -83,6 +83,18 @@ public class TestNetworkConnection imple
return null;
}
+ @Override
+ public int getMaxReadIdle()
+ {
+ return 0;
+ }
+
+ @Override
+ public int getMaxWriteIdle()
+ {
+ return 0;
+ }
+
public void setMaxWriteIdle(int idleTime)
{
Modified: qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java?rev=1413376&r1=1413375&r2=1413376&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java (original)
+++ qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java Sun Nov 25 17:15:43 2012
@@ -127,7 +127,9 @@ public class TransportTest extends QpidT
throw new UnsupportedOperationException();
}
- public NetworkConnection connect(ConnectionSettings settings, Receiver<ByteBuffer> delegate)
+ public NetworkConnection connect(ConnectionSettings settings,
+ Receiver<ByteBuffer> delegate,
+ TransportActivity transportActivity)
{
throw new UnsupportedOperationException();
}
@@ -147,7 +149,7 @@ public class TransportTest extends QpidT
}
public void accept(NetworkTransportConfiguration config,
- ProtocolEngineFactory factory, SSLContext sslContext)
+ ProtocolEngineFactory factory, SSLContext sslContext)
{
throw new UnsupportedOperationException();
}
Added: qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/io/IdleTimeoutTickerTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/io/IdleTimeoutTickerTest.java?rev=1413376&view=auto
==============================================================================
--- qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/io/IdleTimeoutTickerTest.java (added)
+++ qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/io/IdleTimeoutTickerTest.java Sun Nov 25 17:15:43 2012
@@ -0,0 +1,257 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.transport.network.io;
+
+import junit.framework.TestCase;
+
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.security.Principal;
+
+import org.apache.qpid.test.utils.QpidTestCase;
+import org.apache.qpid.transport.Sender;
+import org.apache.qpid.transport.network.NetworkConnection;
+import org.apache.qpid.transport.network.TransportActivity;
+
+public class IdleTimeoutTickerTest extends TestCase implements TransportActivity, NetworkConnection
+{
+ private IdleTimeoutTicker _ticker;
+ private static final int DEFAULT_TIMEOUT = 567890;
+ private long _lastReadTime;
+ private long _lastWriteTime;
+ private long _currentTime;
+ private int _maxWriteIdle;
+ private int _maxReadIdle;
+ private boolean _readerIdle;
+ private boolean _writerIdle;
+
+ @Override
+ public void setUp() throws Exception
+ {
+ super.setUp();
+ _ticker = new IdleTimeoutTicker(this, DEFAULT_TIMEOUT);
+ _ticker.setConnection(this);
+ _readerIdle = false;
+ _writerIdle = false;
+ _lastReadTime = 0l;
+ _lastWriteTime = 0l;
+ _maxReadIdle = 0;
+ _maxWriteIdle = 0;
+ }
+
+ public void testNoIdle() throws Exception
+ {
+ _maxReadIdle = 4;
+ _maxWriteIdle = 2;
+ _lastReadTime = 0;
+ _lastWriteTime = 1500;
+ _currentTime = 3000;
+ // Current time = 3s,
+ // last read = 0s, max read idle = 4s, should check in 1s
+ // last write = 1.5s, max write idle = 2s, should check in 0.5s
+ long nextTime = _ticker.tick(_currentTime);
+ assertEquals("Incorrect next tick calculation", 500l, nextTime);
+ assertFalse("Incorrectly caused reader idle", _readerIdle);
+ assertFalse("Incorrectly caused writer idle", _writerIdle);
+
+
+ // Current time = 3.4s,
+ // last read = 0s, max read idle = 4s, should check in 0.6s
+ // last write = 3.1s, max write idle = 2s, should check in 1.7s
+ _lastWriteTime = 3100;
+ _currentTime = 3400;
+ nextTime = _ticker.tick(_currentTime);
+ assertEquals("Incorrect next tick calculation", 600l, nextTime);
+ assertFalse("Incorrectly caused reader idle", _readerIdle);
+ assertFalse("Incorrectly caused writer idle", _writerIdle);
+
+ _maxReadIdle = 0;
+ nextTime = _ticker.tick(_currentTime);
+ assertEquals("Incorrect next tick calculation", 1700l, nextTime);
+ assertFalse("Incorrectly caused reader idle", _readerIdle);
+ assertFalse("Incorrectly caused writer idle", _writerIdle);
+
+ _maxWriteIdle = 0;
+ nextTime = _ticker.tick(_currentTime);
+ assertEquals("Incorrect next tick calculation", DEFAULT_TIMEOUT, nextTime);
+ assertFalse("Incorrectly caused reader idle", _readerIdle);
+ assertFalse("Incorrectly caused writer idle", _writerIdle);
+
+ }
+
+ public void testReaderIdle() throws Exception
+ {
+ _maxReadIdle = 4;
+ _maxWriteIdle = 0;
+ _lastReadTime = 0;
+ _lastWriteTime = 2500;
+ _currentTime = 4000;
+ // Current time = 4s,
+ // last read = 0s, max read idle = 4s, reader idle
+ long nextTime = _ticker.tick(_currentTime);
+
+ assertTrue(_readerIdle);
+ assertFalse(_writerIdle);
+
+ _readerIdle = false;
+
+ // last write = 2.5s, max write idle = 2s, should check in 0.5s
+ _maxWriteIdle = 2;
+ nextTime = _ticker.tick(_currentTime);
+ assertTrue(_readerIdle);
+ assertFalse(_writerIdle);
+
+ _readerIdle = false;
+ // last write = 1.5s, max write idle = 2s, should check in 0.5s
+
+ _lastWriteTime = 1500;
+ nextTime = _ticker.tick(_currentTime);
+
+ assertTrue(_readerIdle);
+ assertTrue(_writerIdle);
+
+ }
+
+ public void testWriterIdle() throws Exception
+ {
+ _maxReadIdle = 0;
+ _maxWriteIdle = 2;
+ _lastReadTime = 0;
+ _lastWriteTime = 1500;
+ _currentTime = 4000;
+ // Current time = 4s,
+ // last write = 1.5s, max write idle = 2s, writer idle
+ long nextTime = _ticker.tick(_currentTime);
+
+ assertTrue(_writerIdle);
+ assertFalse(_readerIdle);
+ assertEquals(2000l,nextTime);
+
+ _writerIdle = false;
+ _lastWriteTime = 1500;
+ _maxReadIdle = 5;
+
+ nextTime = _ticker.tick(_currentTime);
+
+ assertTrue(_writerIdle);
+ assertFalse(_readerIdle);
+ assertEquals(1000l,nextTime);
+
+ }
+
+ //-------------------------------------------------------------------------
+ // Implement TransportActivity methods
+ //-------------------------------------------------------------------------
+
+ @Override
+ public long getLastReadTime()
+ {
+ return _lastReadTime;
+ }
+
+ @Override
+ public long getLastWriteTime()
+ {
+ return _lastWriteTime;
+ }
+
+ @Override
+ public void writerIdle()
+ {
+ _writerIdle = true;
+ _lastWriteTime = _currentTime;
+ }
+
+ @Override
+ public void readerIdle()
+ {
+ _readerIdle = true;
+ }
+
+ //-------------------------------------------------------------------------
+ // Implement NetworkConnection methods
+ // Only actually use those relating to idle timeouts
+ //-------------------------------------------------------------------------
+
+ @Override
+ public Sender<ByteBuffer> getSender()
+ {
+ return null;
+ }
+
+ @Override
+ public void start()
+ {
+ }
+
+ @Override
+ public void close()
+ {
+ }
+
+ @Override
+ public SocketAddress getRemoteAddress()
+ {
+ return null;
+ }
+
+ @Override
+ public SocketAddress getLocalAddress()
+ {
+ return null;
+ }
+
+ @Override
+ public void setMaxWriteIdle(int sec)
+ {
+ _maxWriteIdle = sec;
+ }
+
+ @Override
+ public void setMaxReadIdle(int sec)
+ {
+ _maxReadIdle = sec;
+ }
+
+ @Override
+ public void setPeerPrincipal(Principal principal)
+ {
+ }
+
+ @Override
+ public Principal getPeerPrincipal()
+ {
+ return null;
+ }
+
+ @Override
+ public int getMaxReadIdle()
+ {
+ return _maxReadIdle;
+ }
+
+ @Override
+ public int getMaxWriteIdle()
+ {
+ return _maxWriteIdle;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org