diff options
Diffstat (limited to 'qpid/java/amqp-1-0-client')
5 files changed, 107 insertions, 58 deletions
diff --git a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Connection.java b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Connection.java index 6157ec53f6..9319d4ddff 100644 --- a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Connection.java +++ b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Connection.java @@ -25,8 +25,10 @@ import java.security.Principal; import java.util.ServiceLoader; import java.util.concurrent.TimeoutException; -import org.apache.qpid.amqp_1_0.framing.ExceptionHandler; +import javax.net.ssl.SSLContext; + import org.apache.qpid.amqp_1_0.framing.ConnectionHandler; +import org.apache.qpid.amqp_1_0.framing.ExceptionHandler; import org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint; import org.apache.qpid.amqp_1_0.transport.Container; import org.apache.qpid.amqp_1_0.transport.Predicate; @@ -37,8 +39,6 @@ import org.apache.qpid.amqp_1_0.type.transport.AmqpError; import org.apache.qpid.amqp_1_0.type.transport.ConnectionError; import org.apache.qpid.amqp_1_0.type.transport.Error; -import javax.net.ssl.SSLContext; - public class Connection implements ExceptionHandler { private static final int MAX_FRAME_SIZE = 65536; @@ -225,7 +225,7 @@ public class Connection implements ExceptionHandler (byte)1, (byte)0, (byte)0), - new ConnectionHandler.FrameToBytesSourceAdapter(saslOut,_conn.getDescribedTypeRegistry()), + new ConnectionHandler.FrameToBytesSourceAdapter(saslOut.asFrameSource(),_conn.getDescribedTypeRegistry()), new ConnectionHandler.HeaderBytesSource(_conn, (byte)'A', (byte)'M', (byte)'Q', @@ -234,7 +234,7 @@ public class Connection implements ExceptionHandler (byte)1, (byte)0, (byte)0), - new ConnectionHandler.FrameToBytesSourceAdapter(out,_conn.getDescribedTypeRegistry()) + new ConnectionHandler.FrameToBytesSourceAdapter(out.asFrameSource(),_conn.getDescribedTypeRegistry()) ); _conn.setSaslFrameOutput(saslOut); @@ -249,7 +249,7 @@ public class Connection implements ExceptionHandler (byte)1, (byte)0, (byte)0), - new ConnectionHandler.FrameToBytesSourceAdapter(out,_conn.getDescribedTypeRegistry()) + new ConnectionHandler.FrameToBytesSourceAdapter(out.asFrameSource(),_conn.getDescribedTypeRegistry()) ); } @@ -258,7 +258,14 @@ public class Connection implements ExceptionHandler transportProvider.connect(_conn,address,port, sslContext, this); - _conn.open(); + try + { + _conn.open(); + } + catch(RuntimeException e) + { + transportProvider.close(); + } } @@ -295,7 +302,14 @@ public class Connection implements ExceptionHandler { if(getEndpoint().isClosed()) { - throw new ConnectionClosedException(getEndpoint().getRemoteError()); + Error remoteError = getEndpoint().getRemoteError(); + if(remoteError == null) + { + remoteError = new Error(); + remoteError.setDescription("Connection closed for unknown reason"); + + } + throw new ConnectionClosedException(remoteError); } } @@ -377,7 +391,7 @@ public class Connection implements ExceptionHandler if(_connectionErrorTask != null) { Thread thread = new Thread(_connectionErrorTask); - thread.run(); + thread.start(); } } } diff --git a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java index ad2924c01e..a2a15779d2 100644 --- a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java +++ b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java @@ -20,26 +20,41 @@ */ package org.apache.qpid.amqp_1_0.client; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeoutException; + import org.apache.qpid.amqp_1_0.messaging.SectionDecoder; import org.apache.qpid.amqp_1_0.transport.DeliveryStateHandler; import org.apache.qpid.amqp_1_0.transport.LinkEndpoint; import org.apache.qpid.amqp_1_0.transport.Predicate; import org.apache.qpid.amqp_1_0.transport.ReceivingLinkEndpoint; import org.apache.qpid.amqp_1_0.transport.ReceivingLinkListener; - -import org.apache.qpid.amqp_1_0.type.*; +import org.apache.qpid.amqp_1_0.type.AmqpErrorException; +import org.apache.qpid.amqp_1_0.type.Binary; import org.apache.qpid.amqp_1_0.type.DeliveryState; -import org.apache.qpid.amqp_1_0.type.messaging.*; +import org.apache.qpid.amqp_1_0.type.Outcome; +import org.apache.qpid.amqp_1_0.type.Section; +import org.apache.qpid.amqp_1_0.type.UnsignedInteger; +import org.apache.qpid.amqp_1_0.type.messaging.Accepted; +import org.apache.qpid.amqp_1_0.type.messaging.Modified; +import org.apache.qpid.amqp_1_0.type.messaging.Released; import org.apache.qpid.amqp_1_0.type.messaging.Source; import org.apache.qpid.amqp_1_0.type.messaging.Target; +import org.apache.qpid.amqp_1_0.type.messaging.TerminusDurability; +import org.apache.qpid.amqp_1_0.type.messaging.TerminusExpiryPolicy; import org.apache.qpid.amqp_1_0.type.transaction.TransactionalState; -import org.apache.qpid.amqp_1_0.type.transport.*; +import org.apache.qpid.amqp_1_0.type.transport.AmqpError; +import org.apache.qpid.amqp_1_0.type.transport.Detach; import org.apache.qpid.amqp_1_0.type.transport.Error; - -import java.nio.ByteBuffer; -import java.util.*; -import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.TimeoutException; +import org.apache.qpid.amqp_1_0.type.transport.ReceiverSettleMode; +import org.apache.qpid.amqp_1_0.type.transport.SenderSettleMode; +import org.apache.qpid.amqp_1_0.type.transport.Transfer; public class Receiver implements DeliveryStateHandler { @@ -193,7 +208,8 @@ public class Receiver implements DeliveryStateHandler { if(_remoteErrorTask != null) { - _remoteErrorTask.run(); + Thread thread = new Thread(_remoteErrorTask); + thread.start(); } } diff --git a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Sender.java b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Sender.java index 1addad2235..adeab4ab5d 100644 --- a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Sender.java +++ b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Sender.java @@ -20,6 +20,13 @@ */ package org.apache.qpid.amqp_1_0.client; +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeoutException; + import org.apache.qpid.amqp_1_0.codec.DescribedTypeConstructor; import org.apache.qpid.amqp_1_0.messaging.SectionEncoder; import org.apache.qpid.amqp_1_0.transport.DeliveryStateHandler; @@ -27,22 +34,21 @@ import org.apache.qpid.amqp_1_0.transport.LinkEndpoint; import org.apache.qpid.amqp_1_0.transport.Predicate; import org.apache.qpid.amqp_1_0.transport.SendingLinkEndpoint; import org.apache.qpid.amqp_1_0.transport.SendingLinkListener; -import org.apache.qpid.amqp_1_0.type.*; +import org.apache.qpid.amqp_1_0.type.Binary; +import org.apache.qpid.amqp_1_0.type.DeliveryState; +import org.apache.qpid.amqp_1_0.type.Outcome; +import org.apache.qpid.amqp_1_0.type.Section; import org.apache.qpid.amqp_1_0.type.Source; import org.apache.qpid.amqp_1_0.type.Target; import org.apache.qpid.amqp_1_0.type.codec.AMQPDescribedTypeRegistry; -import org.apache.qpid.amqp_1_0.type.messaging.*; +import org.apache.qpid.amqp_1_0.type.messaging.Accepted; +import org.apache.qpid.amqp_1_0.type.messaging.TerminusDurability; +import org.apache.qpid.amqp_1_0.type.messaging.TerminusExpiryPolicy; import org.apache.qpid.amqp_1_0.type.transaction.TransactionalState; -import org.apache.qpid.amqp_1_0.type.transport.*; - -import java.nio.ByteBuffer; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.TimeoutException; - +import org.apache.qpid.amqp_1_0.type.transport.Detach; import org.apache.qpid.amqp_1_0.type.transport.Error; +import org.apache.qpid.amqp_1_0.type.transport.SenderSettleMode; +import org.apache.qpid.amqp_1_0.type.transport.Transfer; public class Sender implements DeliveryStateHandler { @@ -488,7 +494,8 @@ public class Sender implements DeliveryStateHandler { if(_remoteErrorTask != null) { - _remoteErrorTask.run(); + Thread thread = new Thread(_remoteErrorTask); + thread.start(); } } diff --git a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/TCPTransportProvier.java b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/TCPTransportProvier.java index ee515c33ef..da084bdc7b 100644 --- a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/TCPTransportProvier.java +++ b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/TCPTransportProvier.java @@ -26,6 +26,9 @@ import java.io.OutputStream; import java.net.Socket; import java.net.SocketTimeoutException; import java.nio.ByteBuffer; +import java.util.concurrent.atomic.AtomicLong; +import java.util.logging.Level; +import java.util.logging.Logger; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLSocket; @@ -39,6 +42,9 @@ import org.apache.qpid.amqp_1_0.type.SaslFrameBody; class TCPTransportProvier implements TransportProvider { + private static final Logger RAW_LOGGER = Logger.getLogger("RAW"); + + private Socket _socket; private final String _transport; // Defines read socket timeout in milliseconds. A value of 0 means that the socket @@ -49,6 +55,7 @@ class TCPTransportProvier implements TransportProvider // the event of a SocketTimeoutException. A value of -1L will disable idle read timeout checking. // Default value is set to -1L, which means disable idle read checks. private long _readIdleTimeout = Long.getLong("qpid.connection_read_idle_timeout", -1L); + private final AtomicLong _threadNameIndex = new AtomicLong(); public TCPTransportProvier(final String transport) { @@ -64,7 +71,6 @@ class TCPTransportProvier implements TransportProvider { try { - final Socket s; if(sslContext != null) { final SSLSocketFactory socketFactory = sslContext.getSocketFactory(); @@ -72,16 +78,16 @@ class TCPTransportProvier implements TransportProvider SSLSocket sslSocket = (SSLSocket) socketFactory.createSocket(address, port); conn.setExternalPrincipal(sslSocket.getSession().getLocalPrincipal()); - s=sslSocket; + _socket=sslSocket; } else { - s = new Socket(address, port); + _socket = new Socket(address, port); } // set socket read timeout - s.setSoTimeout(_readTimeout); + _socket.setSoTimeout(_readTimeout); - conn.setRemoteAddress(s.getRemoteSocketAddress()); + conn.setRemoteAddress(_socket.getRemoteSocketAddress()); ConnectionHandler.FrameOutput<FrameBody> out = new ConnectionHandler.FrameOutput<FrameBody>(conn); @@ -99,7 +105,7 @@ class TCPTransportProvier implements TransportProvider (byte)1, (byte)0, (byte)0), - new ConnectionHandler.FrameToBytesSourceAdapter(saslOut,conn.getDescribedTypeRegistry()), + new ConnectionHandler.FrameToBytesSourceAdapter(saslOut.asFrameSource(),conn.getDescribedTypeRegistry()), new ConnectionHandler.HeaderBytesSource(conn, (byte)'A', (byte)'M', (byte)'Q', @@ -108,7 +114,7 @@ class TCPTransportProvier implements TransportProvider (byte)1, (byte)0, (byte)0), - new ConnectionHandler.FrameToBytesSourceAdapter(out,conn.getDescribedTypeRegistry()) + new ConnectionHandler.FrameToBytesSourceAdapter(out.asFrameSource(),conn.getDescribedTypeRegistry()) ); conn.setSaslFrameOutput(saslOut); @@ -123,22 +129,24 @@ class TCPTransportProvier implements TransportProvider (byte)1, (byte)0, (byte)0), - new ConnectionHandler.FrameToBytesSourceAdapter(out,conn.getDescribedTypeRegistry()) + new ConnectionHandler.FrameToBytesSourceAdapter(out.asFrameSource(),conn.getDescribedTypeRegistry()) ); } - final OutputStream outputStream = s.getOutputStream(); + final OutputStream outputStream = _socket.getOutputStream(); ConnectionHandler.BytesOutputHandler outputHandler = new ConnectionHandler.BytesOutputHandler(outputStream, src, conn, exceptionHandler); - Thread outputThread = new Thread(outputHandler); + long threadIndex = _threadNameIndex.getAndIncrement(); + Thread outputThread = new Thread(outputHandler, "QpidConnectionOutputThread-"+threadIndex); + outputThread.setDaemon(true); outputThread.start(); conn.setFrameOutputHandler(out); final ConnectionHandler handler = new ConnectionHandler(conn); - final InputStream inputStream = s.getInputStream(); + final InputStream inputStream = _socket.getInputStream(); Thread inputThread = new Thread(new Runnable() { @@ -153,21 +161,11 @@ class TCPTransportProvier implements TransportProvider { if(conn.closedForInput() && conn.closedForOutput()) { - try - { - synchronized (outputStream) - { - s.close(); - } - } - catch (IOException e) - { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. - } + close(); } } } - }); + },"QpidConnectionInputThread-"+threadIndex); inputThread.setDaemon(true); inputThread.start(); @@ -178,6 +176,20 @@ class TCPTransportProvier implements TransportProvider throw new ConnectionException(e); } } + + @Override + public void close() + { + try + { + _socket.close(); + } + catch (IOException e) + { + RAW_LOGGER.log(Level.WARNING, "Unexpected Error during TCPTransportProvider socket close", e); + } + } + private void doRead(final ConnectionEndpoint conn, final ConnectionHandler handler, final InputStream inputStream) { byte[] buf = new byte[2<<15]; diff --git a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/TransportProvider.java b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/TransportProvider.java index 2c11d6b6ef..71628679f8 100644 --- a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/TransportProvider.java +++ b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/TransportProvider.java @@ -20,12 +20,10 @@ */ package org.apache.qpid.amqp_1_0.client; -import org.apache.qpid.amqp_1_0.framing.ConnectionHandler; +import javax.net.ssl.SSLContext; + import org.apache.qpid.amqp_1_0.framing.ExceptionHandler; import org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint; -import org.apache.qpid.amqp_1_0.type.FrameBody; - -import javax.net.ssl.SSLContext; public interface TransportProvider { @@ -34,4 +32,6 @@ public interface TransportProvider int port, SSLContext sslContext, ExceptionHandler exceptionHandler) throws ConnectionException; + + void close(); } |