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 2011/09/07 10:55:14 UTC
svn commit: r1166069 [2/2] - in /qpid/trunk/qpid/java: ./
broker/src/main/java/org/apache/qpid/server/
broker/src/main/java/org/apache/qpid/server/connection/
broker/src/main/java/org/apache/qpid/server/handler/
broker/src/main/java/org/apache/qpid/ser...
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java Wed Sep 7 08:55:12 2011
@@ -37,7 +37,6 @@ public class Transport
public static final String QPID_BROKER_TRANSPORT_PROPNAME = "qpid.broker.transport";
// Can't reference the class directly here, as this would preclude the ability to bundle transports separately.
- private static final String MINA_TRANSPORT_CLASSNAME = "org.apache.qpid.transport.network.mina.MinaNetworkTransport";
private static final String IO_TRANSPORT_CLASSNAME = "org.apache.qpid.transport.network.io.IoNetworkTransport";
public static final String TCP = "tcp";
@@ -47,9 +46,9 @@ public class Transport
static
{
final Map<ProtocolVersion,String> map = new HashMap<ProtocolVersion, String>();
- map.put(ProtocolVersion.v8_0, MINA_TRANSPORT_CLASSNAME);
- map.put(ProtocolVersion.v0_9, MINA_TRANSPORT_CLASSNAME);
- map.put(ProtocolVersion.v0_91, MINA_TRANSPORT_CLASSNAME);
+ map.put(ProtocolVersion.v8_0, IO_TRANSPORT_CLASSNAME);
+ map.put(ProtocolVersion.v0_9, IO_TRANSPORT_CLASSNAME);
+ map.put(ProtocolVersion.v0_91, IO_TRANSPORT_CLASSNAME);
map.put(ProtocolVersion.v0_10, IO_TRANSPORT_CLASSNAME);
OUTGOING_PROTOCOL_TO_IMPLDEFAULTS_MAP = Collections.unmodifiableMap(map);
@@ -58,7 +57,7 @@ public class Transport
public static IncomingNetworkTransport getIncomingTransportInstance()
{
return (IncomingNetworkTransport) loadTransportClass(
- System.getProperty(QPID_BROKER_TRANSPORT_PROPNAME, MINA_TRANSPORT_CLASSNAME));
+ System.getProperty(QPID_BROKER_TRANSPORT_PROPNAME, IO_TRANSPORT_CLASSNAME));
}
public static OutgoingNetworkTransport getOutgoingTransportInstance(
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java Wed Sep 7 08:55:12 2011
@@ -37,7 +37,7 @@ public class IoNetworkConnection impleme
private final long _timeout;
private final IoSender _ioSender;
private final IoReceiver _ioReceiver;
-
+
public IoNetworkConnection(Socket socket, Receiver<ByteBuffer> delegate,
int sendBufferSize, int receiveBufferSize, long timeout)
{
@@ -45,9 +45,15 @@ public class IoNetworkConnection impleme
_timeout = timeout;
_ioReceiver = new IoReceiver(_socket, delegate, receiveBufferSize,_timeout);
+
_ioSender = new IoSender(_socket, 2 * sendBufferSize, _timeout);
+
_ioSender.registerCloseListener(_ioReceiver);
+ }
+
+ public void start()
+ {
_ioReceiver.initiate();
_ioSender.initiate();
}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java Wed Sep 7 08:55:12 2011
@@ -21,22 +21,21 @@
package org.apache.qpid.transport.network.io;
import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.net.SocketException;
+import java.net.*;
import java.nio.ByteBuffer;
import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLServerSocketFactory;
-import org.apache.qpid.transport.ConnectionSettings;
-import org.apache.qpid.transport.Receiver;
-import org.apache.qpid.transport.TransportException;
+import org.apache.qpid.protocol.ProtocolEngine;
+import org.apache.qpid.protocol.ProtocolEngineFactory;
+import org.apache.qpid.transport.*;
+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.util.Logger;
-public class IoNetworkTransport implements OutgoingNetworkTransport
+public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNetworkTransport
{
static
{
@@ -51,12 +50,13 @@ public class IoNetworkTransport implemen
private Socket _socket;
private IoNetworkConnection _connection;
private long _timeout = 60000;
-
+ private AcceptingThread _acceptor;
+
public NetworkConnection connect(ConnectionSettings settings, Receiver<ByteBuffer> delegate, SSLContext sslContext)
{
int sendBufferSize = settings.getWriteBufferSize();
int receiveBufferSize = settings.getReadBufferSize();
-
+
try
{
_socket = new Socket();
@@ -84,6 +84,7 @@ public class IoNetworkTransport implemen
try
{
_connection = new IoNetworkConnection(_socket, delegate, sendBufferSize, receiveBufferSize, _timeout);
+ _connection.start();
}
catch(Exception e)
{
@@ -104,11 +105,134 @@ public class IoNetworkTransport implemen
public void close()
{
- _connection.close();
+ if(_connection != null)
+ {
+ _connection.close();
+ }
+ if(_acceptor != null)
+ {
+ _acceptor.close();
+ }
}
public NetworkConnection getConnection()
{
return _connection;
}
+
+ public void accept(NetworkTransportConfiguration config, ProtocolEngineFactory factory, SSLContext sslContext)
+ {
+
+ try
+ {
+ _acceptor = new AcceptingThread(config, factory, sslContext);
+
+ _acceptor.start();
+ }
+ catch (IOException e)
+ {
+ throw new TransportException("Unable to start server socket", e);
+ }
+
+
+ }
+
+ private class AcceptingThread extends Thread
+ {
+ private NetworkTransportConfiguration _config;
+ private ProtocolEngineFactory _factory;
+ private SSLContext _sslContent;
+ private ServerSocket _serverSocket;
+
+ private AcceptingThread(NetworkTransportConfiguration config,
+ ProtocolEngineFactory factory,
+ SSLContext sslContext)
+ throws IOException
+ {
+ _config = config;
+ _factory = factory;
+ _sslContent = sslContext;
+
+ InetSocketAddress address = new InetSocketAddress(config.getHost(), config.getPort());
+
+ if(sslContext == null)
+ {
+ _serverSocket = new ServerSocket();
+ }
+ else
+ {
+ SSLServerSocketFactory socketFactory = sslContext.getServerSocketFactory();
+ _serverSocket = socketFactory.createServerSocket();
+ }
+
+ _serverSocket.bind(address);
+ _serverSocket.setReuseAddress(true);
+
+
+ }
+
+
+ /**
+ Close the underlying ServerSocket if it has not already been closed.
+ */
+ public void close()
+ {
+ if (!_serverSocket.isClosed())
+ {
+ try
+ {
+ _serverSocket.close();
+ }
+ catch (IOException e)
+ {
+ throw new TransportException(e);
+ }
+ }
+ }
+
+ @Override
+ public void run()
+ {
+ try
+ {
+ while (true)
+ {
+ try
+ {
+ Socket socket = _serverSocket.accept();
+ socket.setTcpNoDelay(_config.getTcpNoDelay());
+
+ final Integer sendBufferSize = _config.getSendBufferSize();
+ final Integer receiveBufferSize = _config.getReceiveBufferSize();
+
+ socket.setSendBufferSize(sendBufferSize);
+ socket.setReceiveBufferSize(receiveBufferSize);
+
+ ProtocolEngine engine = _factory.newProtocolEngine();
+
+ NetworkConnection connection = new IoNetworkConnection(socket, engine, sendBufferSize, receiveBufferSize, _timeout);
+
+
+ engine.setNetworkConnection(connection, connection.getSender());
+
+ connection.start();
+
+
+ }
+ catch(RuntimeException e)
+ {
+ LOGGER.error(e, "Error in Acceptor thread " + _config.getPort());
+ }
+ }
+ }
+ catch (IOException e)
+ {
+ LOGGER.debug(e, "SocketException - no new connections will be accepted on port "
+ + _config.getPort());
+ }
+ }
+
+
+ }
+
}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java Wed Sep 7 08:55:12 2011
@@ -92,13 +92,23 @@ final class IoReceiver implements Runnab
{
try
{
- if (shutdownBroken)
+ try
{
- socket.close();
+ if (shutdownBroken)
+ {
+ socket.close();
+ }
+ else
+ {
+ socket.shutdownInput();
+ }
}
- else
+ catch(SocketException se)
{
- socket.shutdownInput();
+ if(!socket.isClosed() && !socket.isInputShutdown())
+ {
+ throw se;
+ }
}
if (block && Thread.currentThread() != receiverThread)
{
@@ -117,6 +127,7 @@ final class IoReceiver implements Runnab
{
throw new TransportException(e);
}
+
}
}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java Wed Sep 7 08:55:12 2011
@@ -31,6 +31,7 @@ import java.util.concurrent.atomic.Atomi
import org.apache.qpid.common.Closeable;
import org.apache.qpid.thread.Threading;
import org.apache.qpid.transport.Sender;
+import org.apache.qpid.transport.SenderClosedException;
import org.apache.qpid.transport.SenderException;
import org.apache.qpid.transport.TransportException;
import org.apache.qpid.transport.util.Logger;
@@ -59,7 +60,7 @@ public final class IoSender implements R
private final AtomicBoolean closed = new AtomicBoolean(false);
private final Thread senderThread;
private final List<Closeable> _listeners = new ArrayList<Closeable>();
-
+
private volatile Throwable exception = null;
public IoSender(Socket socket, int bufferSize, long timeout)
@@ -80,13 +81,13 @@ public final class IoSender implements R
try
{
//Create but deliberately don't start the thread.
- senderThread = Threading.getThreadFactory().createThread(this);
+ senderThread = Threading.getThreadFactory().createThread(this);
}
catch(Exception e)
{
throw new Error("Error creating IOSender thread",e);
}
-
+
senderThread.setDaemon(true);
senderThread.setName(String.format("IoSender - %s", socket.getRemoteSocketAddress()));
}
@@ -110,7 +111,7 @@ public final class IoSender implements R
{
if (closed.get())
{
- throw new SenderException("sender is closed", exception);
+ throw new SenderClosedException("sender is closed", exception);
}
final int size = buffer.length;
@@ -143,7 +144,7 @@ public final class IoSender implements R
if (closed.get())
{
- throw new SenderException("sender is closed", exception);
+ throw new SenderClosedException("sender is closed", exception);
}
if (head - tail >= size)
@@ -255,7 +256,7 @@ public final class IoSender implements R
public void run()
{
- final int size = buffer.length;
+ final int size = buffer.length;
while (true)
{
final int hd = head;
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java Wed Sep 7 08:55:12 2011
@@ -27,7 +27,6 @@ import javax.net.ssl.SSLEngine;
import org.apache.qpid.ssl.SSLContextFactory;
import org.apache.qpid.transport.Connection;
-import org.apache.qpid.transport.ConnectionListener;
import org.apache.qpid.transport.ConnectionSettings;
import org.apache.qpid.transport.Receiver;
import org.apache.qpid.transport.Sender;
@@ -38,148 +37,12 @@ import org.apache.qpid.transport.network
import org.apache.qpid.transport.network.security.ssl.SSLSender;
import org.apache.qpid.transport.network.security.ssl.SSLUtil;
-public class SecurityLayer
+public interface SecurityLayer
{
- ConnectionSettings settings;
- Connection con;
- SSLSecurityLayer sslLayer;
- SASLSecurityLayer saslLayer;
- public SecurityLayer(Connection con)
- {
- this.con = con;
- this.settings = con.getConnectionSettings();
- if (settings.isUseSSL())
- {
- sslLayer = new SSLSecurityLayer();
- }
- if (settings.isUseSASLEncryption())
- {
- saslLayer = new SASLSecurityLayer();
- }
- }
+ public Sender<ByteBuffer> sender(Sender<ByteBuffer> delegate);
+ public Receiver<ByteBuffer> receiver(Receiver<ByteBuffer> delegate);
+ public String getUserID();
- public Sender<ByteBuffer> sender(Sender<ByteBuffer> delegate)
- {
- Sender<ByteBuffer> sender = delegate;
-
- if (settings.isUseSSL())
- {
- sender = sslLayer.sender(sender);
- }
-
- if (settings.isUseSASLEncryption())
- {
- sender = saslLayer.sender(sender);
- }
-
- return sender;
- }
-
- public Receiver<ByteBuffer> receiver(Receiver<ByteBuffer> delegate)
- {
- Receiver<ByteBuffer> receiver = delegate;
-
- if (settings.isUseSSL())
- {
- receiver = sslLayer.receiver(receiver);
- }
-
- if (settings.isUseSASLEncryption())
- {
- receiver = saslLayer.receiver(receiver);
- }
-
- return receiver;
- }
-
- public String getUserID()
- {
- if (settings.isUseSSL())
- {
- return sslLayer.getUserID();
- }
- else
- {
- return null;
- }
- }
-
- class SSLSecurityLayer
- {
- final SSLEngine _engine;
- final SSLStatus _sslStatus = new SSLStatus();
-
- public SSLSecurityLayer()
- {
- SSLContext sslCtx;
- try
- {
- sslCtx = SSLContextFactory
- .buildClientContext(settings.getTrustStorePath(),
- settings.getTrustStorePassword(),
- settings.getTrustStoreCertType(),
- settings.getKeyStorePath(),
- settings.getKeyStorePassword(),
- settings.getKeyStoreCertType(),
- settings.getCertAlias());
- }
- catch (Exception e)
- {
- throw new TransportException("Error creating SSL Context", e);
- }
-
- try
- {
- _engine = sslCtx.createSSLEngine();
- _engine.setUseClientMode(true);
- }
- catch(Exception e)
- {
- throw new TransportException("Error creating SSL Engine", e);
- }
- }
-
- public SSLSender sender(Sender<ByteBuffer> delegate)
- {
- SSLSender sender = new SSLSender(_engine, delegate, _sslStatus);
- sender.setConnectionSettings(settings);
- return sender;
- }
-
- public SSLReceiver receiver(Receiver<ByteBuffer> delegate)
- {
- SSLReceiver receiver = new SSLReceiver(_engine, delegate, _sslStatus);
- receiver.setConnectionSettings(settings);
- return receiver;
- }
-
- public String getUserID()
- {
- return SSLUtil.retriveIdentity(_engine);
- }
-
- }
-
- class SASLSecurityLayer
- {
- public SASLSecurityLayer()
- {
- }
-
- public SASLSender sender(Sender<ByteBuffer> delegate)
- {
- SASLSender sender = new SASLSender(delegate);
- con.addConnectionListener((ConnectionListener)sender);
- return sender;
- }
-
- public SASLReceiver receiver(Receiver<ByteBuffer> delegate)
- {
- SASLReceiver receiver = new SASLReceiver(delegate);
- con.addConnectionListener((ConnectionListener)receiver);
- return receiver;
- }
-
- }
}
+
Added: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayerFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayerFactory.java?rev=1166069&view=auto
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayerFactory.java (added)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayerFactory.java Wed Sep 7 08:55:12 2011
@@ -0,0 +1,161 @@
+package org.apache.qpid.transport.network.security;
+
+import org.apache.qpid.ssl.SSLContextFactory;
+import org.apache.qpid.transport.*;
+import org.apache.qpid.transport.network.security.sasl.SASLReceiver;
+import org.apache.qpid.transport.network.security.sasl.SASLSender;
+import org.apache.qpid.transport.network.security.ssl.SSLReceiver;
+import org.apache.qpid.transport.network.security.ssl.SSLSender;
+import org.apache.qpid.transport.network.security.ssl.SSLUtil;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import java.nio.ByteBuffer;
+
+public class SecurityLayerFactory
+{
+ public static SecurityLayer newInstance(ConnectionSettings settings)
+ {
+
+ SecurityLayer layer = NullSecurityLayer.getInstance();
+
+ if (settings.isUseSSL())
+ {
+ layer = new SSLSecurityLayer(settings, layer);
+ }
+ if (settings.isUseSASLEncryption())
+ {
+ layer = new SASLSecurityLayer(layer);
+ }
+
+ return layer;
+
+ }
+
+ static class SSLSecurityLayer implements SecurityLayer
+ {
+
+ private final SSLEngine _engine;
+ private final SSLStatus _sslStatus = new SSLStatus();
+ private String _hostname;
+ private SecurityLayer _layer;
+
+
+ public SSLSecurityLayer(ConnectionSettings settings, SecurityLayer layer)
+ {
+
+ SSLContext sslCtx;
+ _layer = layer;
+ try
+ {
+ sslCtx = SSLContextFactory
+ .buildClientContext(settings.getTrustStorePath(),
+ settings.getTrustStorePassword(),
+ settings.getTrustStoreCertType(),
+ settings.getKeyStorePath(),
+ settings.getKeyStorePassword(),
+ settings.getKeyStoreCertType(),
+ settings.getCertAlias());
+ }
+ catch (Exception e)
+ {
+ throw new TransportException("Error creating SSL Context", e);
+ }
+
+ if(settings.isVerifyHostname())
+ {
+ _hostname = settings.getHost();
+ }
+
+ try
+ {
+ _engine = sslCtx.createSSLEngine();
+ _engine.setUseClientMode(true);
+ }
+ catch(Exception e)
+ {
+ throw new TransportException("Error creating SSL Engine", e);
+ }
+
+ }
+
+ public Sender<ByteBuffer> sender(Sender<ByteBuffer> delegate)
+ {
+ SSLSender sender = new SSLSender(_engine, _layer.sender(delegate), _sslStatus);
+ sender.setHostname(_hostname);
+ return sender;
+ }
+
+ public Receiver<ByteBuffer> receiver(Receiver<ByteBuffer> delegate)
+ {
+ SSLReceiver receiver = new SSLReceiver(_engine, _layer.receiver(delegate), _sslStatus);
+ receiver.setHostname(_hostname);
+ return receiver;
+ }
+
+ public String getUserID()
+ {
+ return SSLUtil.retriveIdentity(_engine);
+ }
+ }
+
+
+ static class SASLSecurityLayer implements SecurityLayer
+ {
+
+ private SecurityLayer _layer;
+
+ SASLSecurityLayer(SecurityLayer layer)
+ {
+ _layer = layer;
+ }
+
+ public SASLSender sender(Sender<ByteBuffer> delegate)
+ {
+ SASLSender sender = new SASLSender(_layer.sender(delegate));
+ return sender;
+ }
+
+ public SASLReceiver receiver(Receiver<ByteBuffer> delegate)
+ {
+ SASLReceiver receiver = new SASLReceiver(_layer.receiver(delegate));
+ return receiver;
+ }
+
+ public String getUserID()
+ {
+ return _layer.getUserID();
+ }
+ }
+
+
+ static class NullSecurityLayer implements SecurityLayer
+ {
+
+ private static final NullSecurityLayer INSTANCE = new NullSecurityLayer();
+
+ private NullSecurityLayer()
+ {
+ }
+
+ public Sender<ByteBuffer> sender(Sender<ByteBuffer> delegate)
+ {
+ return delegate;
+ }
+
+ public Receiver<ByteBuffer> receiver(Receiver<ByteBuffer> delegate)
+ {
+ return delegate;
+ }
+
+ public String getUserID()
+ {
+ return null;
+ }
+
+ public static NullSecurityLayer getInstance()
+ {
+ return INSTANCE;
+ }
+ }
+}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java Wed Sep 7 08:55:12 2011
@@ -43,10 +43,11 @@ public class SSLReceiver implements Rece
private final int sslBufSize;
private final ByteBuffer localBuffer;
private final SSLStatus _sslStatus;
- private ConnectionSettings settings;
private ByteBuffer appData;
private boolean dataCached = false;
+ private String _hostname;
+
public SSLReceiver(final SSLEngine engine, final Receiver<ByteBuffer> delegate, final SSLStatus sslStatus)
{
this.engine = engine;
@@ -57,9 +58,9 @@ public class SSLReceiver implements Rece
_sslStatus = sslStatus;
}
- public void setConnectionSettings(ConnectionSettings settings)
+ public void setHostname(String hostname)
{
- this.settings = settings;
+ _hostname = hostname;
}
public void closed()
@@ -166,9 +167,9 @@ public class SSLReceiver implements Rece
handshakeStatus = engine.getHandshakeStatus();
case FINISHED:
- if (this.settings != null && this.settings.isVerifyHostname() )
+ if (_hostname != null)
{
- SSLUtil.verifyHostname(engine, this.settings.getHost());
+ SSLUtil.verifyHostname(engine, _hostname);
}
case NEED_WRAP:
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java Wed Sep 7 08:55:12 2011
@@ -44,8 +44,9 @@ public class SSLSender implements Sender
private final ByteBuffer netData;
private final long timeout;
private final SSLStatus _sslStatus;
- private ConnectionSettings settings;
-
+
+ private String _hostname;
+
private final AtomicBoolean closed = new AtomicBoolean(false);
@@ -59,9 +60,9 @@ public class SSLSender implements Sender
_sslStatus = sslStatus;
}
- public void setConnectionSettings(ConnectionSettings settings)
+ public void setHostname(String hostname)
{
- this.settings = settings;
+ _hostname = hostname;
}
public void close()
@@ -237,9 +238,9 @@ public class SSLSender implements Sender
break;
case FINISHED:
- if (this.settings != null && this.settings.isVerifyHostname() )
+ if (_hostname != null)
{
- SSLUtil.verifyHostname(engine, this.settings.getHost());
+ SSLUtil.verifyHostname(engine, _hostname);
}
case NOT_HANDSHAKING:
Modified: qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java (original)
+++ qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java Wed Sep 7 08:55:12 2011
@@ -49,10 +49,12 @@ public class TestNetworkConnection imple
_sender = new MockSender();
}
+
+
public void bind(int port, InetAddress[] addresses, ProtocolEngineFactory protocolFactory,
NetworkTransportConfiguration config, SSLContextFactory sslFactory) throws BindException
{
-
+
}
public SocketAddress getLocalAddress()
@@ -68,37 +70,37 @@ public class TestNetworkConnection imple
public void open(int port, InetAddress destination, ProtocolEngine engine, NetworkTransportConfiguration config,
SSLContextFactory sslFactory) throws OpenException
{
-
+
}
public void setMaxReadIdle(int idleTime)
{
-
+
}
public void setMaxWriteIdle(int idleTime)
{
-
+
}
public void close()
{
-
+
}
public void flush()
{
-
+
}
public void send(ByteBuffer msg)
{
-
+
}
public void setIdleTimeout(int i)
{
-
+
}
public void setPort(int port)
@@ -135,4 +137,8 @@ public class TestNetworkConnection imple
{
return _sender;
}
+
+ public void start()
+ {
+ }
}
Modified: qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java (original)
+++ qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java Wed Sep 7 08:55:12 2011
@@ -33,7 +33,6 @@ import org.apache.qpid.transport.Network
import org.apache.qpid.transport.Receiver;
import org.apache.qpid.transport.TransportException;
import org.apache.qpid.transport.network.io.IoNetworkTransport;
-import org.apache.qpid.transport.network.mina.MinaNetworkTransport;
public class TransportTest extends QpidTestCase
{
@@ -44,7 +43,7 @@ public class TransportTest extends QpidT
{
final OutgoingNetworkTransport networkTransport = Transport.getOutgoingTransportInstance(ProtocolVersion.v8_0);
assertNotNull(networkTransport);
- assertTrue(networkTransport instanceof MinaNetworkTransport);
+ assertTrue(networkTransport instanceof IoNetworkTransport);
}
public void testGloballyOverriddenOutgoingTransportForv0_8() throws Exception
@@ -76,7 +75,7 @@ public class TransportTest extends QpidT
{
final IncomingNetworkTransport networkTransport = Transport.getIncomingTransportInstance();
assertNotNull(networkTransport);
- assertTrue(networkTransport instanceof MinaNetworkTransport);
+ assertTrue(networkTransport instanceof IoNetworkTransport);
}
public void testOverriddenGetIncomingTransport() throws Exception
Modified: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/failover/FailoverMethodTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/failover/FailoverMethodTest.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/failover/FailoverMethodTest.java (original)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/failover/FailoverMethodTest.java Wed Sep 7 08:55:12 2011
@@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
+import org.apache.qpid.AMQConnectionClosedException;
import org.apache.qpid.AMQDisconnectedException;
import org.apache.qpid.AMQException;
import org.apache.qpid.client.AMQConnection;
@@ -139,7 +140,7 @@ public class FailoverMethodTest extends
public void onException(JMSException e)
{
- if (e.getLinkedException() instanceof AMQDisconnectedException)
+ if (e.getLinkedException() instanceof AMQDisconnectedException || e.getLinkedException() instanceof AMQConnectionClosedException)
{
_logger.debug("Received AMQDisconnectedException");
_failoverComplete.countDown();
Modified: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/ExternalACLTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/ExternalACLTest.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/ExternalACLTest.java (original)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/ExternalACLTest.java Wed Sep 7 08:55:12 2011
@@ -67,14 +67,14 @@ public class ExternalACLTest extends Abs
fail("Connection was not created due to:" + e);
}
}
-
+
public void testAccessVhostAuthorisedGuestSuccess() throws IOException, Exception
{
//The 'guest' user has no access to the 'test' vhost, as tested below in testAccessNoRights(), and so
- //is unable to perform actions such as connecting (and by extension, creating a queue, and consuming
- //from a queue etc). In order to test the vhost-wide 'access' ACL right, the 'guest' user has been given
+ //is unable to perform actions such as connecting (and by extension, creating a queue, and consuming
+ //from a queue etc). In order to test the vhost-wide 'access' ACL right, the 'guest' user has been given
//this right in the 'test2' vhost.
-
+
try
{
//get a connection to the 'test2' vhost using the guest user and perform various actions.
@@ -106,7 +106,7 @@ public class ExternalACLTest extends Abs
fail("Test failed due to:" + e.getMessage());
}
}
-
+
public void testAccessNoRightsFailure() throws Exception
{
try
@@ -115,7 +115,7 @@ public class ExternalACLTest extends Abs
Session sess = conn.createSession(true, Session.SESSION_TRANSACTED);
conn.start();
sess.rollback();
-
+
fail("Connection was created.");
}
catch (JMSException e)
@@ -126,11 +126,11 @@ public class ExternalACLTest extends Abs
Throwable cause = linkedException.getCause();
assertNotNull("Cause was null", cause);
assertTrue("Wrong linked exception type", cause instanceof AMQException);
- AMQConstant errorCode = isBroker010() ? AMQConstant.CONTEXT_IN_USE : AMQConstant.ACCESS_REFUSED;
+ AMQConstant errorCode = isBroker010() ? AMQConstant.CONNECTION_FORCED : AMQConstant.ACCESS_REFUSED;
assertEquals("Incorrect error code received", errorCode, ((AMQException) cause).getErrorCode());
}
}
-
+
public void testClientDeleteQueueSuccess() throws Exception
{
try
@@ -155,7 +155,7 @@ public class ExternalACLTest extends Abs
fail("Test failed due to:" + e.getMessage());
}
}
-
+
public void testServerDeleteQueueFailure() throws Exception
{
try
@@ -207,13 +207,13 @@ public class ExternalACLTest extends Abs
try
{
Connection conn = getConnection("test", "client", "guest");
-
+
Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
conn.start();
sess.createConsumer(sess.createQueue("IllegalQueue"));
-
+
fail("Test failed as consumer was created.");
}
catch (JMSException e)
@@ -253,10 +253,10 @@ public class ExternalACLTest extends Abs
Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
conn.start();
-
+
//Create a Named Queue
((AMQSession<?, ?>) sess).createQueue(new AMQShortString("IllegalQueue"), false, false, false);
-
+
fail("Test failed as Queue creation succeded.");
//conn will be automatically closed
}
@@ -385,7 +385,7 @@ public class ExternalACLTest extends Abs
try
{
Connection conn = getConnection("test", "client", "guest");
-
+
Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
conn.start();
@@ -405,13 +405,13 @@ public class ExternalACLTest extends Abs
try
{
Connection conn = getConnection("test", "server", "guest");
-
+
Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
conn.start();
sess.createConsumer(sess.createTemporaryQueue());
-
+
fail("Test failed as consumer was created.");
}
catch (JMSException e)
@@ -446,7 +446,7 @@ public class ExternalACLTest extends Abs
try
{
Connection conn = getConnection("test", "server", "guest");
-
+
Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
conn.start();
@@ -480,13 +480,13 @@ public class ExternalACLTest extends Abs
check403Exception(e.getLinkedException());
}
}
-
+
public void testServerCreateAutoDeleteQueueInvalid() throws NamingException, JMSException, AMQException, Exception
{
try
{
Connection connection = getConnection("test", "server", "guest");
-
+
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
connection.start();
@@ -630,8 +630,8 @@ public class ExternalACLTest extends Abs
check403Exception(e.getLinkedException());
}
}
-
-
+
+
@Override
public String getConfig()
{
Modified: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserAutoAckTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserAutoAckTest.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserAutoAckTest.java (original)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserAutoAckTest.java Wed Sep 7 08:55:12 2011
@@ -59,7 +59,7 @@ public class QueueBrowserAutoAckTest ext
_queue = _clientSession.createQueue(getTestQueueName());
_clientSession.createConsumer(_queue).close();
-
+
//Ensure there are no messages on the queue to start with.
checkQueueDepth(0);
}
@@ -490,7 +490,7 @@ public class QueueBrowserAutoAckTest ext
}
}
- assertTrue("We should get atleast " + messages + " msgs.", msgCount >= messages);
+ assertTrue("We should get atleast " + messages + " msgs (found " + msgCount +").", msgCount >= messages);
if (_logger.isDebugEnabled())
{
Modified: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/ExceptionListenerTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/ExceptionListenerTest.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/ExceptionListenerTest.java (original)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/ExceptionListenerTest.java Wed Sep 7 08:55:12 2011
@@ -47,13 +47,15 @@ public class ExceptionListenerTest exten
{
public void onException(JMSException e)
{
+ _logger.debug("&&&&&&&&&&&&&&&&&&&&&&&&&&&& Caught exception &&&&&&&&&&&&&&&&&&&&&&&&&&&& ", e);
fired.countDown();
}
});
-
+ _logger.debug("%%%%%%%%%%%%%%%% Stopping Broker %%%%%%%%%%%%%%%%%%%%%");
stopBroker();
+ _logger.debug("%%%%%%%%%%%%%%%% Stopped Broker %%%%%%%%%%%%%%%%%%%%%");
- if (!fired.await(3, TimeUnit.SECONDS))
+ if (!fired.await(5, TimeUnit.SECONDS))
{
fail("exception listener was not fired");
}
Modified: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java (original)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java Wed Sep 7 08:55:12 2011
@@ -25,7 +25,6 @@ import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.UnknownHostException;
-import org.apache.mina.transport.vmpipe.VmPipeAddress;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.protocol.AMQProtocolHandler;
import org.apache.qpid.client.protocol.AMQProtocolSession;
Modified: qpid/trunk/qpid/java/test-profiles/Excludes
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/test-profiles/Excludes?rev=1166069&r1=1166068&r2=1166069&view=diff
==============================================================================
--- qpid/trunk/qpid/java/test-profiles/Excludes (original)
+++ qpid/trunk/qpid/java/test-profiles/Excludes Wed Sep 7 08:55:12 2011
@@ -43,3 +43,5 @@ org.apache.qpid.test.unit.ack.Acknowledg
// QPID-2418 : The queue backing the dur sub is not currently deleted at subscription change, so the test will fail.
org.apache.qpid.test.unit.ct.DurableSubscriberTest#testResubscribeWithChangedSelectorAndRestart
+
+org.apache.qpid.client.ssl.SSLTest#testVerifyLocalHostLocalDomain
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org