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/28 22:21:07 UTC
svn commit: r1414927 - in /qpid/branches/0.20/qpid/java: ./
amqp-1-0-client-jms/ amqp-1-0-client/ amqp-1-0-common/ bdbstore/
broker-plugins/management-http/ broker/ broker/bin/ broker/src/
broker/src/main/java/org/apache/qpid/server/message/ broker/src...
Author: rgodfrey
Date: Wed Nov 28 21:21:05 2012
New Revision: 1414927
URL: http://svn.apache.org/viewvc?rev=1414927&view=rev
Log:
QPID-2796 : Restore heartbeating functionality to the Java Broker, and the 0-8,0-9,0-9-1 Java Client
merge from trunk r1413376,1413539,1413549,1413567
Added:
qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/HeartbeatListener.java
- copied unchanged from r1413539, qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/HeartbeatListener.java
qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Ticker.java
- copied unchanged from r1413376, qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Ticker.java
qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/TransportActivity.java
- copied unchanged from r1413376, qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/TransportActivity.java
qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java
- copied, changed from r1413376, qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java
qpid/branches/0.20/qpid/java/common/src/test/java/org/apache/qpid/transport/network/io/IdleTimeoutTickerTest.java
- copied unchanged from r1413376, qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/io/IdleTimeoutTickerTest.java
qpid/branches/0.20/qpid/java/systests/src/main/java/org/apache/qpid/client/HeartbeatTest.java
- copied, changed from r1413539, qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/client/HeartbeatTest.java
Removed:
qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/HeartbeatDiagnostics.java
Modified:
qpid/branches/0.20/qpid/java/ (props changed)
qpid/branches/0.20/qpid/java/amqp-1-0-client/ (props changed)
qpid/branches/0.20/qpid/java/amqp-1-0-client-jms/ (props changed)
qpid/branches/0.20/qpid/java/amqp-1-0-common/ (props changed)
qpid/branches/0.20/qpid/java/bdbstore/ (props changed)
qpid/branches/0.20/qpid/java/broker/ (props changed)
qpid/branches/0.20/qpid/java/broker-plugins/management-http/ (props changed)
qpid/branches/0.20/qpid/java/broker/bin/ (props changed)
qpid/branches/0.20/qpid/java/broker/src/ (props changed)
qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_1_0.java (props changed)
qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java (contents, props changed)
qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java (contents, props changed)
qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/ (props changed)
qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/ (props changed)
qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/ (props changed)
qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate.java
qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java
qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java
qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/transport/ClientConnectionDelegate.java
qpid/branches/0.20/qpid/java/common/bin/ (props changed)
qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java
qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java
qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java
qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java
qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
qpid/branches/0.20/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
qpid/branches/0.20/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java
qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/LoggingManagement.java (props changed)
qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java (props changed)
qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java (props changed)
qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedExchange.java (props changed)
qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java (props changed)
qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/UserManagement.java (props changed)
qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanAttribute.java (props changed)
qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanConstructor.java (props changed)
qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanDescription.java (props changed)
qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperation.java (props changed)
qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperationParameter.java (props changed)
qpid/branches/0.20/qpid/java/systests/src/main/java/org/apache/qpid/server/SupportedProtocolVersionsTest.java (props changed)
qpid/branches/0.20/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java (props changed)
qpid/branches/0.20/qpid/java/test-profiles/ (props changed)
qpid/branches/0.20/qpid/java/test-profiles/CPPExcludes (contents, props changed)
qpid/branches/0.20/qpid/java/test-profiles/Excludes (props changed)
qpid/branches/0.20/qpid/java/test-profiles/Java010Excludes
qpid/branches/0.20/qpid/java/test-profiles/JavaBDBExcludes (props changed)
qpid/branches/0.20/qpid/java/test-profiles/JavaExcludes (props changed)
qpid/branches/0.20/qpid/java/test-profiles/JavaPre010Excludes (props changed)
qpid/branches/0.20/qpid/java/test-profiles/JavaTransientExcludes (props changed)
qpid/branches/0.20/qpid/java/test-profiles/XAExcludes (props changed)
qpid/branches/0.20/qpid/java/test-profiles/cpp.async.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/cpp.cluster.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/cpp.noprefetch.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/cpp.ssl.excludes (props changed)
qpid/branches/0.20/qpid/java/test-profiles/cpp.ssl.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/cpp.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/java-bdb-spawn.0-9-1.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/java-bdb.0-9-1.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/java-dby-spawn.0-9-1.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/java-dby.0-9-1.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/java-mms-spawn.0-10.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/java-mms-spawn.0-9-1.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/java-mms.0-9-1.testprofile (props changed)
qpid/branches/0.20/qpid/java/test-profiles/log4j-test.xml (props changed)
qpid/branches/0.20/qpid/java/test-profiles/test-provider.properties (props changed)
qpid/branches/0.20/qpid/java/test-profiles/test_resources/ (props changed)
qpid/branches/0.20/qpid/java/test-profiles/testprofile.defaults (props changed)
Propchange: qpid/branches/0.20/qpid/java/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/amqp-1-0-client/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/amqp-1-0-client:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/amqp-1-0-client-jms/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/amqp-1-0-client-jms:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/amqp-1-0-common/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/amqp-1-0-common:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/bdbstore/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/bdbstore:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/broker/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/broker-plugins/management-http/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker-plugins/management-http:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/broker/bin/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/bin:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/broker/src/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_1_0.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_1_0.java:r1413376,1413539,1413549,1413567
Modified: qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java (original)
+++ qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java Wed Nov 28 21:21:05 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/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java (original)
+++ qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java Wed Nov 28 21:21:05 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/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java (original)
+++ qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java Wed Nov 28 21:21:05 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/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java (original)
+++ qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java Wed Nov 28 21:21:05 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;
@@ -182,6 +184,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();
@@ -324,6 +327,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());
@@ -378,4 +382,13 @@ public class ProtocolEngine_1_0_0 implem
return _connectionId;
}
+ public long getLastReadTime()
+ {
+ return _lastReadTime;
+ }
+
+ public long getLastWriteTime()
+ {
+ return _lastWriteTime;
+ }
}
Propchange: qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java:r1413376,1413539,1413549,1413567
Modified: qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java (original)
+++ qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java Wed Nov 28 21:21:05 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;
+ }
}
Propchange: qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue:r1413376,1413539,1413549,1413567
Modified: qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java (original)
+++ qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java Wed Nov 28 21:21:05 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/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java (original)
+++ qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java Wed Nov 28 21:21:05 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
Propchange: qpid/branches/0.20/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost:r1413376,1413539,1413549,1413567
Modified: qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java (original)
+++ qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java Wed Nov 28 21:21:05 2012
@@ -1560,4 +1560,9 @@ public class AMQConnection extends Close
+ localAddress + " to " + remoteAddress);
}
}
+
+ void setHeartbeatListener(HeartbeatListener listener)
+ {
+ _delegate.setHeartbeatListener(listener);
+ }
}
Modified: qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate.java (original)
+++ qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate.java Wed Nov 28 21:21:05 2012
@@ -78,4 +78,6 @@ public interface AMQConnectionDelegate
* @return true if the feature is supported by the server
*/
boolean isSupportedServerFeature(final String featureName);
+
+ void setHeartbeatListener(HeartbeatListener listener);
}
Modified: qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java (original)
+++ qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java Wed Nov 28 21:21:05 2012
@@ -422,6 +422,12 @@ public class AMQConnectionDelegate_0_10
return featureSupported;
}
+ @Override
+ public void setHeartbeatListener(HeartbeatListener listener)
+ {
+ ((ClientConnectionDelegate)(_qpidConnection.getConnectionDelegate())).setHeartbeatListener(listener);
+ }
+
private ConnectionSettings retrieveConnectionSettings(BrokerDetails brokerDetail)
{
ConnectionSettings conSettings = brokerDetail.buildConnectionSettings();
@@ -484,10 +490,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/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java (original)
+++ qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java Wed Nov 28 21:21:05 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);
@@ -376,4 +378,10 @@ public class AMQConnectionDelegate_8_0 i
// we just hardcode JMS selectors as supported.
return ServerPropertyNames.FEATURE_QPID_JMS_SELECTOR.equals(featureName);
}
+
+ @Override
+ public void setHeartbeatListener(HeartbeatListener listener)
+ {
+ _conn.getProtocolHandler().setHeartbeatListener(listener);
+ }
}
Modified: qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java (original)
+++ qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java Wed Nov 28 21:21:05 2012
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.client.protocol;
+import org.apache.qpid.client.HeartbeatListener;
import org.apache.qpid.util.BytesDataOutput;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -178,6 +179,9 @@ public class AMQProtocolHandler implemen
private NetworkConnection _network;
private Sender<ByteBuffer> _sender;
+ private long _lastReadTime = System.currentTimeMillis();
+ private long _lastWriteTime = System.currentTimeMillis();
+ private HeartbeatListener _heartbeatListener = HeartbeatListener.DEFAULT;
/**
* Creates a new protocol handler, associated with the specified client connection instance.
@@ -300,7 +304,6 @@ public class AMQProtocolHandler implemen
{
_logger.debug("Protocol Session [" + this + "] idle: reader");
// failover:
- HeartbeatDiagnostics.timeout();
_logger.warn("Timed out while waiting for heartbeat from peer.");
_network.close();
}
@@ -309,7 +312,7 @@ public class AMQProtocolHandler implemen
{
_logger.debug("Protocol Session [" + this + "] idle: reader");
writeFrame(HeartbeatBody.FRAME);
- HeartbeatDiagnostics.sent();
+ _heartbeatListener.heartbeatSent();
}
/**
@@ -442,6 +445,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);
@@ -470,8 +474,6 @@ public class AMQProtocolHandler implemen
final AMQBody bodyFrame = frame.getBodyFrame();
- HeartbeatDiagnostics.received(bodyFrame instanceof HeartbeatBody);
-
bodyFrame.handle(frame.getChannel(), _protocolSession);
_connection.bytesReceived(_readBytes);
@@ -560,6 +562,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 +885,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;
@@ -894,7 +909,6 @@ public class AMQProtocolHandler implemen
{
_network.setMaxWriteIdle(delay);
_network.setMaxReadIdle(HeartbeatConfig.CONFIG.getTimeout(delay));
- HeartbeatDiagnostics.init(delay, HeartbeatConfig.CONFIG.getTimeout(delay));
}
}
@@ -909,5 +923,13 @@ public class AMQProtocolHandler implemen
}
+ public void setHeartbeatListener(HeartbeatListener listener)
+ {
+ _heartbeatListener = listener == null ? HeartbeatListener.DEFAULT : listener;
+ }
+ public void heartbeatBodyReceived()
+ {
+ _heartbeatListener.heartbeatReceived();
+ }
}
Modified: qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java (original)
+++ qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java Wed Nov 28 21:21:05 2012
@@ -267,7 +267,7 @@ public class AMQProtocolSession implemen
public void heartbeatBodyReceived(int channelId, HeartbeatBody body) throws AMQException
{
-
+ _protocolHandler.heartbeatBodyReceived();
}
/**
Modified: qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/transport/ClientConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/transport/ClientConnectionDelegate.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/transport/ClientConnectionDelegate.java (original)
+++ qpid/branches/0.20/qpid/java/client/src/main/java/org/apache/qpid/client/transport/ClientConnectionDelegate.java Wed Nov 28 21:21:05 2012
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.client.transport;
+import org.apache.qpid.client.HeartbeatListener;
+import org.apache.qpid.transport.ConnectionHeartbeat;
import org.ietf.jgss.GSSContext;
import org.ietf.jgss.GSSException;
import org.ietf.jgss.GSSManager;
@@ -70,6 +72,7 @@ public class ClientConnectionDelegate ex
}
private final ConnectionURL _connectionURL;
+ private HeartbeatListener _heartbeatListener = HeartbeatListener.DEFAULT;
/**
* @param settings
@@ -165,4 +168,19 @@ public class ClientConnectionDelegate ex
return null;
}
+
+ @Override
+ public void connectionHeartbeat(Connection conn, ConnectionHeartbeat hearbeat)
+ {
+ // ClientDelegate simply responds to heartbeats with heartbeats
+ _heartbeatListener.heartbeatReceived();
+ super.connectionHeartbeat(conn, hearbeat);
+ _heartbeatListener.heartbeatSent();
+ }
+
+
+ public void setHeartbeatListener(HeartbeatListener listener)
+ {
+ _heartbeatListener = listener == null ? HeartbeatListener.DEFAULT : listener;
+ }
}
Propchange: qpid/branches/0.20/qpid/java/common/bin/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/common/bin:r1413376,1413539,1413549,1413567
Modified: qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java (original)
+++ qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java Wed Nov 28 21:21:05 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/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java (original)
+++ qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java Wed Nov 28 21:21:05 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, new ConnectionActivity());
+
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/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java (original)
+++ qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java Wed Nov 28 21:21:05 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/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java (original)
+++ qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java Wed Nov 28 21:21:05 2012
@@ -50,4 +50,8 @@ public interface NetworkConnection
void setPeerPrincipal(Principal principal);
Principal getPeerPrincipal();
+
+ int getMaxReadIdle();
+
+ int getMaxWriteIdle();
}
Modified: qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java (original)
+++ qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java Wed Nov 28 21:21:05 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
Copied: qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java (from r1413376, qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java)
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java?p2=qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java&p1=qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java&r1=1413376&r2=1414927&rev=1414927&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java (original)
+++ qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IdleTimeoutTicker.java Wed Nov 28 21:21:05 2012
@@ -43,14 +43,14 @@ class IdleTimeoutTicker implements Ticke
long nextTime = -1;
final long maxReadIdle = 1000l * _connection.getMaxReadIdle();
- if(maxReadIdle != 0)
+ if(maxReadIdle > 0)
{
nextTime = _transport.getLastReadTime() + maxReadIdle;
}
long maxWriteIdle = 1000l * _connection.getMaxWriteIdle();
- if(maxWriteIdle != 0)
+ if(maxWriteIdle > 0)
{
long writeTime = _transport.getLastWriteTime() + maxWriteIdle;
if(nextTime == -1l || writeTime < nextTime)
@@ -66,13 +66,13 @@ class IdleTimeoutTicker implements Ticke
{
// writer Idle
long maxWriteIdle = 1000l * _connection.getMaxWriteIdle();
- if(maxWriteIdle != 0 && maxWriteIdle+ _transport.getLastWriteTime() <= currentTime)
+ if(maxWriteIdle > 0 && maxWriteIdle+ _transport.getLastWriteTime() <= currentTime)
{
_transport.writerIdle();
}
// reader Idle
final long maxReadIdle = 1000l * _connection.getMaxReadIdle();
- if(maxReadIdle != 0 && maxReadIdle+ _transport.getLastReadTime() <= currentTime)
+ if(maxReadIdle > 0 && maxReadIdle+ _transport.getLastReadTime() <= currentTime)
{
_transport.readerIdle();
Modified: qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java (original)
+++ qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java Wed Nov 28 21:21:05 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/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java (original)
+++ qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java Wed Nov 28 21:21:05 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/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java (original)
+++ qpid/branches/0.20/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java Wed Nov 28 21:21:05 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,64 @@ 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)
+ {
+ int tick = _ticker.getTimeToNextTick(currentTime);
+ if(tick <= 0)
+ {
+ tick = _ticker.tick(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 +246,15 @@ final class IoReceiver implements Runnab
return !brokenClose && !sslSocketClosed;
}
+ public Ticker getTicker()
+ {
+ return _ticker;
+ }
+
+ public void setTicker(Ticker ticker)
+ {
+ _ticker = ticker;
+ }
+
+
}
Modified: qpid/branches/0.20/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java (original)
+++ qpid/branches/0.20/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java Wed Nov 28 21:21:05 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/branches/0.20/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java (original)
+++ qpid/branches/0.20/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java Wed Nov 28 21:21:05 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();
}
Propchange: qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/LoggingManagement.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/LoggingManagement.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedExchange.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedExchange.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/UserManagement.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/UserManagement.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanAttribute.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanAttribute.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanConstructor.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanConstructor.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanDescription.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanDescription.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperation.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperation.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperationParameter.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/annotations/MBeanOperationParameter.java:r1413376,1413539,1413549,1413567
Copied: qpid/branches/0.20/qpid/java/systests/src/main/java/org/apache/qpid/client/HeartbeatTest.java (from r1413539, qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/client/HeartbeatTest.java)
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/systests/src/main/java/org/apache/qpid/client/HeartbeatTest.java?p2=qpid/branches/0.20/qpid/java/systests/src/main/java/org/apache/qpid/client/HeartbeatTest.java&p1=qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/client/HeartbeatTest.java&r1=1413539&r2=1414927&rev=1414927&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/client/HeartbeatTest.java (original)
+++ qpid/branches/0.20/qpid/java/systests/src/main/java/org/apache/qpid/client/HeartbeatTest.java Wed Nov 28 21:21:05 2012
@@ -18,13 +18,17 @@
*/
package org.apache.qpid.client;
+import javax.jms.Destination;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
import org.apache.qpid.test.utils.QpidBrokerTestCase;
public class HeartbeatTest extends QpidBrokerTestCase
{
public void testHeartbeats() throws Exception
{
- setTestSystemProperty("amqj.heartbeat.delay","1");
+ setTestSystemProperty("amqj.heartbeat.delay", "1");
AMQConnection conn = (AMQConnection) getConnection();
TestListener listener = new TestListener();
conn.setHeartbeatListener(listener);
@@ -40,7 +44,7 @@ public class HeartbeatTest extends QpidB
public void testNoHeartbeats() throws Exception
{
- setTestSystemProperty("amqj.heartbeat.delay","0");
+ setTestSystemProperty("amqj.heartbeat.delay", "0");
AMQConnection conn = (AMQConnection) getConnection();
TestListener listener = new TestListener();
conn.setHeartbeatListener(listener);
@@ -54,6 +58,45 @@ public class HeartbeatTest extends QpidB
conn.close();
}
+ public void testReadOnlyConnectionHeartbeats() throws Exception
+ {
+ setTestSystemProperty("amqj.heartbeat.delay","1");
+ AMQConnection receiveConn = (AMQConnection) getConnection();
+ AMQConnection sendConn = (AMQConnection) getConnection();
+ Destination destination = getTestQueue();
+ TestListener receiveListener = new TestListener();
+ TestListener sendListener = new TestListener();
+
+
+ Session receiveSession = receiveConn.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+ Session senderSession = sendConn.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+
+ MessageConsumer consumer = receiveSession.createConsumer(destination);
+ MessageProducer producer = senderSession.createProducer(destination);
+
+ receiveConn.setHeartbeatListener(receiveListener);
+ sendConn.setHeartbeatListener(sendListener);
+ receiveConn.start();
+
+ for(int i = 0; i < 5; i++)
+ {
+ producer.send(senderSession.createTextMessage("Msg " + i));
+ Thread.sleep(500);
+ assertNotNull("Expected to received message", consumer.receive(500));
+ }
+
+
+
+ assertTrue("Too few heartbeats sent "+receiveListener._heartbeatsSent+" (expected at least 2)", receiveListener._heartbeatsSent>=2);
+ assertEquals("Unexpected sent at the sender: ",0,sendListener._heartbeatsSent);
+
+ assertTrue("Too few heartbeats received at the sender "+sendListener._heartbeatsReceived+" (expected at least 2)", sendListener._heartbeatsReceived>=2);
+ assertEquals("Unexpected received at the receiver: ",0,receiveListener._heartbeatsReceived);
+
+ receiveConn.close();
+ sendConn.close();
+ }
+
private class TestListener implements HeartbeatListener
{
int _heartbeatsReceived;
Propchange: qpid/branches/0.20/qpid/java/systests/src/main/java/org/apache/qpid/server/SupportedProtocolVersionsTest.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/SupportedProtocolVersionsTest.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles:r1413376,1413539,1413549,1413567
Modified: qpid/branches/0.20/qpid/java/test-profiles/CPPExcludes
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/test-profiles/CPPExcludes?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/test-profiles/CPPExcludes (original)
+++ qpid/branches/0.20/qpid/java/test-profiles/CPPExcludes Wed Nov 28 21:21:05 2012
@@ -181,3 +181,7 @@ org.apache.qpid.client.failover.Multiple
// Uses Java broker specific configuration
org.apache.qpid.client.ssl.SSLTest#testClientCertMissingWhilstWanting
+
+// QPID-2796 : Java 0-10 client only sends heartbeats in response to heartbeats from the server, not timeout based
+org.apache.qpid.client.HeartbeatTest#testReadOnlyConnectionHeartbeats
+
Propchange: qpid/branches/0.20/qpid/java/test-profiles/CPPExcludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/CPPExcludes:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/Excludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/Excludes:r1413376,1413539,1413549,1413567
Modified: qpid/branches/0.20/qpid/java/test-profiles/Java010Excludes
URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/java/test-profiles/Java010Excludes?rev=1414927&r1=1414926&r2=1414927&view=diff
==============================================================================
--- qpid/branches/0.20/qpid/java/test-profiles/Java010Excludes (original)
+++ qpid/branches/0.20/qpid/java/test-profiles/Java010Excludes Wed Nov 28 21:21:05 2012
@@ -64,3 +64,6 @@ org.apache.qpid.client.failover.AddressB
// QPID-3604: Immediate Prefetch no longer supported by 0-10
org.apache.qpid.client.AsynchMessageListenerTest#testImmediatePrefetchWithMessageListener
+
+// QPID-2796 : Java 0-10 client only sends heartbeats in response to heartbeats from the server, not timeout based
+org.apache.qpid.client.HeartbeatTest#testReadOnlyConnectionHeartbeats
Propchange: qpid/branches/0.20/qpid/java/test-profiles/JavaBDBExcludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/JavaBDBExcludes:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/JavaExcludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/JavaExcludes:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/JavaPre010Excludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/JavaPre010Excludes:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/JavaTransientExcludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/JavaTransientExcludes:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/XAExcludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/XAExcludes:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/cpp.async.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.async.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/cpp.cluster.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.cluster.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/cpp.noprefetch.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.noprefetch.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/cpp.ssl.excludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.ssl.excludes:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/cpp.ssl.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.ssl.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/cpp.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/java-bdb-spawn.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-bdb-spawn.0-9-1.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/java-bdb.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-bdb.0-9-1.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/java-dby-spawn.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-dby-spawn.0-9-1.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/java-dby.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-dby.0-9-1.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/java-mms-spawn.0-10.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-mms-spawn.0-10.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/java-mms-spawn.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-mms-spawn.0-9-1.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/java-mms.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-mms.0-9-1.testprofile:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/log4j-test.xml
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/log4j-test.xml:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/test-provider.properties
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/test-provider.properties:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/test_resources/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/test_resources:r1413376,1413539,1413549,1413567
Propchange: qpid/branches/0.20/qpid/java/test-profiles/testprofile.defaults
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/testprofile.defaults:r1413376,1413539,1413549,1413567
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org