diff options
Diffstat (limited to 'qpid/java')
| -rw-r--r-- | qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java | 15 | ||||
| -rw-r--r-- | qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java | 5 |
2 files changed, 3 insertions, 17 deletions
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java index 3bcd102858..a6fcadad0f 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java @@ -201,7 +201,6 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol return _actor; } - @Override public void received(final ByteBuffer msg) { _lastIoTime = System.currentTimeMillis(); @@ -210,7 +209,6 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol final ArrayList<AMQDataBlock> dataBlocks = _codecFactory.getDecoder().decodeBuffer(msg); Job.fireAsynchEvent(_poolReference.getPool(), _readJob, new Runnable() { - @Override public void run() { // Decode buffer @@ -457,7 +455,6 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol _writtenBytes += buf.remaining(); Job.fireAsynchEvent(_poolReference.getPool(), _writeJob, new Runnable() { - @Override public void run() { _networkDriver.send(buf); @@ -679,6 +676,7 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol /** This must be called when the session is _closed in order to free up any resources managed by the session. */ public void closeSession() throws AMQException { + // REMOVE THIS SHOULD NOT BE HERE. if (CurrentActor.get() == null) { CurrentActor.set(_actor); @@ -694,6 +692,8 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol if (_managedObject != null) { _managedObject.unregister(); + // Ensure we only do this once. + _managedObject = null; } for (Task task : _taskList) @@ -920,7 +920,6 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol return _dispatcher; } - @Override public void closed() { try @@ -933,25 +932,21 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol } } - @Override public void readerIdle() { // Nothing } - @Override public void setNetworkDriver(NetworkDriver driver) { _networkDriver = driver; } - @Override public void writerIdle() { _networkDriver.send(HeartbeatBody.FRAME.toNioByteBuffer()); } - @Override public void exception(Throwable throwable) { if (throwable instanceof AMQProtocolHeaderException) @@ -980,19 +975,16 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol } } - @Override public void init() { // Do nothing } - @Override public void setSender(Sender<ByteBuffer> sender) { // Do nothing } - @Override public long getReadBytes() { return _readBytes; @@ -1018,7 +1010,6 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol return (_clientVersion == null) ? null : _clientVersion.toString(); } - @Override public void closeIfLingeringClosedChannels() { for (Entry<Integer, Long>id : _closingChannelsList.entrySet()) diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java index d625cdb6f2..35bc521c80 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java @@ -285,7 +285,6 @@ public class AMQProtocolHandler implements ProtocolEngine failoverThread.start(); } - @Override public void readerIdle() { _logger.debug("Protocol Session [" + this + "] idle: reader"); @@ -295,7 +294,6 @@ public class AMQProtocolHandler implements ProtocolEngine _networkDriver.close(); } - @Override public void writerIdle() { _logger.debug("Protocol Session [" + this + "] idle: reader"); @@ -421,7 +419,6 @@ public class AMQProtocolHandler implements ProtocolEngine private static int _messageReceivedCount; - @Override public void received(ByteBuffer msg) { try @@ -431,7 +428,6 @@ public class AMQProtocolHandler implements ProtocolEngine Job.fireAsynchEvent(_poolReference.getPool(), _readJob, new Runnable() { - @Override public void run() { // Decode buffer @@ -566,7 +562,6 @@ public class AMQProtocolHandler implements ProtocolEngine _writtenBytes += buf.remaining(); Job.fireAsynchEvent(_poolReference.getPool(), _writeJob, new Runnable() { - @Override public void run() { _networkDriver.send(buf); |