You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ta...@apache.org on 2015/01/15 23:56:34 UTC
[4/5] qpid-jms git commit: User proper netty buffer reference
counting to ensure the dispatched buffer is not freed prior to processing.
User proper netty buffer reference counting to ensure the dispatched
buffer is not freed prior to processing.
Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/8ba0199c
Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/8ba0199c
Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/8ba0199c
Branch: refs/heads/master
Commit: 8ba0199c9f4ac633e23e1c366409401691b93360
Parents: 015280c
Author: Timothy Bish <ta...@gmail.com>
Authored: Thu Jan 15 16:50:45 2015 -0500
Committer: Timothy Bish <ta...@gmail.com>
Committed: Thu Jan 15 16:50:45 2015 -0500
----------------------------------------------------------------------
.../org/apache/qpid/jms/provider/amqp/AmqpProvider.java | 10 +++++++---
.../qpid/jms/transports/netty/NettyTcpTransport.java | 10 +++++-----
2 files changed, 12 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/8ba0199c/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
index d56e0b0..61e84e7 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
@@ -17,6 +17,7 @@
package org.apache.qpid.jms.provider.amqp;
import io.netty.buffer.ByteBuf;
+import io.netty.util.ReferenceCountUtil;
import java.io.IOException;
import java.net.URI;
@@ -600,11 +601,14 @@ public class AmqpProvider extends AbstractProvider implements TransportListener
@Override
public void onData(final ByteBuf input) {
+ // We need to retain until the serializer gets around to processing it.
+ ReferenceCountUtil.retain(input);
+
serializer.execute(new Runnable() {
@Override
public void run() {
- LOG.trace("Received from Broker {} bytes:", input.readableBytes());
+ LOG.trace("Received from Broker {} bytes: {}", input.readableBytes(), input);
ByteBuffer source = input.nioBuffer();
@@ -618,6 +622,8 @@ public class AmqpProvider extends AbstractProvider implements TransportListener
source.position(source.position() + limit);
} while (source.hasRemaining());
+ ReferenceCountUtil.release(input);
+
// Process the state changes from the latest data and then answer back
// any pending updates to the Broker.
processUpdates();
@@ -659,8 +665,6 @@ public class AmqpProvider extends AbstractProvider implements TransportListener
*/
@Override
public void onTransportClosed() {
- // TODO: improve or delete this logging
- LOG.debug("onTransportClosed listener called");
if (!serializer.isShutdown()) {
serializer.execute(new Runnable() {
@Override
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/8ba0199c/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyTcpTransport.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyTcpTransport.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyTcpTransport.java
index 03db5ed..84815f9 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyTcpTransport.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyTcpTransport.java
@@ -163,7 +163,7 @@ public class NettyTcpTransport implements Transport {
return;
}
- LOG.info("Attempted write of: {} bytes", length);
+ LOG.trace("Attempted write of: {} bytes", length);
channel.write(output);
channel.flush();
@@ -222,12 +222,12 @@ public class NettyTcpTransport implements Transport {
@Override
public void channelActive(ChannelHandlerContext context) throws Exception {
- LOG.info("Channel has become active! Channel is {}", context.channel());
+ LOG.trace("Channel has become active! Channel is {}", context.channel());
}
@Override
public void channelInactive(ChannelHandlerContext context) throws Exception {
- LOG.info("Channel has gone inactive! Channel is {}", context.channel());
+ LOG.trace("Channel has gone inactive! Channel is {}", context.channel());
if (!closed.get()) {
connected.set(false);
listener.onTransportClosed();
@@ -236,7 +236,7 @@ public class NettyTcpTransport implements Transport {
@Override
public void exceptionCaught(ChannelHandlerContext context, Throwable cause) throws Exception {
- LOG.info("Exception on channel! Channel is {}", context.channel());
+ LOG.trace("Exception on channel! Channel is {}", context.channel());
if (!closed.get()) {
connected.set(false);
listener.onTransportError(cause);
@@ -245,7 +245,7 @@ public class NettyTcpTransport implements Transport {
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
- LOG.info("New data read: {} bytes incoming", buffer.readableBytes());
+ LOG.trace("New data read: {} bytes incoming: {}", buffer.readableBytes(), buffer);
listener.onData(buffer);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org