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 2016/07/11 21:37:17 UTC
[2/2] qpid-jms git commit: QPIDJMS-191 Add support for WebSockets
connections
QPIDJMS-191 Add support for WebSockets connections
Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/91fd06d6
Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/91fd06d6
Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/91fd06d6
Branch: refs/heads/master
Commit: 91fd06d65828540db3651a097578f4218f64273d
Parents: 3f0b2e3
Author: Timothy Bish <ta...@gmail.com>
Authored: Mon Jul 11 17:37:07 2016 -0400
Committer: Timothy Bish <ta...@gmail.com>
Committed: Mon Jul 11 17:37:07 2016 -0400
----------------------------------------------------------------------
pom.xml | 9 +-
qpid-jms-client/pom.xml | 4 +
.../qpid/jms/provider/amqp/AmqpProvider.java | 12 +-
.../qpid/jms/transports/SSLTransport.java | 23 --
.../apache/qpid/jms/transports/Transport.java | 15 +-
.../qpid/jms/transports/TransportOptions.java | 4 +
.../jms/transports/TransportSslOptions.java | 5 +
.../jms/transports/netty/NettySslTransport.java | 120 ---------
.../netty/NettySslTransportFactory.java | 10 +-
.../jms/transports/netty/NettyTcpTransport.java | 263 ++++++++++++-------
.../jms/transports/netty/NettyWsTransport.java | 160 +++++++++++
.../netty/NettyWsTransportFactory.java | 38 +++
.../netty/NettyWssTransportFactory.java | 36 +++
.../org/apache/qpid/jms/provider/amqpws | 18 ++
.../org/apache/qpid/jms/provider/amqpwss | 18 ++
.../services/org/apache/qpid/jms/transports/ws | 17 ++
.../services/org/apache/qpid/jms/transports/wss | 17 ++
.../jms/transports/netty/NettyEchoServer.java | 70 +++--
.../netty/NettySslTransportFactoryTest.java | 6 +-
.../transports/netty/NettySslTransportTest.java | 34 ++-
.../transports/netty/NettyTcpTransportTest.java | 44 ++--
.../transports/netty/NettyWsTransportTest.java | 43 +++
.../transports/netty/NettyWssTransportTest.java | 42 +++
.../qpid-jms-activemq-tests/pom.xml | 23 +-
.../apache/qpid/jms/JmsWSConnectionTest.java | 93 +++++++
.../apache/qpid/jms/JmsWSSConnectionTest.java | 123 +++++++++
.../src/test/resources/log4j.properties | 11 +-
27 files changed, 939 insertions(+), 319 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index a4598e6..0d5d6b3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -44,14 +44,14 @@
<!-- Dependency Versions for this Project -->
<proton-version>0.13.0</proton-version>
- <netty-version>4.0.17.Final</netty-version>
+ <netty-version>4.0.38.Final</netty-version>
<slf4j-version>1.7.21</slf4j-version>
<geronimo-jms-1-1-spec-version>1.1.1</geronimo-jms-1-1-spec-version>
<!-- Test Dependency Versions for this Project -->
<activemq-version>5.13.3</activemq-version>
<junit-version>4.12</junit-version>
<fuse-joram-tests-version>1.0</fuse-joram-tests-version>
- <jetty-version>8.1.17.v20150415</jetty-version>
+ <jetty-version>9.2.13.v20150730</jetty-version>
<mockito-version>1.10.19</mockito-version>
<hamcrest-version>1.3</hamcrest-version>
@@ -137,6 +137,11 @@
<version>${netty-version}</version>
</dependency>
<dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-codec-http</artifactId>
+ <version>${netty-version}</version>
+ </dependency>
+ <dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>${slf4j-version}</version>
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/pom.xml
----------------------------------------------------------------------
diff --git a/qpid-jms-client/pom.xml b/qpid-jms-client/pom.xml
index 3e7a15a..2b01261 100644
--- a/qpid-jms-client/pom.xml
+++ b/qpid-jms-client/pom.xml
@@ -60,6 +60,10 @@
<groupId>io.netty</groupId>
<artifactId>netty-transport</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-codec-http</artifactId>
+ </dependency>
<!-- =================================== -->
<!-- Testing Dependencies -->
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/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 4dc1659..b2ece7f 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
@@ -19,7 +19,6 @@ package org.apache.qpid.jms.provider.amqp;
import java.io.IOException;
import java.net.URI;
import java.nio.ByteBuffer;
-import java.security.Principal;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
@@ -55,7 +54,6 @@ import org.apache.qpid.jms.provider.ProviderConstants.ACK_TYPE;
import org.apache.qpid.jms.provider.ProviderFuture;
import org.apache.qpid.jms.provider.ProviderListener;
import org.apache.qpid.jms.provider.amqp.builders.AmqpConnectionBuilder;
-import org.apache.qpid.jms.transports.SSLTransport;
import org.apache.qpid.jms.transports.TransportFactory;
import org.apache.qpid.jms.transports.TransportListener;
import org.apache.qpid.jms.util.IOExceptionSupport;
@@ -296,7 +294,7 @@ public class AmqpProvider implements Provider, TransportListener , AmqpResourceP
sasl.setRemoteHostname(hostname);
- authenticator = new AmqpSaslAuthenticator(sasl, connectionInfo, getLocalPrincipal(), saslMechanisms);
+ authenticator = new AmqpSaslAuthenticator(sasl, connectionInfo, transport.getLocalPrincipal(), saslMechanisms);
}
AmqpConnectionBuilder builder = new AmqpConnectionBuilder(AmqpProvider.this, connectionInfo);
@@ -1172,14 +1170,6 @@ public class AmqpProvider implements Provider, TransportListener , AmqpResourceP
return null;
}
- Principal getLocalPrincipal() {
- if (transport instanceof SSLTransport) {
- return ((SSLTransport) transport).getLocalPrincipal();
- }
-
- return null;
- }
-
//----- Internal implementation ------------------------------------------//
private void checkClosed() throws ProviderClosedException {
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/SSLTransport.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/SSLTransport.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/SSLTransport.java
deleted file mode 100644
index dfe5c92..0000000
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/SSLTransport.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.qpid.jms.transports;
-
-import java.security.Principal;
-
-public interface SSLTransport extends Transport {
- Principal getLocalPrincipal();
-}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/Transport.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/Transport.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/Transport.java
index 673b80b..26843ba 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/Transport.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/Transport.java
@@ -16,10 +16,11 @@
*/
package org.apache.qpid.jms.transports;
-import io.netty.buffer.ByteBuf;
-
import java.io.IOException;
import java.net.URI;
+import java.security.Principal;
+
+import io.netty.buffer.ByteBuf;
/**
* Base class for all QpidJMS Transport instances.
@@ -40,6 +41,11 @@ public interface Transport {
boolean isConnected();
/**
+ * @return true if transport is connected using a secured channel (SSL).
+ */
+ boolean isSecure();
+
+ /**
* Close the Transport, no additional send operations are accepted.
*
* @throws IOException if an error occurs while closing the connection.
@@ -97,4 +103,9 @@ public interface Transport {
*/
URI getRemoteLocation();
+ /**
+ * @return the local principal for a Transport that is using a secure connection.
+ */
+ Principal getLocalPrincipal();
+
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportOptions.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportOptions.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportOptions.java
index 78a558f..34825c0 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportOptions.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportOptions.java
@@ -168,6 +168,10 @@ public class TransportOptions implements Cloneable {
return copyOptions(new TransportOptions());
}
+ public boolean isSSL() {
+ return false;
+ }
+
protected TransportOptions copyOptions(TransportOptions copy) {
copy.setConnectTimeout(getConnectTimeout());
copy.setReceiveBufferSize(getReceiveBufferSize());
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportSslOptions.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportSslOptions.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportSslOptions.java
index 34bd727..834f022 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportSslOptions.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportSslOptions.java
@@ -267,6 +267,11 @@ public class TransportSslOptions extends TransportOptions {
return copyOptions(new TransportSslOptions());
}
+ @Override
+ public boolean isSSL() {
+ return true;
+ }
+
protected TransportSslOptions copyOptions(TransportSslOptions copy) {
super.copyOptions(copy);
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettySslTransport.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettySslTransport.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettySslTransport.java
deleted file mode 100644
index 740b685..0000000
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettySslTransport.java
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.qpid.jms.transports.netty;
-
-import io.netty.channel.Channel;
-import io.netty.handler.ssl.SslHandler;
-import io.netty.util.concurrent.Future;
-import io.netty.util.concurrent.GenericFutureListener;
-
-import java.net.URI;
-import java.security.Principal;
-
-import org.apache.qpid.jms.transports.SSLTransport;
-import org.apache.qpid.jms.transports.TransportListener;
-import org.apache.qpid.jms.transports.TransportOptions;
-import org.apache.qpid.jms.transports.TransportSslOptions;
-import org.apache.qpid.jms.transports.TransportSupport;
-import org.apache.qpid.jms.util.IOExceptionSupport;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Extends the Netty based TCP transport to add SSL support.
- */
-public class NettySslTransport extends NettyTcpTransport implements SSLTransport {
-
- private static final Logger LOG = LoggerFactory.getLogger(NettySslTransport.class);
-
- /**
- * Create a new transport instance
- *
- * @param remoteLocation
- * the URI that defines the remote resource to connect to.
- * @param options
- * the transport options used to configure the socket connection.
- */
- public NettySslTransport(URI remoteLocation, TransportOptions options) {
- this(null, remoteLocation, options);
- }
-
- /**
- * Create a new transport instance
- *
- * @param listener
- * the TransportListener that will receive events from this Transport.
- * @param remoteLocation
- * the URI that defines the remote resource to connect to.
- * @param options
- * the transport options used to configure the socket connection.
- */
- public NettySslTransport(TransportListener listener, URI remoteLocation, TransportOptions options) {
- super(listener, remoteLocation, options);
- }
-
- @Override
- protected void configureChannel(final Channel channel) throws Exception {
- SslHandler sslHandler = TransportSupport.createSslHandler(getRemoteLocation(), getSslOptions());
- sslHandler.handshakeFuture().addListener(new GenericFutureListener<Future<Channel>>() {
- @Override
- public void operationComplete(Future<Channel> future) throws Exception {
- if (future.isSuccess()) {
- LOG.trace("SSL Handshake has completed: {}", channel);
- connectionEstablished(channel);
- } else {
- LOG.trace("SSL Handshake has failed: {}", channel);
- connectionFailed(channel, IOExceptionSupport.create(future.cause()));
- }
- }
- });
-
- channel.pipeline().addLast(sslHandler);
-
- super.configureChannel(channel);
- }
-
- @Override
- protected void handleConnected(final Channel channel) throws Exception {
- // In this transport, the next step is taken by the handshake future
- // completion listener added when configuring the channel above
- }
-
- @Override
- public TransportOptions getTransportOptions() {
- if (options == null) {
- options = TransportSslOptions.INSTANCE;
- }
-
- return options;
- }
-
- @Override
- protected int getRemotePort() {
- return remote.getPort() != -1 ? remote.getPort() : getSslOptions().getDefaultSslPort();
- }
-
- private TransportSslOptions getSslOptions() {
- return (TransportSslOptions) getTransportOptions();
- }
-
- @Override
- public Principal getLocalPrincipal() {
- SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
-
- return sslHandler.engine().getSession().getLocalPrincipal();
- }
-}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettySslTransportFactory.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettySslTransportFactory.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettySslTransportFactory.java
index 6cec43b..9eda198 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettySslTransportFactory.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettySslTransportFactory.java
@@ -16,16 +16,13 @@
*/
package org.apache.qpid.jms.transports.netty;
-import java.net.URI;
-
-import org.apache.qpid.jms.transports.TransportFactory;
import org.apache.qpid.jms.transports.TransportOptions;
import org.apache.qpid.jms.transports.TransportSslOptions;
/**
* Creates a Netty based SSL transport.
*/
-public class NettySslTransportFactory extends TransportFactory {
+public class NettySslTransportFactory extends NettyTcpTransportFactory {
@Override
protected TransportOptions doCreateTransportOptions() {
@@ -33,11 +30,6 @@ public class NettySslTransportFactory extends TransportFactory {
}
@Override
- protected NettyTcpTransport doCreateTransport(URI remoteURI, TransportOptions transportOptions) throws Exception {
- return new NettySslTransport(remoteURI, transportOptions);
- }
-
- @Override
public String getName() {
return "SSL";
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/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 e900e60..2a49a3e 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
@@ -16,32 +16,40 @@
*/
package org.apache.qpid.jms.transports.netty;
+import java.io.IOException;
+import java.net.URI;
+import java.security.Principal;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.qpid.jms.transports.Transport;
+import org.apache.qpid.jms.transports.TransportListener;
+import org.apache.qpid.jms.transports.TransportOptions;
+import org.apache.qpid.jms.transports.TransportSslOptions;
+import org.apache.qpid.jms.transports.TransportSupport;
+import org.apache.qpid.jms.util.IOExceptionSupport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.FixedRecvByteBufAllocator;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
-
-import java.io.IOException;
-import java.net.URI;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.apache.qpid.jms.transports.Transport;
-import org.apache.qpid.jms.transports.TransportListener;
-import org.apache.qpid.jms.transports.TransportOptions;
-import org.apache.qpid.jms.util.IOExceptionSupport;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
/**
* TCP based transport that uses Netty as the underlying IO layer.
@@ -57,9 +65,9 @@ public class NettyTcpTransport implements Transport {
protected EventLoopGroup group;
protected Channel channel;
protected TransportListener listener;
- protected TransportOptions options;
- protected final URI remote;
+ private final TransportOptions options;
+ private final URI remote;
private final AtomicBoolean connected = new AtomicBoolean();
private final AtomicBoolean closed = new AtomicBoolean();
private final CountDownLatch connectLatch = new CountDownLatch(1);
@@ -89,6 +97,14 @@ public class NettyTcpTransport implements Transport {
* the transport options used to configure the socket connection.
*/
public NettyTcpTransport(TransportListener listener, URI remoteLocation, TransportOptions options) {
+ if (options == null) {
+ throw new IllegalArgumentException("Transport Options cannot be null");
+ }
+
+ if (remoteLocation == null) {
+ throw new IllegalArgumentException("Transport remote location cannot be null");
+ }
+
this.options = options;
this.listener = listener;
this.remote = remoteLocation;
@@ -117,19 +133,9 @@ public class NettyTcpTransport implements Transport {
configureNetty(bootstrap, getTransportOptions());
ChannelFuture future = bootstrap.connect(getRemoteHost(), getRemotePort());
- future.addListener(new ChannelFutureListener() {
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- if (future.isSuccess()) {
- handleConnected(future.channel());
- } else if (future.isCancelled()) {
- connectionFailed(future.channel(), new IOException("Connection attempt was cancelled"));
- } else {
- connectionFailed(future.channel(), IOExceptionSupport.create(future.cause()));
- }
- }
- });
+ // Route all events through the channel handlers for consistent processing.
+ future.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
try {
connectLatch.await();
@@ -171,6 +177,11 @@ public class NettyTcpTransport implements Transport {
}
@Override
+ public boolean isSecure() {
+ return options.isSSL();
+ }
+
+ @Override
public void close() throws IOException {
if (closed.compareAndSet(false, true)) {
connected.set(false);
@@ -214,10 +225,6 @@ public class NettyTcpTransport implements Transport {
@Override
public TransportOptions getTransportOptions() {
- if (options == null) {
- options = TransportOptions.INSTANCE;
- }
-
return options;
}
@@ -226,112 +233,189 @@ public class NettyTcpTransport implements Transport {
return remote;
}
- //----- Internal implementation details, can be overridden as needed --//
+ @Override
+ public Principal getLocalPrincipal() {
+ Principal result = null;
+
+ if (isSecure()) {
+ SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
+ result = sslHandler.engine().getSession().getLocalPrincipal();
+ }
+
+ return result;
+ }
+
+ //----- Internal implementation details, can be overridden as needed -----//
protected String getRemoteHost() {
return remote.getHost();
}
protected int getRemotePort() {
- return remote.getPort() != -1 ? remote.getPort() : getTransportOptions().getDefaultTcpPort();
- }
-
- protected void configureNetty(Bootstrap bootstrap, TransportOptions options) {
- bootstrap.option(ChannelOption.TCP_NODELAY, options.isTcpNoDelay());
- bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, options.getConnectTimeout());
- bootstrap.option(ChannelOption.SO_KEEPALIVE, options.isTcpKeepAlive());
- bootstrap.option(ChannelOption.SO_LINGER, options.getSoLinger());
- bootstrap.option(ChannelOption.ALLOCATOR, PartialPooledByteBufAllocator.INSTANCE);
-
- if (options.getSendBufferSize() != -1) {
- bootstrap.option(ChannelOption.SO_SNDBUF, options.getSendBufferSize());
+ if (remote.getPort() != -1) {
+ return remote.getPort();
+ } else {
+ return isSecure() ? getSslOptions().getDefaultSslPort() : getTransportOptions().getDefaultTcpPort();
}
+ }
- if (options.getReceiveBufferSize() != -1) {
- bootstrap.option(ChannelOption.SO_RCVBUF, options.getReceiveBufferSize());
- bootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(options.getReceiveBufferSize()));
- }
+ protected void addAdditionalHandlers(ChannelPipeline pipeline) {
- if (options.getTrafficClass() != -1) {
- bootstrap.option(ChannelOption.IP_TOS, options.getTrafficClass());
- }
}
- protected void configureChannel(Channel channel) throws Exception {
- channel.pipeline().addLast(new NettyTcpTransportHandler());
+ protected ChannelInboundHandlerAdapter getChannelHandler() {
+ return new NettyTcpTransportHandler();
}
+ //----- Event Handlers which can be overridden in subclasses -------------//
+
protected void handleConnected(Channel channel) throws Exception {
+ LOG.trace("Channel has become active! Channel is {}", channel);
connectionEstablished(channel);
}
+ protected void handleChannelInactive(Channel channel) throws Exception {
+ LOG.trace("Channel has gone inactive! Channel is {}", channel);
+ if (connected.compareAndSet(true, false) && !closed.get()) {
+ LOG.trace("Firing onTransportClosed listener");
+ listener.onTransportClosed();
+ }
+ }
+
+ protected void handleException(Channel channel, Throwable cause) throws Exception {
+ LOG.trace("Exception on channel! Channel is {}", channel);
+ if (connected.compareAndSet(true, false) && !closed.get()) {
+ LOG.trace("Firing onTransportError listener");
+ if (pendingFailure != null) {
+ listener.onTransportError(pendingFailure);
+ } else {
+ listener.onTransportError(cause);
+ }
+ } else {
+ // Hold the first failure for later dispatch if connect succeeds.
+ // This will then trigger disconnect using the first error reported.
+ if (pendingFailure == null) {
+ LOG.trace("Holding error until connect succeeds: {}", cause.getMessage());
+ pendingFailure = cause;
+ }
+
+ connectionFailed(channel, IOExceptionSupport.create(pendingFailure));
+ }
+ }
+
//----- State change handlers and checks ---------------------------------//
- /**
+ protected final void checkConnected() throws IOException {
+ if (!connected.get()) {
+ throw new IOException("Cannot send to a non-connected transport.");
+ }
+ }
+
+ /*
* Called when the transport has successfully connected and is ready for use.
*/
- protected void connectionEstablished(Channel connectedChannel) {
+ private void connectionEstablished(Channel connectedChannel) {
channel = connectedChannel;
connected.set(true);
connectLatch.countDown();
}
- /**
+ /*
* Called when the transport connection failed and an error should be returned.
- *
- * @param failedChannel
- * The Channel instance that failed.
- * @param cause
- * An IOException that describes the cause of the failed connection.
*/
- protected void connectionFailed(Channel failedChannel, IOException cause) {
+ private void connectionFailed(Channel failedChannel, IOException cause) {
failureCause = IOExceptionSupport.create(cause);
channel = failedChannel;
connected.set(false);
connectLatch.countDown();
}
- private void checkConnected() throws IOException {
- if (!connected.get()) {
- throw new IOException("Cannot send to a non-connected transport.");
+ private TransportSslOptions getSslOptions() {
+ return (TransportSslOptions) getTransportOptions();
+ }
+
+ private void configureNetty(Bootstrap bootstrap, TransportOptions options) {
+ bootstrap.option(ChannelOption.TCP_NODELAY, options.isTcpNoDelay());
+ bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, options.getConnectTimeout());
+ bootstrap.option(ChannelOption.SO_KEEPALIVE, options.isTcpKeepAlive());
+ bootstrap.option(ChannelOption.SO_LINGER, options.getSoLinger());
+ bootstrap.option(ChannelOption.ALLOCATOR, PartialPooledByteBufAllocator.INSTANCE);
+
+ if (options.getSendBufferSize() != -1) {
+ bootstrap.option(ChannelOption.SO_SNDBUF, options.getSendBufferSize());
+ }
+
+ if (options.getReceiveBufferSize() != -1) {
+ bootstrap.option(ChannelOption.SO_RCVBUF, options.getReceiveBufferSize());
+ bootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(options.getReceiveBufferSize()));
+ }
+
+ if (options.getTrafficClass() != -1) {
+ bootstrap.option(ChannelOption.IP_TOS, options.getTrafficClass());
}
}
- //----- Handle connection events -----------------------------------------//
+ private void configureChannel(final Channel channel) throws Exception {
+ // Default event handlers takes care of exceptions and channel inactivity, which
+ // prevents missed errors if the SslHandler fails during creation and removes the
+ // need for copying this implementation.
+ channel.pipeline().addLast(new NettyDefaultHandler());
+
+ if (isSecure()) {
- private class NettyTcpTransportHandler extends SimpleChannelInboundHandler<ByteBuf> {
+ SslHandler sslHandler = TransportSupport.createSslHandler(getRemoteLocation(), getSslOptions());
+
+ sslHandler.handshakeFuture().addListener(new GenericFutureListener<Future<Channel>>() {
+ @Override
+ public void operationComplete(Future<Channel> future) throws Exception {
+ if (future.isSuccess()) {
+ LOG.trace("SSL Handshake has completed: {}", channel);
+ handleConnected(channel);
+ } else {
+ LOG.trace("SSL Handshake has failed: {}", channel);
+ handleException(channel, future.cause());
+ }
+ }
+ });
+
+ channel.pipeline().addLast(sslHandler);
+ }
+
+ addAdditionalHandlers(channel.pipeline());
+
+ channel.pipeline().addLast(getChannelHandler());
+ }
+
+ //----- Handle connection errors -----------------------------------------//
+
+ private final class NettyDefaultHandler extends ChannelInboundHandlerAdapter {
@Override
- public void channelActive(ChannelHandlerContext context) throws Exception {
- LOG.trace("Channel has become active! Channel is {}", context.channel());
+ public void channelRegistered(ChannelHandlerContext context) throws Exception {
+ channel = context.channel();
}
@Override
public void channelInactive(ChannelHandlerContext context) throws Exception {
- LOG.trace("Channel has gone inactive! Channel is {}", context.channel());
- if (connected.compareAndSet(true, false) && !closed.get()) {
- LOG.trace("Firing onTransportClosed listener");
- listener.onTransportClosed();
- }
+ handleChannelInactive(context.channel());
}
@Override
public void exceptionCaught(ChannelHandlerContext context, Throwable cause) throws Exception {
- LOG.trace("Exception on channel! Channel is {}", context.channel());
- if (connected.compareAndSet(true, false) && !closed.get()) {
- LOG.trace("Firing onTransportError listener");
- if (pendingFailure != null) {
- listener.onTransportError(pendingFailure);
- } else {
- listener.onTransportError(cause);
- }
- } else {
- // Hold the first failure for later dispatch if connect succeeds.
- // This will then trigger disconnect using the first error reported.
- if (pendingFailure != null) {
- LOG.trace("Holding error until connect succeeds: {}", cause.getMessage());
- pendingFailure = cause;
- }
+ handleException(context.channel(), cause);
+ }
+ }
+
+ //----- Handle connection events -----------------------------------------//
+
+ protected class NettyTcpTransportHandler extends SimpleChannelInboundHandler<ByteBuf> {
+
+ @Override
+ public void channelActive(ChannelHandlerContext context) throws Exception {
+ // In the Secure case we need to let the handshake complete before we
+ // trigger the connected event.
+ if (!isSecure()) {
+ handleConnected(context.channel());
}
}
@@ -342,4 +426,3 @@ public class NettyTcpTransport implements Transport {
}
}
}
-
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransport.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransport.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransport.java
new file mode 100644
index 0000000..a69642d
--- /dev/null
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransport.java
@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.qpid.jms.transports.netty;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.qpid.jms.transports.TransportListener;
+import org.apache.qpid.jms.transports.TransportOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.DefaultHttpHeaders;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpClientCodec;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
+import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
+import io.netty.handler.codec.http.websocketx.WebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketVersion;
+
+/**
+ * Netty based WebSockets Transport that wraps and extends the TCP Transport.
+ */
+public class NettyWsTransport extends NettyTcpTransport {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NettyWsTransport.class);
+
+ /**
+ * Create a new transport instance
+ *
+ * @param remoteLocation
+ * the URI that defines the remote resource to connect to.
+ * @param options
+ * the transport options used to configure the socket connection.
+ */
+ public NettyWsTransport(URI remoteLocation, TransportOptions options) {
+ super(null, remoteLocation, options);
+ }
+
+ /**
+ * Create a new transport instance
+ *
+ * @param listener
+ * the TransportListener that will receive events from this Transport.
+ * @param remoteLocation
+ * the URI that defines the remote resource to connect to.
+ * @param options
+ * the transport options used to configure the socket connection.
+ */
+ public NettyWsTransport(TransportListener listener, URI remoteLocation, TransportOptions options) {
+ super(listener, remoteLocation, options);
+ }
+
+ @Override
+ public void send(ByteBuf output) throws IOException {
+ checkConnected();
+ int length = output.readableBytes();
+ if (length == 0) {
+ return;
+ }
+
+ LOG.trace("Attempted write of: {} bytes", length);
+
+ channel.writeAndFlush(new BinaryWebSocketFrame(output));
+ }
+
+ @Override
+ protected ChannelInboundHandlerAdapter getChannelHandler() {
+ return new NettyWebSocketTransportHandler();
+ }
+
+ @Override
+ protected void addAdditionalHandlers(ChannelPipeline pipeline) {
+ pipeline.addLast(new HttpClientCodec());
+ pipeline.addLast(new HttpObjectAggregator(8192));
+ }
+
+ @Override
+ protected void handleConnected(Channel channel) throws Exception {
+ LOG.trace("Channel has become active, awaiting WebSocket handshake! Channel is {}", channel);
+ }
+
+ //----- Handle connection events -----------------------------------------//
+
+ private class NettyWebSocketTransportHandler extends SimpleChannelInboundHandler<Object> {
+
+ private final WebSocketClientHandshaker handshaker;
+
+ public NettyWebSocketTransportHandler() {
+ handshaker = WebSocketClientHandshakerFactory.newHandshaker(
+ getRemoteLocation(), WebSocketVersion.V13, "amqp", true, new DefaultHttpHeaders());
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext context) throws Exception {
+ handshaker.handshake(context.channel());
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, Object message) throws Exception {
+ LOG.trace("New data read: incoming: {}", message);
+
+ Channel ch = ctx.channel();
+ if (!handshaker.isHandshakeComplete()) {
+ handshaker.finishHandshake(ch, (FullHttpResponse) message);
+ LOG.info("WebSocket Client connected! {}", ctx.channel());
+ // Now trigger super processing as we are really connected.
+ NettyWsTransport.super.handleConnected(ch);
+ return;
+ }
+
+ // We shouldn't get this since we handle the handshake previously.
+ if (message instanceof FullHttpResponse) {
+ FullHttpResponse response = (FullHttpResponse) message;
+ throw new IllegalStateException(
+ "Unexpected FullHttpResponse (getStatus=" + response.getStatus() +
+ ", content=" + response.content().toString(StandardCharsets.UTF_8) + ')');
+ }
+
+ WebSocketFrame frame = (WebSocketFrame) message;
+ if (frame instanceof TextWebSocketFrame) {
+ TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
+ LOG.warn("WebSocket Client received message: " + textFrame.text());
+ ctx.fireExceptionCaught(new IOException("Received invalid frame over WebSocket."));
+ } else if (frame instanceof BinaryWebSocketFrame) {
+ BinaryWebSocketFrame binaryFrame = (BinaryWebSocketFrame) frame;
+ LOG.info("WebSocket Client received data: {} bytes", binaryFrame.content().readableBytes());
+ listener.onData(binaryFrame.content());
+ } else if (frame instanceof CloseWebSocketFrame) {
+ LOG.trace("WebSocket Client received closing");
+ ch.close();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransportFactory.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransportFactory.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransportFactory.java
new file mode 100644
index 0000000..2f416bc
--- /dev/null
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransportFactory.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.qpid.jms.transports.netty;
+
+import java.net.URI;
+
+import org.apache.qpid.jms.transports.TransportFactory;
+import org.apache.qpid.jms.transports.TransportOptions;
+
+/**
+ * Factory for creating the Netty based WebSocket Transport.
+ */
+public class NettyWsTransportFactory extends TransportFactory {
+
+ @Override
+ protected NettyTcpTransport doCreateTransport(URI remoteURI, TransportOptions transportOptions) throws Exception {
+ return new NettyWsTransport(remoteURI, transportOptions);
+ }
+
+ @Override
+ public String getName() {
+ return "WS";
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWssTransportFactory.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWssTransportFactory.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWssTransportFactory.java
new file mode 100644
index 0000000..9e44870
--- /dev/null
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWssTransportFactory.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.qpid.jms.transports.netty;
+
+import org.apache.qpid.jms.transports.TransportOptions;
+import org.apache.qpid.jms.transports.TransportSslOptions;
+
+/**
+ * Creates a Netty based Secure WebSocket transport.
+ */
+public class NettyWssTransportFactory extends NettyWsTransportFactory {
+
+ @Override
+ protected TransportOptions doCreateTransportOptions() {
+ return TransportSslOptions.INSTANCE.clone();
+ }
+
+ @Override
+ public String getName() {
+ return "WSS";
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/provider/amqpws
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/provider/amqpws b/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/provider/amqpws
new file mode 100644
index 0000000..9cd4d1c
--- /dev/null
+++ b/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/provider/amqpws
@@ -0,0 +1,18 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ---------------------------------------------------------------------------
+class=org.apache.qpid.jms.provider.amqp.AmqpProviderFactory
+transportType=ws
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/provider/amqpwss
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/provider/amqpwss b/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/provider/amqpwss
new file mode 100644
index 0000000..4c2b8c9
--- /dev/null
+++ b/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/provider/amqpwss
@@ -0,0 +1,18 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ---------------------------------------------------------------------------
+class=org.apache.qpid.jms.provider.amqp.AmqpProviderFactory
+transportType=wss
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/transports/ws
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/transports/ws b/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/transports/ws
new file mode 100644
index 0000000..1a19640
--- /dev/null
+++ b/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/transports/ws
@@ -0,0 +1,17 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ---------------------------------------------------------------------------
+class=org.apache.qpid.jms.transports.netty.NettyWsTransportFactory
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/transports/wss
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/transports/wss b/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/transports/wss
new file mode 100644
index 0000000..aa71461
--- /dev/null
+++ b/qpid-jms-client/src/main/resources/META-INF/services/org/apache/qpid/jms/transports/wss
@@ -0,0 +1,17 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ---------------------------------------------------------------------------
+class=org.apache.qpid.jms.transports.netty.NettyWssTransportFactory
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyEchoServer.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyEchoServer.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyEchoServer.java
index 68870d0..933e872 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyEchoServer.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyEchoServer.java
@@ -16,21 +16,6 @@
*/
package org.apache.qpid.jms.transports.netty;
-import io.netty.bootstrap.ServerBootstrap;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.handler.logging.LogLevel;
-import io.netty.handler.logging.LoggingHandler;
-import io.netty.handler.ssl.SslHandler;
-import io.netty.util.concurrent.Future;
-import io.netty.util.concurrent.GenericFutureListener;
-
import java.io.IOException;
import java.net.ServerSocket;
import java.util.concurrent.TimeUnit;
@@ -46,6 +31,26 @@ import org.apache.qpid.jms.transports.TransportSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
+import io.netty.handler.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
+
/**
* Simple Netty Server used to echo all data.
*/
@@ -53,6 +58,7 @@ public class NettyEchoServer implements AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(NettyEchoServer.class);
static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
+ static final String WEBSOCKET_PATH = "/";
private EventLoopGroup bossGroup;
private EventLoopGroup workerGroup;
@@ -60,6 +66,7 @@ public class NettyEchoServer implements AutoCloseable {
private final TransportOptions options;
private int serverPort;
private final boolean needClientAuth;
+ private final boolean webSocketServer;
private volatile SslHandler sslHandler;
private final AtomicBoolean started = new AtomicBoolean();
@@ -69,8 +76,13 @@ public class NettyEchoServer implements AutoCloseable {
}
public NettyEchoServer(TransportOptions options, boolean needClientAuth) {
+ this(options, needClientAuth, false);
+ }
+
+ public NettyEchoServer(TransportOptions options, boolean needClientAuth, boolean webSocketServer) {
this.options = options;
this.needClientAuth = needClientAuth;
+ this.webSocketServer = webSocketServer;
}
public void start() throws Exception {
@@ -99,6 +111,13 @@ public class NettyEchoServer implements AutoCloseable {
sslHandler = new SslHandler(engine);
ch.pipeline().addLast(sslHandler);
}
+
+ if (webSocketServer) {
+ ch.pipeline().addLast(new HttpServerCodec());
+ ch.pipeline().addLast(new HttpObjectAggregator(65536));
+ ch.pipeline().addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, "amqp", true));
+ }
+
ch.pipeline().addLast(new EchoServerHandler());
}
});
@@ -149,17 +168,17 @@ public class NettyEchoServer implements AutoCloseable {
return serverPort;
}
- private class EchoServerHandler extends ChannelInboundHandlerAdapter {
+ private class EchoServerHandler extends SimpleChannelInboundHandler<Object> {
@Override
public void channelActive(final ChannelHandlerContext ctx) {
- LOG.info("New active channel: {}", ctx.channel());
+ LOG.info("Server -> New active channel: {}", ctx.channel());
SslHandler handler = ctx.pipeline().get(SslHandler.class);
if (handler != null) {
handler.handshakeFuture().addListener(new GenericFutureListener<Future<Channel>>() {
@Override
public void operationComplete(Future<Channel> future) throws Exception {
- LOG.info("SSL handshake completed. Succeeded: {}", future.isSuccess());
+ LOG.info("Server -> SSL handshake completed. Succeeded: {}", future.isSuccess());
if (!future.isSuccess()) {
sslHandler.close();
ctx.close();
@@ -176,8 +195,19 @@ public class NettyEchoServer implements AutoCloseable {
}
@Override
- public void channelRead(ChannelHandlerContext ctx, Object msg) {
- ctx.write(msg);
+ public void channelRead0(ChannelHandlerContext ctx, Object msg) {
+ LOG.trace("Channel read: {}", msg);
+ if (webSocketServer && msg instanceof BinaryWebSocketFrame) {
+ BinaryWebSocketFrame frame = (BinaryWebSocketFrame) msg;
+ ctx.write(frame.copy());
+ return;
+ } else if (msg instanceof ByteBuf) {
+ ctx.write(((ByteBuf) msg).copy());
+ return;
+ }
+
+ String message = "unsupported frame type: " + msg.getClass().getName();
+ throw new UnsupportedOperationException(message);
}
@Override
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportFactoryTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportFactoryTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportFactoryTest.java
index 06b666c..2e35e7b 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportFactoryTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportFactoryTest.java
@@ -70,8 +70,9 @@ public class NettySslTransportFactoryTest {
Transport transport = factory.createTransport(BASE_URI);
assertNotNull(transport);
- assertTrue(transport instanceof NettySslTransport);
+ assertTrue(transport instanceof NettyTcpTransport);
assertFalse(transport.isConnected());
+ assertTrue(transport.isSecure());
TransportOptions options = transport.getTransportOptions();
assertNotNull(options);
@@ -140,8 +141,9 @@ public class NettySslTransportFactoryTest {
Transport transport = factory.createTransport(configuredURI);
assertNotNull(transport);
- assertTrue(transport instanceof NettySslTransport);
+ assertTrue(transport instanceof NettyTcpTransport);
assertFalse(transport.isConnected());
+ assertTrue(transport.isSecure());
TransportOptions options = transport.getTransportOptions();
assertNotNull(options);
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportTest.java
index 23b3b2d..e3eab61 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportTest.java
@@ -40,7 +40,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Test basic functionality of the Netty based SSL Transport.
+ * Test basic functionality of the Netty based TCP Transport ruuing in secure mode (SSL).
*/
@RunWith(QpidJMSTestRunner.class)
public class NettySslTransportTest extends NettyTcpTransportTest {
@@ -65,16 +65,19 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
@Override
@Test(timeout = 60 * 1000)
- public void testCreateWithNullOptionsUsesDefaults() throws Exception {
+ public void testCreateWithNullOptionsThrowsIAE() throws Exception {
URI serverLocation = new URI("tcp://localhost:5762");
- Transport transport = createTransport(serverLocation, testListener, null);
- assertEquals(TransportSslOptions.INSTANCE, transport.getTransportOptions());
+ try {
+ createTransport(serverLocation, testListener, null);
+ fail("Should have thrown IllegalArgumentException");
+ } catch (IllegalArgumentException iae) {
+ }
}
@Test(timeout = 60 * 1000)
public void testConnectToServerWithoutTrustStoreFails() throws Exception {
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -101,7 +104,7 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
@Test(timeout = 60 * 1000)
@Repeat(repetitions = 1)
public void testConnectToServerUsingUntrustedKeyFails() throws Exception {
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -128,7 +131,7 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
@Test(timeout = 60 * 1000)
public void testConnectToServerClientTrustsAll() throws Exception {
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -143,6 +146,7 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
}
assertTrue(transport.isConnected());
+ assertTrue(transport.isSecure());
transport.close();
}
@@ -155,7 +159,7 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
public void testConnectWithNeedClientAuth() throws Exception {
TransportSslOptions serverOptions = createServerOptions();
- try (NettyEchoServer server = new NettyEchoServer(serverOptions, true)) {
+ try (NettyEchoServer server = createEchoServer(serverOptions, true)) {
server.start();
int port = server.getServerPort();
@@ -163,7 +167,7 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
TransportSslOptions clientOptions = createClientOptions();
- NettySslTransport transport = createTransport(serverLocation, testListener, clientOptions);
+ NettyTcpTransport transport = createTransport(serverLocation, testListener, clientOptions);
try {
transport.connect();
LOG.info("Connection established to test server: {}", serverLocation);
@@ -172,6 +176,7 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
}
assertTrue(transport.isConnected());
+ assertTrue(transport.isSecure());
// Verify there was a certificate sent to the server
assertNotNull(server.getSslHandler().engine().getSession().getPeerCertificates());
@@ -192,7 +197,7 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
private void doClientAuthAliasTestImpl(String alias, String expectedDN) throws Exception, URISyntaxException, IOException, InterruptedException {
TransportSslOptions serverOptions = createServerOptions();
- try (NettyEchoServer server = new NettyEchoServer(serverOptions, true)) {
+ try (NettyEchoServer server = createEchoServer(serverOptions, true)) {
server.start();
int port = server.getServerPort();
@@ -202,7 +207,7 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
clientOptions.setKeyStoreLocation(CLIENT_MULTI_KEYSTORE);
clientOptions.setKeyAlias(alias);
- NettySslTransport transport = createTransport(serverLocation, testListener, clientOptions);
+ NettyTcpTransport transport = createTransport(serverLocation, testListener, clientOptions);
try {
transport.connect();
LOG.info("Connection established to test server: {}", serverLocation);
@@ -211,6 +216,7 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
}
assertTrue(transport.isConnected());
+ assertTrue(transport.isSecure());
Certificate[] peerCertificates = server.getSslHandler().engine().getSession().getPeerCertificates();
assertNotNull(peerCertificates);
@@ -241,7 +247,7 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
TransportSslOptions serverOptions = createServerOptions();
serverOptions.setKeyStoreLocation(SERVER_WRONG_HOST_KEYSTORE);
- try (NettyEchoServer server = new NettyEchoServer(serverOptions)) {
+ try (NettyEchoServer server = createEchoServer(serverOptions)) {
server.start();
int port = server.getServerPort();
@@ -281,8 +287,8 @@ public class NettySslTransportTest extends NettyTcpTransportTest {
}
@Override
- protected NettySslTransport createTransport(URI serverLocation, TransportListener listener, TransportOptions options) {
- return new NettySslTransport(listener, serverLocation, options);
+ protected NettyTcpTransport createTransport(URI serverLocation, TransportListener listener, TransportOptions options) {
+ return new NettyTcpTransport(listener, serverLocation, options);
}
@Override
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpTransportTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpTransportTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpTransportTest.java
index f5de30d..58f3b6b 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpTransportTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpTransportTest.java
@@ -22,8 +22,6 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
import java.io.IOException;
import java.net.URI;
@@ -40,6 +38,9 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+
/**
* Test basic functionality of the Netty based TCP transport.
*/
@@ -71,16 +72,19 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
}
@Test(timeout = 60 * 1000)
- public void testCreateWithNullOptionsUsesDefaults() throws Exception {
+ public void testCreateWithNullOptionsThrowsIAE() throws Exception {
URI serverLocation = new URI("tcp://localhost:5762");
- Transport transport = createTransport(serverLocation, testListener, null);
- assertEquals(TransportOptions.INSTANCE, transport.getTransportOptions());
+ try {
+ createTransport(serverLocation, testListener, null);
+ fail("Should have thrown IllegalArgumentException");
+ } catch (IllegalArgumentException iae) {
+ }
}
@Test(timeout = 60 * 1000)
public void testConnectWithoutRunningServer() throws Exception {
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -108,7 +112,7 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
@Test(timeout = 60 * 1000)
public void testConnectWithoutListenerFails() throws Exception {
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -130,7 +134,7 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
@Test(timeout = 60 * 1000)
public void testConnectAfterListenerSetWorks() throws Exception {
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -156,7 +160,7 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
@Test(timeout = 60 * 1000)
public void testConnectToServer() throws Exception {
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -188,7 +192,7 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
public void testMultipleConnectionsToServer() throws Exception {
final int CONNECTION_COUNT = 10;
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -228,7 +232,7 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
sendBuffer.writeByte('A');
}
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -268,7 +272,7 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
public void testDetectServerClose() throws Exception {
Transport transport = null;
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -307,7 +311,7 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
@Test(timeout = 60 * 1000)
public void testZeroSizedSentNoErrors() throws Exception {
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -335,7 +339,7 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
@Test(timeout = 60 * 1000)
public void testDataSentIsReceived() throws Exception {
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -387,7 +391,7 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
public void doMultipleDataPacketsSentAndReceive(final int byteCount, final int iterations) throws Exception {
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -431,7 +435,7 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
public void testSendToClosedTransportFails() throws Exception {
Transport transport = null;
- try (NettyEchoServer server = new NettyEchoServer(createServerOptions())) {
+ try (NettyEchoServer server = createEchoServer(createServerOptions())) {
server.start();
int port = server.getServerPort();
@@ -482,6 +486,14 @@ public class NettyTcpTransportTest extends QpidJmsTestCase {
}
}
+ protected NettyEchoServer createEchoServer(TransportOptions options) {
+ return createEchoServer(options, false);
+ }
+
+ protected NettyEchoServer createEchoServer(TransportOptions options, boolean needClientAuth) {
+ return new NettyEchoServer(options, needClientAuth);
+ }
+
private class NettyTransportListener implements TransportListener {
@Override
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWsTransportTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWsTransportTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWsTransportTest.java
new file mode 100644
index 0000000..7348775
--- /dev/null
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWsTransportTest.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.qpid.jms.transports.netty;
+
+import java.net.URI;
+
+import org.apache.qpid.jms.transports.Transport;
+import org.apache.qpid.jms.transports.TransportListener;
+import org.apache.qpid.jms.transports.TransportOptions;
+
+/**
+ * Test the Netty based WebSocket Transport
+ */
+public class NettyWsTransportTest extends NettyTcpTransportTest {
+
+ @Override
+ protected NettyEchoServer createEchoServer(TransportOptions options, boolean needClientAuth) {
+ return new NettyEchoServer(options, needClientAuth, true);
+ }
+
+ @Override
+ protected Transport createTransport(URI serverLocation, TransportListener listener, TransportOptions options) {
+ if (listener == null) {
+ return new NettyWsTransport(serverLocation, options);
+ } else {
+ return new NettyWsTransport(listener, serverLocation, options);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWssTransportTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWssTransportTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWssTransportTest.java
new file mode 100644
index 0000000..69fdd57
--- /dev/null
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWssTransportTest.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.qpid.jms.transports.netty;
+
+import java.net.URI;
+
+import org.apache.qpid.jms.transports.TransportListener;
+import org.apache.qpid.jms.transports.TransportOptions;
+
+/**
+ * Test the NettyWsTransport with channel level security enabled.
+ */
+public class NettyWssTransportTest extends NettySslTransportTest {
+
+ @Override
+ protected NettyEchoServer createEchoServer(TransportOptions options, boolean needClientAuth) {
+ return new NettyEchoServer(options, needClientAuth, true);
+ }
+
+ @Override
+ protected NettyTcpTransport createTransport(URI serverLocation, TransportListener listener, TransportOptions options) {
+ if (listener == null) {
+ return new NettyWsTransport(serverLocation, options);
+ } else {
+ return new NettyWsTransport(listener, serverLocation, options);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-interop-tests/qpid-jms-activemq-tests/pom.xml
----------------------------------------------------------------------
diff --git a/qpid-jms-interop-tests/qpid-jms-activemq-tests/pom.xml b/qpid-jms-interop-tests/qpid-jms-activemq-tests/pom.xml
index d5931e6..4931c29 100644
--- a/qpid-jms-interop-tests/qpid-jms-activemq-tests/pom.xml
+++ b/qpid-jms-interop-tests/qpid-jms-activemq-tests/pom.xml
@@ -62,8 +62,6 @@
<artifactId>slf4j-log4j12</artifactId>
<scope>test</scope>
</dependency>
-
- <!-- Joram JMS conformance tests -->
<dependency>
<groupId>org.fusesource.joram-jms-tests</groupId>
<artifactId>joram-jms-tests</artifactId>
@@ -71,9 +69,24 @@
<scope>test</scope>
</dependency>
<dependency>
- <!-- using it for Jetty's JNDI context to work /w Joram tests. -->
<groupId>org.eclipse.jetty.aggregate</groupId>
- <artifactId>jetty-all-server</artifactId>
+ <artifactId>jetty-all</artifactId>
+ <version>${jetty-version}</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>javax.servlet</groupId>
+ <artifactId>javax.servlet-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>javax.websocket</groupId>
+ <artifactId>javax.websocket-api</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty.websocket</groupId>
+ <artifactId>websocket-server</artifactId>
<version>${jetty-version}</version>
<scope>test</scope>
</dependency>
@@ -103,7 +116,7 @@
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
- <artifactId>activemq-spring</artifactId>
+ <artifactId>activemq-http</artifactId>
<version>${activemq-version}</version>
<scope>test</scope>
</dependency>
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/91fd06d6/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/JmsWSConnectionTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/JmsWSConnectionTest.java b/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/JmsWSConnectionTest.java
new file mode 100644
index 0000000..eaa30dd
--- /dev/null
+++ b/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/JmsWSConnectionTest.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.qpid.jms;
+
+import static org.junit.Assert.assertNotNull;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.URI;
+
+import javax.net.ServerSocketFactory;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnector;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test connections can be established to remote peers via WebSockets
+ */
+@Ignore("Until a release with WS support for AMQP is ready")
+public class JmsWSConnectionTest {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(JmsWSConnectionTest.class);
+
+ private BrokerService brokerService;
+ private URI connectionURI;
+ private final int DEFAULT_WS_PORT = 5679;
+
+ @Before
+ public void setUp() throws Exception {
+ brokerService = new BrokerService();
+ brokerService.setPersistent(false);
+ brokerService.setAdvisorySupport(false);
+ brokerService.setDeleteAllMessagesOnStartup(true);
+ brokerService.setUseJmx(false);
+
+ TransportConnector connector = brokerService.addConnector("ws://0.0.0.0:" + getProxyPort());
+ connectionURI = connector.getPublishableConnectURI();
+ LOG.debug("Using amqp+ws connection: {}", connectionURI);
+
+ brokerService.start();
+ brokerService.waitUntilStarted();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ brokerService.stop();
+ brokerService.waitUntilStopped();
+ }
+
+ @Test(timeout=30000)
+ public void testCreateConnectionAndStart() throws Exception {
+ JmsConnectionFactory factory = new JmsConnectionFactory(getConnectionURI());
+ JmsConnection connection = (JmsConnection) factory.createConnection();
+ assertNotNull(connection);
+ connection.start();
+ connection.close();
+ }
+
+ protected String getConnectionURI() throws Exception {
+ return "amqpws://localhost:" + connectionURI.getPort();
+ }
+
+ protected int getProxyPort() {
+ int proxyPort = DEFAULT_WS_PORT;
+
+ try (ServerSocket ss = ServerSocketFactory.getDefault().createServerSocket(0)) {
+ proxyPort = ss.getLocalPort();
+ } catch (IOException e) {
+ }
+
+ return proxyPort;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org