diff options
Diffstat (limited to 'qpid/java/client')
5 files changed, 14 insertions, 41 deletions
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java index 46473900c0..19720ea386 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java @@ -101,7 +101,7 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic { ssn.flushAcknowledgments(true); } - catch (Throwable t) + catch (Exception t) { _logger.error("error flushing acks", t); } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java b/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java index 71d07b1fa0..9748038b9b 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java @@ -20,14 +20,12 @@ */ package org.apache.qpid.client.message; -import java.io.ByteArrayOutputStream; import java.io.EOFException; import java.io.IOException; import java.io.InputStream; import java.nio.ByteBuffer; import java.util.Iterator; import java.util.List; -import java.util.zip.GZIPInputStream; import javax.jms.JMSException; @@ -164,30 +162,6 @@ public abstract class AbstractJMSMessageFactory implements MessageFactory return message; } - private ByteBuffer uncompressBody(final InputStream bodyInputStream) throws AMQException - { - final ByteBuffer data; - try(GZIPInputStream gzipInputStream = new GZIPInputStream(bodyInputStream)) - { - ByteArrayOutputStream uncompressedBuffer = new ByteArrayOutputStream(); - int read; - byte[] buf = new byte[4096]; - while((read = gzipInputStream.read(buf))!=-1) - { - uncompressedBuffer.write(buf,0,read); - } - byte[] uncompressedBytes = uncompressedBuffer.toByteArray(); - data = ByteBuffer.wrap(uncompressedBytes); - } - catch (IOException e) - { - // TODO - shouldn't happen - throw new AMQException("Error uncompressing gzipped message data", e); - } - return data; - } - - public AbstractJMSMessage createMessage(long messageNbr, boolean redelivered, ContentHeaderBody contentHeader, AMQShortString exchange, AMQShortString routingKey, List bodies, AMQSession_0_8.DestinationCache<AMQQueue> queueDestinationCache, diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/message/FieldTableSupport.java b/qpid/java/client/src/main/java/org/apache/qpid/client/message/FieldTableSupport.java index c3f36a545a..3d3b753ad3 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/message/FieldTableSupport.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/message/FieldTableSupport.java @@ -20,12 +20,12 @@ */ package org.apache.qpid.client.message; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.FieldTable; - import java.util.HashMap; import java.util.Map; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.FieldTable; + public class FieldTableSupport { private FieldTableSupport() 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 681082526c..48bfe74cfc 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 @@ -172,6 +172,10 @@ public class AMQProtocolHandler implements ProtocolEngine private long _writtenBytes; private long _readBytes; + private int _messageReceivedCount; + private int _messagesOut; + + private NetworkConnection _network; private Sender<ByteBuffer> _sender; private long _lastReadTime = System.currentTimeMillis(); @@ -451,9 +455,6 @@ public class AMQProtocolHandler implements ProtocolEngine _lastFailoverException = null; } - private static int _messageReceivedCount; - - public void received(ByteBuffer msg) { _readBytes += msg.remaining(); @@ -559,8 +560,6 @@ public class AMQProtocolHandler implements ProtocolEngine } - private static int _messagesOut; - public StateWaiter createWaiter(Set<AMQState> states) throws AMQException { return getStateManager().createWaiter(states); diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/util/BlockingWaiter.java b/qpid/java/client/src/main/java/org/apache/qpid/client/util/BlockingWaiter.java index 53b6730ef7..21f1623dd1 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/util/BlockingWaiter.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/util/BlockingWaiter.java @@ -20,6 +20,11 @@ */ package org.apache.qpid.client.util; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -27,11 +32,6 @@ import org.apache.qpid.AMQException; import org.apache.qpid.AMQTimeoutException; import org.apache.qpid.client.failover.FailoverException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.locks.Condition; -import java.util.concurrent.locks.ReentrantLock; - /** * BlockingWaiter is a 'rendezvous' which delegates handling of * incoming Objects to a listener implemented as a sub-class of this and hands off the process or @@ -243,7 +243,7 @@ public abstract class BlockingWaiter<T> } else { - _logger.error("WARNING: new error '" + e == null ? "null" : e.getMessage() + "' arrived while old one not yet processed:" + _error.getMessage()); + _logger.error("WARNING: new error '" + (e == null ? "null" : e.getMessage()) + "' arrived while old one not yet processed:" + _error.getMessage()); } if (_waiting.get()) |