Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-1159 Fixes and Improvements to the AMQP test client. 

Port fixes to the AMQP test client recently made in the 5.x version.
Fixes some thread safety issues in the Transport.  Ensures more 
timely shutdown of the Connection executor.  Uses a dynamic Proxy 
to generate Read-Only Proton wrappers instead of the hand crafted 
versions.  Adds additional logging for test data
This commit is contained in:
Timothy Bish 2017-05-11 16:46:22 -04:00
parent a98dccb35d
commit 4ad78c7fd0
21 changed files with 565 additions and 1835 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java
View File

@ -25,8 +25,8 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@ -39,7 +39,7 @@ import org.apache.activemq.transport.amqp.client.util.AsyncResult;
import org.apache.activemq.transport.amqp.client.util.ClientFuture;
import org.apache.activemq.transport.amqp.client.util.IdGenerator;
import org.apache.activemq.transport.amqp.client.util.NoOpAsyncResult;
import org.apache.activemq.transport.amqp.client.util.UnmodifiableConnection;
import org.apache.activemq.transport.amqp.client.util.UnmodifiableProxy;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.engine.Collector;
import org.apache.qpid.proton.engine.Connection;
@ -74,7 +74,7 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
public static final long DEFAULT_CLOSE_TIMEOUT = 30000;
public static final long DEFAULT_DRAIN_TIMEOUT = 60000;
private final ScheduledExecutorService serializer;
private ScheduledThreadPoolExecutor serializer;
private final AtomicBoolean closed = new AtomicBoolean();
private final AtomicBoolean connected = new AtomicBoolean();
private final AtomicLong sessionIdGenerator = new AtomicLong();
@ -121,7 +121,7 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
this.connectionId = CONNECTION_ID_GENERATOR.generateId();
this.remoteURI = transport.getRemoteLocation();
this.serializer = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
this.serializer = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable runner) {
@ -132,6 +132,10 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
}
});
// Ensure timely shutdown
this.serializer.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
this.serializer.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
this.transport.setTransportListener(this);
}
@ -434,7 +438,7 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
}
public Connection getConnection() {
return new UnmodifiableConnection(getEndpoint());
return UnmodifiableProxy.connectionProxy(getEndpoint());
}
public AmqpConnectionListener getListener() {

4
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/AmqpMessage.java
View File

@ -20,7 +20,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.NoSuchElementException;
import org.apache.activemq.transport.amqp.client.util.UnmodifiableDelivery;
import org.apache.activemq.transport.amqp.client.util.UnmodifiableProxy;
import org.apache.qpid.proton.Proton;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.DescribedType;
@ -100,7 +100,7 @@ public class AmqpMessage {
*/
public Delivery getWrappedDelivery() {
if (delivery != null) {
return new UnmodifiableDelivery(delivery);
return UnmodifiableProxy.deliveryProxy(delivery);
}
return null;

4
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/AmqpReceiver.java
View File

@ -35,7 +35,7 @@ import javax.jms.InvalidDestinationException;
import org.apache.activemq.transport.amqp.client.util.AsyncResult;
import org.apache.activemq.transport.amqp.client.util.ClientFuture;
import org.apache.activemq.transport.amqp.client.util.IOExceptionSupport;
import org.apache.activemq.transport.amqp.client.util.UnmodifiableReceiver;
import org.apache.activemq.transport.amqp.client.util.UnmodifiableProxy;
import org.apache.qpid.jms.JmsOperationTimedOutException;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.DescribedType;
@ -677,7 +677,7 @@ public class AmqpReceiver extends AmqpAbstractResource<Receiver> {
* @return an unmodifiable view of the underlying Receiver instance.
*/
public Receiver getReceiver() {
return new UnmodifiableReceiver(getEndpoint());
return UnmodifiableProxy.receiverProxy(getEndpoint());
}
// ----- Receiver configuration properties --------------------------------//

4
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/AmqpSender.java
View File

@ -29,7 +29,7 @@ import javax.jms.InvalidDestinationException;
import org.apache.activemq.transport.amqp.client.util.AsyncResult;
import org.apache.activemq.transport.amqp.client.util.ClientFuture;
import org.apache.activemq.transport.amqp.client.util.UnmodifiableSender;
import org.apache.activemq.transport.amqp.client.util.UnmodifiableProxy;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.Accepted;
@ -231,7 +231,7 @@ public class AmqpSender extends AmqpAbstractResource<Sender> {
* @return an unmodifiable view of the underlying Sender instance.
*/
public Sender getSender() {
return new UnmodifiableSender(getEndpoint());
return UnmodifiableProxy.senderProxy(getEndpoint());
}
/**

4
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/AmqpSession.java
View File

@ -24,7 +24,7 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.activemq.transport.amqp.client.util.AsyncResult;
import org.apache.activemq.transport.amqp.client.util.ClientFuture;
import org.apache.activemq.transport.amqp.client.util.UnmodifiableSession;
import org.apache.activemq.transport.amqp.client.util.UnmodifiableProxy;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.Source;
import org.apache.qpid.proton.amqp.messaging.Target;
@ -598,7 +598,7 @@ public class AmqpSession extends AmqpAbstractResource<Session> {
}
public Session getSession() {
return new UnmodifiableSession(getEndpoint());
return UnmodifiableProxy.sessionProxy(getEndpoint());
}
public boolean isInTransaction() {

322
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTcpTransport.java
View File

@ -1,4 +1,4 @@
/**
/*
* 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.
@ -6,7 +6,7 @@
* (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
* 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,
@ -23,25 +23,29 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.activemq.transport.amqp.client.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 io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.apache.activemq.transport.amqp.client.util.IOExceptionSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* TCP based transport that uses Netty as the underlying IO layer.
@ -50,28 +54,27 @@ public class NettyTcpTransport implements NettyTransport {
private static final Logger LOG = LoggerFactory.getLogger(NettyTcpTransport.class);
private static final int QUIET_PERIOD = 20;
private static final int SHUTDOWN_TIMEOUT = 100;
protected Bootstrap bootstrap;
protected EventLoopGroup group;
protected Channel channel;
protected NettyTransportListener listener;
protected NettyTransportOptions options;
protected final NettyTransportOptions options;
protected final URI remote;
protected boolean secure;
private final AtomicBoolean connected = new AtomicBoolean();
private final AtomicBoolean closed = new AtomicBoolean();
private final CountDownLatch connectLatch = new CountDownLatch(1);
private IOException failureCause;
private Throwable pendingFailure;
private volatile IOException failureCause;
/**
* 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.
* @param remoteLocation
* the URI that defines the remote resource to connect to.
* @param options
* the transport options used to configure the socket connection.
*/
public NettyTcpTransport(URI remoteLocation, NettyTransportOptions options) {
this(null, remoteLocation, options);
@ -80,15 +83,25 @@ public class NettyTcpTransport implements NettyTransport {
/**
* 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.
* @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 NettyTcpTransport(NettyTransportListener listener, URI remoteLocation, NettyTransportOptions 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;
this.secure = remoteLocation.getScheme().equalsIgnoreCase("ssl");
}
@Override
@ -98,16 +111,27 @@ public class NettyTcpTransport implements NettyTransport {
throw new IllegalStateException("A transport listener must be set before connection attempts.");
}
final SslHandler sslHandler;
if (isSSL()) {
try {
sslHandler = NettyTransportSupport.createSslHandler(getRemoteLocation(), getSslOptions());
} catch (Exception ex) {
// TODO: can we stop it throwing Exception?
throw IOExceptionSupport.create(ex);
}
} else {
sslHandler = null;
}
group = new NioEventLoopGroup(1);
bootstrap = new Bootstrap();
bootstrap.group(group);
bootstrap.channel(NioSocketChannel.class);
bootstrap.handler(new ChannelInitializer<Channel>() {
@Override
public void initChannel(Channel connectedChannel) throws Exception {
configureChannel(connectedChannel);
configureChannel(connectedChannel, sslHandler);
}
});
@ -118,12 +142,8 @@ public class NettyTcpTransport implements NettyTransport {
@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()));
if (!future.isSuccess()) {
handleException(future.channel(), IOExceptionSupport.create(future.cause()));
}
}
});
@ -143,7 +163,10 @@ public class NettyTcpTransport implements NettyTransport {
channel = null;
}
if (group != null) {
group.shutdownGracefully(QUIET_PERIOD, SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS);
Future<?> fut = group.shutdownGracefully(0, SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS);
if (!fut.awaitUninterruptibly(2 * SHUTDOWN_TIMEOUT)) {
LOG.trace("Channel group shutdown failed to complete in allotted time");
}
group = null;
}
@ -154,8 +177,8 @@ public class NettyTcpTransport implements NettyTransport {
@Override
public void run() {
if (pendingFailure != null) {
channel.pipeline().fireExceptionCaught(pendingFailure);
if (failureCause != null) {
channel.pipeline().fireExceptionCaught(failureCause);
}
}
});
@ -169,18 +192,24 @@ public class NettyTcpTransport implements NettyTransport {
@Override
public boolean isSSL() {
return secure;
return options.isSSL();
}
@Override
public void close() throws IOException {
if (closed.compareAndSet(false, true)) {
connected.set(false);
if (channel != null) {
channel.close().syncUninterruptibly();
}
if (group != null) {
group.shutdownGracefully(QUIET_PERIOD, SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS);
try {
if (channel != null) {
channel.close().syncUninterruptibly();
}
} finally {
if (group != null) {
Future<?> fut = group.shutdownGracefully(0, SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS);
if (!fut.awaitUninterruptibly(2 * SHUTDOWN_TIMEOUT)) {
LOG.trace("Channel group shutdown failed to complete in allotted time");
}
}
}
}
}
@ -216,14 +245,6 @@ public class NettyTcpTransport implements NettyTransport {
@Override
public NettyTransportOptions getTransportOptions() {
if (options == null) {
if (isSSL()) {
options = NettyTransportSslOptions.INSTANCE;
} else {
options = NettyTransportOptions.INSTANCE;
}
}
return options;
}
@ -234,36 +255,106 @@ public class NettyTcpTransport implements NettyTransport {
@Override
public Principal getLocalPrincipal() {
if (!isSSL()) {
throw new UnsupportedOperationException("Not connected to a secure channel");
Principal result = null;
if (isSSL()) {
SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
result = sslHandler.engine().getSession().getLocalPrincipal();
}
SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
return sslHandler.engine().getSession().getLocalPrincipal();
return result;
}
//----- Internal implementation details, can be overridden as needed --//
// ----- Internal implementation details, can be overridden as needed -----//
protected String getRemoteHost() {
return remote.getHost();
}
protected int getRemotePort() {
int port = remote.getPort();
if (port <= 0) {
if (isSSL()) {
port = getSslOptions().getDefaultSslPort();
} else {
port = getTransportOptions().getDefaultTcpPort();
}
if (remote.getPort() != -1) {
return remote.getPort();
} else {
return isSSL() ? getSslOptions().getDefaultSslPort() : getTransportOptions().getDefaultTcpPort();
}
return port;
}
protected void configureNetty(Bootstrap bootstrap, NettyTransportOptions options) {
protected void addAdditionalHandlers(ChannelPipeline pipeline) {
}
protected ChannelInboundHandlerAdapter createChannelHandler() {
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 (failureCause != null) {
listener.onTransportError(failureCause);
} 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 (failureCause == null) {
LOG.trace("Holding error until connect succeeds: {}", cause.getMessage());
failureCause = IOExceptionSupport.create(cause);
}
connectionFailed(channel, failureCause);
}
}
// ----- 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.
*/
private void connectionEstablished(Channel connectedChannel) {
channel = connectedChannel;
connected.set(true);
connectLatch.countDown();
}
/*
* Called when the transport connection failed and an error should be returned.
*/
private void connectionFailed(Channel failedChannel, IOException cause) {
failureCause = cause;
channel = failedChannel;
connected.set(false);
connectLatch.countDown();
}
private NettyTransportSslOptions getSslOptions() {
return (NettyTransportSslOptions) getTransportOptions();
}
private void configureNetty(Bootstrap bootstrap, NettyTransportOptions options) {
bootstrap.option(ChannelOption.TCP_NODELAY, options.isTcpNoDelay());
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, options.getConnectTimeout());
bootstrap.option(ChannelOption.SO_KEEPALIVE, options.isTcpKeepAlive());
@ -283,105 +374,66 @@ public class NettyTcpTransport implements NettyTransport {
}
}
protected void configureChannel(final Channel channel) throws Exception {
private void configureChannel(final Channel channel, final SslHandler sslHandler) throws Exception {
if (isSSL()) {
SslHandler sslHandler = NettyTransportSupport.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);
}
channel.pipeline().addLast(new NettyTcpTransportHandler());
}
protected void handleConnected(final Channel channel) throws Exception {
if (!isSSL()) {
connectionEstablished(channel);
if (getTransportOptions().isTraceBytes()) {
channel.pipeline().addLast("logger", new LoggingHandler(getClass()));
}
addAdditionalHandlers(channel.pipeline());
channel.pipeline().addLast(createChannelHandler());
}
//----- State change handlers and checks ---------------------------------//
// ----- Handle connection events -----------------------------------------//
/**
* Called when the transport has successfully connected and is ready for use.
*/
protected void connectionEstablished(Channel connectedChannel) {
channel = connectedChannel;
connected.set(true);
connectLatch.countDown();
}
protected abstract class NettyDefaultHandler<E> extends SimpleChannelInboundHandler<E> {
/**
* 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) {
failureCause = IOExceptionSupport.create(cause);
channel = failedChannel;
connected.set(false);
connectLatch.countDown();
}
private NettyTransportSslOptions getSslOptions() {
return (NettyTransportSslOptions) getTransportOptions();
}
private void checkConnected() throws IOException {
if (!connected.get()) {
throw new IOException("Cannot send to a non-connected transport.");
@Override
public void channelRegistered(ChannelHandlerContext context) throws Exception {
channel = context.channel();
}
}
//----- Handle connection events -----------------------------------------//
private class NettyTcpTransportHandler extends SimpleChannelInboundHandler<ByteBuf> {
@Override
public void channelActive(ChannelHandlerContext context) throws Exception {
LOG.trace("Channel has become active! Channel is {}", context.channel());
// In the Secure case we need to let the handshake complete before we
// trigger the connected event.
if (!isSSL()) {
handleConnected(context.channel());
} else {
SslHandler sslHandler = context.pipeline().get(SslHandler.class);
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());
}
}
});
}
}
@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 Binary data from connection -------------------------------//
protected class NettyTcpTransportHandler extends NettyDefaultHandler<ByteBuf> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {

2
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTransport.java
View File

@ -23,7 +23,7 @@ import java.security.Principal;
import io.netty.buffer.ByteBuf;
/**
*
* Base for all Netty based Transports in this client.
*/
public interface NettyTransport {

16
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTransportFactory.java
View File

@ -1,4 +1,4 @@
/**
/*
* 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.
@ -6,7 +6,7 @@
* (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
* 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,
@ -30,12 +30,16 @@ public final class NettyTransportFactory {
}
/**
* Creates an instance of the given Transport and configures it using the
* properties set on the given remote broker URI.
* Creates an instance of the given Transport and configures it using the properties set on
* the given remote broker URI.
*
* @param remoteURI
* The URI used to connect to a remote Peer.
*
* @param remoteURI The URI used to connect to a remote Peer.
* @return a new Transport instance.
* @throws Exception if an error occurs while creating the Transport instance.
*
* @throws Exception
* if an error occurs while creating the Transport instance.
*/
public static NettyTransport createTransport(URI remoteURI) throws Exception {
Map<String, String> map = PropertyUtil.parseQuery(remoteURI.getQuery());

14
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTransportListener.java
View File

@ -1,4 +1,4 @@
/**
/*
* 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.
@ -6,7 +6,7 @@
* (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
* 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,
@ -19,15 +19,16 @@ package org.apache.activemq.transport.amqp.client.transport;
import io.netty.buffer.ByteBuf;
/**
* Listener interface that should be implemented by users of the various
* QpidJMS Transport classes.
* Listener interface that should be implemented by users of the various QpidJMS Transport
* classes.
*/
public interface NettyTransportListener {
/**
* Called when new incoming data has become available.
*
* @param incoming the next incoming packet of data.
* @param incoming
* the next incoming packet of data.
*/
void onData(ByteBuf incoming);
@ -39,7 +40,8 @@ public interface NettyTransportListener {
/**
* Called when an error occurs during normal Transport operations.
*
* @param cause the error that triggered this event.
* @param cause
* the error that triggered this event.
*/
void onTransportError(Throwable cause);

59
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTransportOptions.java
View File

@ -1,4 +1,4 @@
/**
/*
* 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.
@ -6,7 +6,7 @@
* (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
* 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,
@ -30,6 +30,7 @@ public class NettyTransportOptions implements Cloneable {
public static final int DEFAULT_SO_TIMEOUT = -1;
public static final int DEFAULT_CONNECT_TIMEOUT = 60000;
public static final int DEFAULT_TCP_PORT = 5672;
public static final boolean DEFAULT_TRACE_BYTES = false;
public static final NettyTransportOptions INSTANCE = new NettyTransportOptions();
@ -42,6 +43,7 @@ public class NettyTransportOptions implements Cloneable {
private boolean tcpKeepAlive = DEFAULT_TCP_KEEP_ALIVE;
private boolean tcpNoDelay = DEFAULT_TCP_NO_DELAY;
private int defaultTcpPort = DEFAULT_TCP_PORT;
private boolean traceBytes = DEFAULT_TRACE_BYTES;
/**
* @return the currently set send buffer size in bytes.
@ -51,11 +53,14 @@ public class NettyTransportOptions implements Cloneable {
}
/**
* Sets the send buffer size in bytes, the value must be greater than zero
* or an {@link IllegalArgumentException} will be thrown.
* Sets the send buffer size in bytes, the value must be greater than zero or an
* {@link IllegalArgumentException} will be thrown.
*
* @param sendBufferSize the new send buffer size for the TCP Transport.
* @throws IllegalArgumentException if the value given is not in the valid range.
* @param sendBufferSize
* the new send buffer size for the TCP Transport.
*
* @throws IllegalArgumentException
* if the value given is not in the valid range.
*/
public void setSendBufferSize(int sendBufferSize) {
if (sendBufferSize <= 0) {
@ -73,11 +78,14 @@ public class NettyTransportOptions implements Cloneable {
}
/**
* Sets the receive buffer size in bytes, the value must be greater than zero
* or an {@link IllegalArgumentException} will be thrown.
* Sets the receive buffer size in bytes, the value must be greater than zero or an
* {@link IllegalArgumentException} will be thrown.
*
* @param receiveBufferSize the new receive buffer size for the TCP Transport.
* @throws IllegalArgumentException if the value given is not in the valid range.
* @param receiveBufferSize
* the new receive buffer size for the TCP Transport.
*
* @throws IllegalArgumentException
* if the value given is not in the valid range.
*/
public void setReceiveBufferSize(int receiveBufferSize) {
if (receiveBufferSize <= 0) {
@ -95,11 +103,13 @@ public class NettyTransportOptions implements Cloneable {
}
/**
* Sets the traffic class value used by the TCP connection, valid
* range is between 0 and 255.
* Sets the traffic class value used by the TCP connection, valid range is between 0 and 255.
*
* @param trafficClass the new traffic class value.
* @throws IllegalArgumentException if the value given is not in the valid range.
* @param trafficClass
* the new traffic class value.
*
* @throws IllegalArgumentException
* if the value given is not in the valid range.
*/
public void setTrafficClass(int trafficClass) {
if (trafficClass < 0 || trafficClass > 255) {
@ -157,6 +167,27 @@ public class NettyTransportOptions implements Cloneable {
this.defaultTcpPort = defaultTcpPort;
}
/**
* @return true if the transport should enable byte tracing
*/
public boolean isTraceBytes() {
return traceBytes;
}
/**
* Determines if the transport should add a logger for bytes in / out
*
* @param traceBytes
* should the transport log the bytes in and out.
*/
public void setTraceBytes(boolean traceBytes) {
this.traceBytes = traceBytes;
}
public boolean isSSL() {
return false;
}
@Override
public NettyTransportOptions clone() {
return copyOptions(new NettyTransportOptions());

56
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTransportSslOptions.java
View File

@ -1,4 +1,4 @@
/**
/*
* 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.
@ -6,7 +6,7 @@
* (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
* 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,
@ -21,9 +21,8 @@ import java.util.Collections;
import java.util.List;
/**
* Holds the defined SSL options for connections that operate over a secure
* transport. Options are read from the environment and can be overridden by
* specifying them on the connection URI.
* Holds the defined SSL options for connections that operate over a secure transport. Options
* are read from the environment and can be overridden by specifying them on the connection URI.
*/
public class NettyTransportSslOptions extends NettyTransportOptions {
@ -31,7 +30,7 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
public static final String DEFAULT_CONTEXT_PROTOCOL = "TLS";
public static final boolean DEFAULT_TRUST_ALL = false;
public static final boolean DEFAULT_VERIFY_HOST = false;
public static final List<String> DEFAULT_DISABLED_PROTOCOLS = Collections.unmodifiableList(Arrays.asList(new String[]{"SSLv2Hello", "SSLv3"}));
public static final List<String> DEFAULT_DISABLED_PROTOCOLS = Collections.unmodifiableList(Arrays.asList(new String[] {"SSLv2Hello", "SSLv3"}));
public static final int DEFAULT_SSL_PORT = 5671;
public static final NettyTransportSslOptions INSTANCE = new NettyTransportSslOptions();
@ -69,7 +68,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
/**
* Sets the location on disk of the key store to use.
*
* @param keyStoreLocation the keyStoreLocation to use to create the key manager.
* @param keyStoreLocation
* the keyStoreLocation to use to create the key manager.
*/
public void setKeyStoreLocation(String keyStoreLocation) {
this.keyStoreLocation = keyStoreLocation;
@ -83,7 +83,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
}
/**
* @param keyStorePassword the keyStorePassword to set
* @param keyStorePassword
* the keyStorePassword to set
*/
public void setKeyStorePassword(String keyStorePassword) {
this.keyStorePassword = keyStorePassword;
@ -97,7 +98,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
}
/**
* @param trustStoreLocation the trustStoreLocation to set
* @param trustStoreLocation
* the trustStoreLocation to set
*/
public void setTrustStoreLocation(String trustStoreLocation) {
this.trustStoreLocation = trustStoreLocation;
@ -111,7 +113,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
}
/**
* @param trustStorePassword the trustStorePassword to set
* @param trustStorePassword
* the trustStorePassword to set
*/
public void setTrustStorePassword(String trustStorePassword) {
this.trustStorePassword = trustStorePassword;
@ -125,7 +128,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
}
/**
* @param storeType the format that the store files are encoded in.
* @param storeType
* the format that the store files are encoded in.
*/
public void setStoreType(String storeType) {
this.storeType = storeType;
@ -139,7 +143,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
}
/**
* @param enabledCipherSuites the enabledCipherSuites to set
* @param enabledCipherSuites
* the enabledCipherSuites to set
*/
public void setEnabledCipherSuites(String[] enabledCipherSuites) {
this.enabledCipherSuites = enabledCipherSuites;
@ -153,7 +158,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
}
/**
* @param disabledCipherSuites the disabledCipherSuites to set
* @param disabledCipherSuites
* the disabledCipherSuites to set
*/
public void setDisabledCipherSuites(String[] disabledCipherSuites) {
this.disabledCipherSuites = disabledCipherSuites;
@ -169,13 +175,15 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
/**
* The protocols to be set as enabled.
*
* @param enabledProtocols the enabled protocols to set, or null if the defaults should be used.
* @param enabledProtocols
* the enabled protocols to set, or null if the defaults should be used.
*/
public void setEnabledProtocols(String[] enabledProtocols) {
this.enabledProtocols = enabledProtocols;
}
/**
*
* @return the protocols to disable or null if none should be
*/
public String[] getDisabledProtocols() {
@ -185,7 +193,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
/**
* The protocols to be disable.
*
* @param disabledProtocols the protocols to disable, or null if none should be.
* @param disabledProtocols
* the protocols to disable, or null if none should be.
*/
public void setDisabledProtocols(String[] disabledProtocols) {
this.disabledProtocols = disabledProtocols;
@ -202,7 +211,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
* The protocol value to use when creating an SSLContext via
* SSLContext.getInstance(protocol).
*
* @param contextProtocol the context protocol to use.
* @param contextProtocol
* the context protocol to use.
*/
public void setContextProtocol(String contextProtocol) {
this.contextProtocol = contextProtocol;
@ -216,7 +226,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
}
/**
* @param trustAll the trustAll to set
* @param trustAll
* the trustAll to set
*/
public void setTrustAll(boolean trustAll) {
this.trustAll = trustAll;
@ -230,7 +241,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
}
/**
* @param verifyHost the verifyHost to set
* @param verifyHost
* the verifyHost to set
*/
public void setVerifyHost(boolean verifyHost) {
this.verifyHost = verifyHost;
@ -244,7 +256,8 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
}
/**
* @param keyAlias the key alias to use
* @param keyAlias
* the key alias to use
*/
public void setKeyAlias(String keyAlias) {
this.keyAlias = keyAlias;
@ -258,6 +271,11 @@ public class NettyTransportSslOptions extends NettyTransportOptions {
this.defaultSslPort = defaultSslPort;
}
@Override
public boolean isSSL() {
return true;
}
@Override
public NettyTransportSslOptions clone() {
return copyOptions(new NettyTransportSslOptions());

81
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTransportSupport.java
View File

@ -1,4 +1,4 @@
/**
/*
* 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.
@ -6,7 +6,7 @@
* (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
* 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,
@ -16,15 +16,6 @@
*/
package org.apache.activemq.transport.amqp.client.transport;
import javax.net.ssl.KeyManager;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLParameters;
import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactory;
import javax.net.ssl.X509ExtendedKeyManager;
import javax.net.ssl.X509TrustManager;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
@ -38,10 +29,21 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import io.netty.handler.ssl.SslHandler;
import javax.net.ssl.KeyManager;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLParameters;
import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactory;
import javax.net.ssl.X509ExtendedKeyManager;
import javax.net.ssl.X509TrustManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.handler.ssl.SslHandler;
/**
* Static class that provides various utility methods used by Transport implementations.
*/
@ -50,13 +52,18 @@ public class NettyTransportSupport {
private static final Logger LOG = LoggerFactory.getLogger(NettyTransportSupport.class);
/**
* Creates a Netty SslHandler instance for use in Transports that require
* an SSL encoder / decoder.
* Creates a Netty SslHandler instance for use in Transports that require an SSL encoder /
* decoder.
*
* @param remote
* The URI of the remote peer that the SslHandler will be used against.
* @param options
* The SSL options object to build the SslHandler instance from.
*
* @param remote The URI of the remote peer that the SslHandler will be used against.
* @param options The SSL options object to build the SslHandler instance from.
* @return a new SslHandler that is configured from the given options.
* @throws Exception if an error occurs while creating the SslHandler instance.
*
* @throws Exception
* if an error occurs while creating the SslHandler instance.
*/
public static SslHandler createSslHandler(URI remote, NettyTransportSslOptions options) throws Exception {
return new SslHandler(createSslEngine(remote, createSslContext(options), options));
@ -66,9 +73,13 @@ public class NettyTransportSupport {
* Create a new SSLContext using the options specific in the given TransportSslOptions
* instance.
*
* @param options the configured options used to create the SSLContext.
* @param options
* the configured options used to create the SSLContext.
*
* @return a new SSLContext instance.
* @throws Exception if an error occurs while creating the context.
*
* @throws Exception
* if an error occurs while creating the context.
*/
public static SSLContext createSslContext(NettyTransportSslOptions options) throws Exception {
try {
@ -91,10 +102,15 @@ public class NettyTransportSupport {
* Create a new SSLEngine instance in client mode from the given SSLContext and
* TransportSslOptions instances.
*
* @param context the SSLContext to use when creating the engine.
* @param options the TransportSslOptions to use to configure the new SSLEngine.
* @param context
* the SSLContext to use when creating the engine.
* @param options
* the TransportSslOptions to use to configure the new SSLEngine.
*
* @return a new SSLEngine instance in client mode.
* @throws Exception if an error occurs while creating the new SSLEngine.
*
* @throws Exception
* if an error occurs while creating the new SSLEngine.
*/
public static SSLEngine createSslEngine(SSLContext context, NettyTransportSslOptions options) throws Exception {
return createSslEngine(null, context, options);
@ -104,15 +120,20 @@ public class NettyTransportSupport {
* Create a new SSLEngine instance in client mode from the given SSLContext and
* TransportSslOptions instances.
*
* @param remote the URI of the remote peer that will be used to initialize the engine, may be null if none should.
* @param context the SSLContext to use when creating the engine.
* @param options the TransportSslOptions to use to configure the new SSLEngine.
* @param remote
* the URI of the remote peer that will be used to initialize the engine, may be null
* if none should.
* @param context
* the SSLContext to use when creating the engine.
* @param options
* the TransportSslOptions to use to configure the new SSLEngine.
*
* @return a new SSLEngine instance in client mode.
* @throws Exception if an error occurs while creating the new SSLEngine.
*
* @throws Exception
* if an error occurs while creating the new SSLEngine.
*/
public static SSLEngine createSslEngine(URI remote,
SSLContext context,
NettyTransportSslOptions options) throws Exception {
public static SSLEngine createSslEngine(URI remote, SSLContext context, NettyTransportSslOptions options) throws Exception {
SSLEngine engine = null;
if (remote == null) {
engine = context.createSSLEngine();
@ -185,7 +206,7 @@ public class NettyTransportSupport {
private static TrustManager[] loadTrustManagers(NettyTransportSslOptions options) throws Exception {
if (options.isTrustAll()) {
return new TrustManager[]{createTrustAllTrustManager()};
return new TrustManager[] {createTrustAllTrustManager()};
}
if (options.getTrustStoreLocation() == null) {

378
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyWSTransport.java
View File

@ -18,73 +18,46 @@ package org.apache.activemq.transport.amqp.client.transport;
import java.io.IOException;
import java.net.URI;
import java.security.Principal;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.nio.charset.StandardCharsets;
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.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPromise;
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 io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPipeline;
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.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
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;
import io.netty.handler.ssl.SslHandler;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.apache.activemq.transport.amqp.client.util.IOExceptionSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Transport for communicating over WebSockets
*/
public class NettyWSTransport implements NettyTransport {
public class NettyWSTransport extends NettyTcpTransport {
private static final Logger LOG = LoggerFactory.getLogger(NettyWSTransport.class);
private static final int QUIET_PERIOD = 20;
private static final int SHUTDOWN_TIMEOUT = 100;
protected Bootstrap bootstrap;
protected EventLoopGroup group;
protected Channel channel;
protected NettyTransportListener listener;
protected NettyTransportOptions options;
protected final URI remote;
protected boolean secure;
private final AtomicBoolean connected = new AtomicBoolean();
private final AtomicBoolean closed = new AtomicBoolean();
private ChannelPromise handshakeFuture;
private IOException failureCause;
private Throwable pendingFailure;
private static final String AMQP_SUB_PROTOCOL = "amqp";
/**
* 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.
* @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, NettyTransportOptions options) {
this(null, remoteLocation, options);
@ -93,119 +66,15 @@ public class NettyWSTransport implements NettyTransport {
/**
* 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.
* @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(NettyTransportListener listener, URI remoteLocation, NettyTransportOptions options) {
this.options = options;
this.listener = listener;
this.remote = remoteLocation;
this.secure = remoteLocation.getScheme().equalsIgnoreCase("wss");
}
@Override
public void connect() throws IOException {
if (listener == null) {
throw new IllegalStateException("A transport listener must be set before connection attempts.");
}
group = new NioEventLoopGroup(1);
bootstrap = new Bootstrap();
bootstrap.group(group);
bootstrap.channel(NioSocketChannel.class);
bootstrap.handler(new ChannelInitializer<Channel>() {
@Override
public void initChannel(Channel connectedChannel) throws Exception {
configureChannel(connectedChannel);
}
});
configureNetty(bootstrap, getTransportOptions());
ChannelFuture future;
try {
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()));
}
}
});
future.sync();
// Now wait for WS protocol level handshake completion
handshakeFuture.await();
} catch (InterruptedException ex) {
LOG.debug("Transport connection attempt was interrupted.");
Thread.interrupted();
failureCause = IOExceptionSupport.create(ex);
}
if (failureCause != null) {
// Close out any Netty resources now as they are no longer needed.
if (channel != null) {
channel.close().syncUninterruptibly();
channel = null;
}
if (group != null) {
group.shutdownGracefully(QUIET_PERIOD, SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS);
group = null;
}
throw failureCause;
} else {
// Connected, allow any held async error to fire now and close the transport.
channel.eventLoop().execute(new Runnable() {
@Override
public void run() {
if (pendingFailure != null) {
channel.pipeline().fireExceptionCaught(pendingFailure);
}
}
});
}
}
@Override
public boolean isConnected() {
return connected.get();
}
@Override
public boolean isSSL() {
return secure;
}
@Override
public void close() throws IOException {
if (closed.compareAndSet(false, true)) {
connected.set(false);
if (channel != null) {
channel.close().syncUninterruptibly();
}
if (group != null) {
group.shutdownGracefully(QUIET_PERIOD, SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS);
}
}
}
@Override
public ByteBuf allocateSendBuffer(int size) throws IOException {
checkConnected();
return channel.alloc().ioBuffer(size, size);
super(listener, remoteLocation, options);
}
@Override
@ -222,202 +91,37 @@ public class NettyWSTransport implements NettyTransport {
}
@Override
public NettyTransportListener getTransportListener() {
return listener;
protected ChannelInboundHandlerAdapter createChannelHandler() {
return new NettyWebSocketTransportHandler();
}
@Override
public void setTransportListener(NettyTransportListener listener) {
this.listener = listener;
protected void addAdditionalHandlers(ChannelPipeline pipeline) {
pipeline.addLast(new HttpClientCodec());
pipeline.addLast(new HttpObjectAggregator(8192));
}
@Override
public NettyTransportOptions getTransportOptions() {
if (options == null) {
if (isSSL()) {
options = NettyTransportSslOptions.INSTANCE;
} else {
options = NettyTransportOptions.INSTANCE;
}
}
return options;
protected void handleConnected(Channel channel) throws Exception {
LOG.trace("Channel has become active, awaiting WebSocket handshake! Channel is {}", channel);
}
@Override
public URI getRemoteLocation() {
return remote;
}
// ----- Handle connection events -----------------------------------------//
@Override
public Principal getLocalPrincipal() {
if (!isSSL()) {
throw new UnsupportedOperationException("Not connected to a secure channel");
}
SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
return sslHandler.engine().getSession().getLocalPrincipal();
}
//----- Internal implementation details, can be overridden as needed --//
protected String getRemoteHost() {
return remote.getHost();
}
protected int getRemotePort() {
int port = remote.getPort();
if (port <= 0) {
if (isSSL()) {
port = getSslOptions().getDefaultSslPort();
} else {
port = getTransportOptions().getDefaultTcpPort();
}
}
return port;
}
protected void configureNetty(Bootstrap bootstrap, NettyTransportOptions 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());
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());
}
}
protected void configureChannel(final Channel channel) throws Exception {
if (isSSL()) {
SslHandler sslHandler = NettyTransportSupport.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);
}
channel.pipeline().addLast(new HttpClientCodec());
channel.pipeline().addLast(new HttpObjectAggregator(8192));
channel.pipeline().addLast(new NettyTcpTransportHandler());
}
protected void handleConnected(final Channel channel) throws Exception {
if (!isSSL()) {
connectionEstablished(channel);
}
}
//----- State change handlers and checks ---------------------------------//
/**
* Called when the transport has successfully connected and is ready for use.
*/
protected void connectionEstablished(Channel connectedChannel) {
LOG.info("WebSocket connectionEstablished! {}", connectedChannel);
channel = connectedChannel;
connected.set(true);
}
/**
* 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) {
failureCause = IOExceptionSupport.create(cause);
channel = failedChannel;
connected.set(false);
handshakeFuture.setFailure(cause);
}
private NettyTransportSslOptions getSslOptions() {
return (NettyTransportSslOptions) getTransportOptions();
}
private void checkConnected() throws IOException {
if (!connected.get()) {
throw new IOException("Cannot send to a non-connected transport.");
}
}
//----- Handle connection events -----------------------------------------//
private class NettyTcpTransportHandler extends SimpleChannelInboundHandler<Object> {
private class NettyWebSocketTransportHandler extends NettyDefaultHandler<Object> {
private final WebSocketClientHandshaker handshaker;
NettyTcpTransportHandler() {
handshaker = WebSocketClientHandshakerFactory.newHandshaker(remote, WebSocketVersion.V13, "amqp", false, new DefaultHttpHeaders());
}
@Override
public void handlerAdded(ChannelHandlerContext context) {
LOG.trace("Handler has become added! Channel is {}", context.channel());
handshakeFuture = context.newPromise();
NettyWebSocketTransportHandler() {
handshaker = WebSocketClientHandshakerFactory.newHandshaker(getRemoteLocation(), WebSocketVersion.V13, AMQP_SUB_PROTOCOL, true,
new DefaultHttpHeaders());
}
@Override
public void channelActive(ChannelHandlerContext context) throws Exception {
LOG.trace("Channel has become active! Channel is {}", context.channel());
handshaker.handshake(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();
}
}
@Override
public void exceptionCaught(ChannelHandlerContext context, Throwable cause) throws Exception {
LOG.trace("Exception on channel! Channel is {} -> {}", context.channel(), cause.getMessage());
LOG.trace("Error Stack: ", cause);
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;
}
if (!handshakeFuture.isDone()) {
handshakeFuture.setFailure(cause);
}
}
super.channelActive(context);
}
@Override
@ -427,16 +131,17 @@ public class NettyWSTransport implements NettyTransport {
Channel ch = ctx.channel();
if (!handshaker.isHandshakeComplete()) {
handshaker.finishHandshake(ch, (FullHttpResponse) message);
LOG.info("WebSocket Client connected! {}", ctx.channel());
handshakeFuture.setSuccess();
LOG.trace("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(CharsetUtil.UTF_8) + ')');
throw new IllegalStateException(
"Unexpected FullHttpResponse (getStatus=" + response.status() + ", content=" + response.content().toString(StandardCharsets.UTF_8) + ')');
}
WebSocketFrame frame = (WebSocketFrame) message;
@ -446,10 +151,11 @@ public class NettyWSTransport implements NettyTransport {
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());
LOG.trace("WebSocket Client received data: {} bytes", binaryFrame.content().readableBytes());
listener.onData(binaryFrame.content());
} else if (frame instanceof PongWebSocketFrame) {
LOG.trace("WebSocket Client received pong");
} else if (frame instanceof PingWebSocketFrame) {
LOG.trace("WebSocket Client received ping, response with pong");
ch.write(new PongWebSocketFrame(frame.content()));
} else if (frame instanceof CloseWebSocketFrame) {
LOG.trace("WebSocket Client received closing");
ch.close();

202
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableConnection.java
View File

@ -1,202 +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.activemq.transport.amqp.client.util;
import java.util.EnumSet;
import java.util.Map;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
import org.apache.qpid.proton.engine.Collector;
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.EndpointState;
import org.apache.qpid.proton.engine.Link;
import org.apache.qpid.proton.engine.Record;
import org.apache.qpid.proton.engine.Session;
import org.apache.qpid.proton.engine.Transport;
import org.apache.qpid.proton.reactor.Reactor;
/**
* Unmodifiable Connection wrapper used to prevent test code from accidentally
* modifying Connection state.
*/
public class UnmodifiableConnection implements Connection {
private final Connection connection;
public UnmodifiableConnection(Connection connection) {
this.connection = connection;
}
@Override
public EndpointState getLocalState() {
return connection.getLocalState();
}
@Override
public EndpointState getRemoteState() {
return connection.getRemoteState();
}
@Override
public ErrorCondition getCondition() {
return connection.getCondition();
}
@Override
public void setCondition(ErrorCondition condition) {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public ErrorCondition getRemoteCondition() {
return connection.getRemoteCondition();
}
@Override
public void free() {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public void open() {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public void close() {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public Session session() {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public Session sessionHead(EnumSet<EndpointState> local, EnumSet<EndpointState> remote) {
Session head = connection.sessionHead(local, remote);
if (head != null) {
head = new UnmodifiableSession(head);
}
return head;
}
@Override
public Link linkHead(EnumSet<EndpointState> local, EnumSet<EndpointState> remote) {
// TODO - If implemented this method should return an unmodifiable link instance.
return null;
}
@Override
public Delivery getWorkHead() {
// TODO - If implemented this method should return an unmodifiable delivery instance.
return null;
}
@Override
public void setContainer(String container) {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public void setHostname(String hostname) {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public String getHostname() {
return connection.getHostname();
}
@Override
public String getRemoteContainer() {
return connection.getRemoteContainer();
}
@Override
public String getRemoteHostname() {
return connection.getRemoteHostname();
}
@Override
public void setOfferedCapabilities(Symbol[] capabilities) {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public void setDesiredCapabilities(Symbol[] capabilities) {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public Symbol[] getRemoteOfferedCapabilities() {
return connection.getRemoteOfferedCapabilities();
}
@Override
public Symbol[] getRemoteDesiredCapabilities() {
return connection.getRemoteDesiredCapabilities();
}
@Override
public Map<Symbol, Object> getRemoteProperties() {
return connection.getRemoteProperties();
}
@Override
public void setProperties(Map<Symbol, Object> properties) {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public Object getContext() {
return connection.getContext();
}
@Override
public void setContext(Object context) {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public void collect(Collector collector) {
throw new UnsupportedOperationException("Cannot alter the Connection");
}
@Override
public String getContainer() {
return connection.getContainer();
}
@Override
public Transport getTransport() {
return new UnmodifiableTransport(connection.getTransport());
}
@Override
public Record attachments() {
return connection.attachments();
}
@Override
public Reactor getReactor() {
return connection.getReactor();
}
}

179
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableDelivery.java
View File

@ -1,179 +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.activemq.transport.amqp.client.util;
import org.apache.qpid.proton.amqp.transport.DeliveryState;
import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.Link;
import org.apache.qpid.proton.engine.Receiver;
import org.apache.qpid.proton.engine.Record;
import org.apache.qpid.proton.engine.Sender;
/**
* Unmodifiable Delivery wrapper used to prevent test code from accidentally
* modifying Delivery state.
*/
public class UnmodifiableDelivery implements Delivery {
private final Delivery delivery;
public UnmodifiableDelivery(Delivery delivery) {
this.delivery = delivery;
}
@Override
public byte[] getTag() {
return delivery.getTag();
}
@Override
public Link getLink() {
if (delivery.getLink() instanceof Sender) {
return new UnmodifiableSender((Sender) delivery.getLink());
} else if (delivery.getLink() instanceof Receiver) {
return new UnmodifiableReceiver((Receiver) delivery.getLink());
} else {
throw new IllegalStateException("Delivery has unknown link type");
}
}
/* waiting Pull Request sent
@Override
public int getDataLength() {
return delivery.getDataLength();
} */
@Override
public int available() {
return delivery.available();
}
@Override
public DeliveryState getLocalState() {
return delivery.getLocalState();
}
@Override
public DeliveryState getRemoteState() {
return delivery.getRemoteState();
}
@Override
public int getMessageFormat() {
return delivery.getMessageFormat();
}
@Override
public void disposition(DeliveryState state) {
throw new UnsupportedOperationException("Cannot alter the Delivery state");
}
@Override
public void settle() {
throw new UnsupportedOperationException("Cannot alter the Delivery state");
}
@Override
public boolean isSettled() {
return delivery.isSettled();
}
@Override
public boolean remotelySettled() {
return delivery.remotelySettled();
}
@Override
public void free() {
throw new UnsupportedOperationException("Cannot alter the Delivery state");
}
@Override
public Delivery getWorkNext() {
return new UnmodifiableDelivery(delivery.getWorkNext());
}
@Override
public Delivery next() {
return new UnmodifiableDelivery(delivery.next());
}
@Override
public boolean isWritable() {
return delivery.isWritable();
}
@Override
public boolean isReadable() {
return delivery.isReadable();
}
@Override
public void setContext(Object o) {
throw new UnsupportedOperationException("Cannot alter the Delivery state");
}
@Override
public Object getContext() {
return delivery.getContext();
}
@Override
public boolean isUpdated() {
return delivery.isUpdated();
}
@Override
public void clear() {
throw new UnsupportedOperationException("Cannot alter the Delivery state");
}
@Override
public boolean isPartial() {
return delivery.isPartial();
}
@Override
public int pending() {
return delivery.pending();
}
@Override
public boolean isBuffered() {
return delivery.isBuffered();
}
@Override
public Record attachments() {
return delivery.attachments();
}
@Override
public DeliveryState getDefaultDeliveryState() {
return delivery.getDefaultDeliveryState();
}
@Override
public void setDefaultDeliveryState(DeliveryState state) {
throw new UnsupportedOperationException("Cannot alter the Delivery");
}
@Override
public void setMessageFormat(int messageFormat) {
throw new UnsupportedOperationException("Cannot alter the Delivery");
}
}

306
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableLink.java
View File

@ -1,306 +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.activemq.transport.amqp.client.util;
import java.util.EnumSet;
import java.util.Map;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode;
import org.apache.qpid.proton.amqp.transport.SenderSettleMode;
import org.apache.qpid.proton.amqp.transport.Source;
import org.apache.qpid.proton.amqp.transport.Target;
import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.EndpointState;
import org.apache.qpid.proton.engine.Link;
import org.apache.qpid.proton.engine.Receiver;
import org.apache.qpid.proton.engine.Record;
import org.apache.qpid.proton.engine.Sender;
import org.apache.qpid.proton.engine.Session;
/**
* Unmodifiable Session wrapper used to prevent test code from accidentally
* modifying Session state.
*/
public class UnmodifiableLink implements Link {
private final Link link;
public UnmodifiableLink(Link link) {
this.link = link;
}
@Override
public EndpointState getLocalState() {
return link.getLocalState();
}
@Override
public EndpointState getRemoteState() {
return link.getRemoteState();
}
@Override
public ErrorCondition getCondition() {
return link.getCondition();
}
@Override
public void setCondition(ErrorCondition condition) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public ErrorCondition getRemoteCondition() {
return link.getRemoteCondition();
}
@Override
public void free() {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public void open() {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public void close() {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public void setContext(Object o) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public Object getContext() {
return link.getContext();
}
@Override
public String getName() {
return link.getName();
}
@Override
public Delivery delivery(byte[] tag) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public Delivery delivery(byte[] tag, int offset, int length) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public Delivery head() {
return new UnmodifiableDelivery(link.head());
}
@Override
public Delivery current() {
return new UnmodifiableDelivery(link.current());
}
@Override
public boolean advance() {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public Source getSource() {
// TODO Figure out a simple way to wrap the odd Source types in Proton-J
return link.getSource();
}
@Override
public Target getTarget() {
// TODO Figure out a simple way to wrap the odd Source types in Proton-J
return link.getTarget();
}
@Override
public void setSource(Source address) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public void setTarget(Target address) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public Source getRemoteSource() {
// TODO Figure out a simple way to wrap the odd Source types in Proton-J
return link.getRemoteSource();
}
@Override
public Target getRemoteTarget() {
// TODO Figure out a simple way to wrap the odd Target types in Proton-J
return link.getRemoteTarget();
}
@Override
public Link next(EnumSet<EndpointState> local, EnumSet<EndpointState> remote) {
Link next = link.next(local, remote);
if (next != null) {
if (next instanceof Sender) {
next = new UnmodifiableSender((Sender) next);
} else {
next = new UnmodifiableReceiver((Receiver) next);
}
}
return next;
}
@Override
public int getCredit() {
return link.getCredit();
}
@Override
public int getQueued() {
return link.getQueued();
}
@Override
public int getUnsettled() {
return link.getUnsettled();
}
@Override
public Session getSession() {
return new UnmodifiableSession(link.getSession());
}
@Override
public SenderSettleMode getSenderSettleMode() {
return link.getSenderSettleMode();
}
@Override
public void setSenderSettleMode(SenderSettleMode senderSettleMode) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public SenderSettleMode getRemoteSenderSettleMode() {
return link.getRemoteSenderSettleMode();
}
@Override
public ReceiverSettleMode getReceiverSettleMode() {
return link.getReceiverSettleMode();
}
@Override
public void setReceiverSettleMode(ReceiverSettleMode receiverSettleMode) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public ReceiverSettleMode getRemoteReceiverSettleMode() {
return link.getRemoteReceiverSettleMode();
}
@Override
public void setRemoteSenderSettleMode(SenderSettleMode remoteSenderSettleMode) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public int drained() {
return link.drained(); // TODO - Is this a mutating call?
}
@Override
public int getRemoteCredit() {
return link.getRemoteCredit();
}
@Override
public boolean getDrain() {
return link.getDrain();
}
@Override
public void detach() {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public boolean detached() {
return link.detached();
}
@Override
public Record attachments() {
return link.attachments();
}
@Override
public Map<Symbol, Object> getProperties() {
return link.getProperties();
}
@Override
public void setProperties(Map<Symbol, Object> properties) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public Map<Symbol, Object> getRemoteProperties() {
return link.getRemoteProperties();
}
@Override
public Symbol[] getDesiredCapabilities() {
return link.getDesiredCapabilities();
}
@Override
public Symbol[] getOfferedCapabilities() {
return link.getOfferedCapabilities();
}
@Override
public Symbol[] getRemoteDesiredCapabilities() {
return link.getRemoteDesiredCapabilities();
}
@Override
public Symbol[] getRemoteOfferedCapabilities() {
return link.getRemoteOfferedCapabilities();
}
@Override
public void setDesiredCapabilities(Symbol[] capabilities) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public void setOfferedCapabilities(Symbol[] capabilities) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
}

167
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableProxy.java Normal file
View File

@ -0,0 +1,167 @@
/*
* 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.activemq.transport.amqp.client.util;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Map;
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.Link;
import org.apache.qpid.proton.engine.Receiver;
import org.apache.qpid.proton.engine.Sasl;
import org.apache.qpid.proton.engine.Sender;
import org.apache.qpid.proton.engine.Session;
import org.apache.qpid.proton.engine.Transport;
/**
* Utility that creates proxy objects for the Proton objects which won't allow any mutating
* operations to be applied so that the test code does not interact with the proton engine
* outside the client serialization thread.
*/
public final class UnmodifiableProxy {
private static ArrayList<String> blacklist = new ArrayList<>();
// These methods are mutating but don't take an arguments so they
// aren't automatically filtered out. We will have to keep an eye
// on proton API in the future and modify this list as it evolves.
static {
blacklist.add("close");
blacklist.add("free");
blacklist.add("open");
blacklist.add("sasl");
blacklist.add("session");
blacklist.add("close_head");
blacklist.add("close_tail");
blacklist.add("outputConsumed");
blacklist.add("process");
blacklist.add("processInput");
blacklist.add("unbind");
blacklist.add("settle");
blacklist.add("clear");
blacklist.add("detach");
blacklist.add("abort");
}
private UnmodifiableProxy() {
}
public static Transport transportProxy(final Transport target) {
Transport wrap = wrap(Transport.class, target);
return wrap;
}
public static Sasl saslProxy(final Sasl target) {
return wrap(Sasl.class, target);
}
public static Connection connectionProxy(final Connection target) {
return wrap(Connection.class, target);
}
public static Session sessionProxy(final Session target) {
return wrap(Session.class, target);
}
public static Delivery deliveryProxy(final Delivery target) {
return wrap(Delivery.class, target);
}
public static Link linkProxy(final Link target) {
return wrap(Link.class, target);
}
public static Receiver receiverProxy(final Receiver target) {
return wrap(Receiver.class, target);
}
public static Sender senderProxy(final Sender target) {
return wrap(Sender.class, target);
}
private static boolean isProtonType(Class<?> clazz) {
String packageName = clazz.getPackage().getName();
if (packageName.startsWith("org.apache.qpid.proton.")) {
return true;
}
return false;
}
private static <T> T wrap(Class<T> type, final Object target) {
return type.cast(java.lang.reflect.Proxy.newProxyInstance(type.getClassLoader(), new Class[] {type}, new InvocationHandler() {
@Override
public Object invoke(Object o, Method method, Object[] objects) throws Throwable {
if ("toString".equals(method.getName()) && method.getParameterTypes().length == 0) {
return "Unmodifiable proxy -> (" + method.invoke(target, objects) + ")";
}
// Don't let methods that mutate be invoked.
if (method.getParameterTypes().length > 0) {
throw new UnsupportedOperationException("Cannot mutate outside the Client work thread");
}
if (blacklist.contains(method.getName())) {
throw new UnsupportedOperationException("Cannot mutate outside the Client work thread");
}
Class<?> returnType = method.getReturnType();
try {
Object result = method.invoke(target, objects);
if (result == null) {
return null;
}
if (returnType.isPrimitive() || returnType.isArray() || Object.class.equals(returnType)) {
// Skip any other checks
} else if (returnType.isAssignableFrom(ByteBuffer.class)) {
// Buffers are modifiable but we can just return null to indicate
// there's nothing there to access.
result = null;
} else if (returnType.isAssignableFrom(Map.class)) {
// Prevent return of modifiable maps
result = Collections.unmodifiableMap((Map<?, ?>) result);
} else if (isProtonType(returnType) && returnType.isInterface()) {
// Can't handle the crazy Source / Target types yet as there's two
// different types for Source and Target the result can't be cast to
// the one people actually want to use.
if (!returnType.getName().equals("org.apache.qpid.proton.amqp.transport.Source")
&& !returnType.getName().equals("org.apache.qpid.proton.amqp.messaging.Source")
&& !returnType.getName().equals("org.apache.qpid.proton.amqp.transport.Target")
&& !returnType.getName().equals("org.apache.qpid.proton.amqp.messaging.Target")) {
result = wrap(returnType, result);
}
}
return result;
} catch (InvocationTargetException e) {
throw e.getTargetException();
}
}
}));
}
}

65
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableReceiver.java
View File

@ -1,65 +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.activemq.transport.amqp.client.util;
import org.apache.qpid.proton.codec.WritableBuffer;
import org.apache.qpid.proton.engine.Receiver;
/**
* Unmodifiable Receiver wrapper used to prevent test code from accidentally
* modifying Receiver state.
*/
public class UnmodifiableReceiver extends UnmodifiableLink implements Receiver {
private final Receiver receiver;
public UnmodifiableReceiver(Receiver receiver) {
super(receiver);
this.receiver = receiver;
}
@Override
public void flow(int credits) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public int recv(byte[] bytes, int offset, int size) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public int recv(WritableBuffer buffer) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public void drain(int credit) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public boolean draining() {
return receiver.draining();
}
@Override
public void setDrain(boolean drain) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
}

51
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSender.java
View File

@ -1,51 +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.activemq.transport.amqp.client.util;
import org.apache.qpid.proton.codec.ReadableBuffer;
import org.apache.qpid.proton.engine.Sender;
/**
* Unmodifiable Sender wrapper used to prevent test code from accidentally
* modifying Sender state.
*/
public class UnmodifiableSender extends UnmodifiableLink implements Sender {
public UnmodifiableSender(Sender sender) {
super(sender);
}
@Override
public void offer(int credits) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public int send(byte[] bytes, int offset, int length) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public int send(ReadableBuffer buffer) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public void abort() {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
}

198
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableSession.java
View File

@ -1,198 +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.activemq.transport.amqp.client.util;
import java.util.EnumSet;
import java.util.Map;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.EndpointState;
import org.apache.qpid.proton.engine.Receiver;
import org.apache.qpid.proton.engine.Record;
import org.apache.qpid.proton.engine.Sender;
import org.apache.qpid.proton.engine.Session;
/**
* Unmodifiable Session wrapper used to prevent test code from accidentally
* modifying Session state.
*/
public class UnmodifiableSession implements Session {
private final Session session;
public UnmodifiableSession(Session session) {
this.session = session;
}
@Override
public EndpointState getLocalState() {
return session.getLocalState();
}
@Override
public EndpointState getRemoteState() {
return session.getRemoteState();
}
@Override
public ErrorCondition getCondition() {
return session.getCondition();
}
@Override
public void setCondition(ErrorCondition condition) {
throw new UnsupportedOperationException("Cannot alter the Session");
}
@Override
public ErrorCondition getRemoteCondition() {
return session.getRemoteCondition();
}
@Override
public void free() {
throw new UnsupportedOperationException("Cannot alter the Session");
}
@Override
public void open() {
throw new UnsupportedOperationException("Cannot alter the Session");
}
@Override
public void close() {
throw new UnsupportedOperationException("Cannot alter the Session");
}
@Override
public void setContext(Object o) {
throw new UnsupportedOperationException("Cannot alter the Session");
}
@Override
public Object getContext() {
return session.getContext();
}
@Override
public Sender sender(String name) {
throw new UnsupportedOperationException("Cannot alter the Session");
}
@Override
public Receiver receiver(String name) {
throw new UnsupportedOperationException("Cannot alter the Session");
}
@Override
public Session next(EnumSet<EndpointState> local, EnumSet<EndpointState> remote) {
Session next = session.next(local, remote);
if (next != null) {
next = new UnmodifiableSession(next);
}
return next;
}
@Override
public Connection getConnection() {
return new UnmodifiableConnection(session.getConnection());
}
@Override
public int getIncomingCapacity() {
return session.getIncomingCapacity();
}
@Override
public void setIncomingCapacity(int bytes) {
throw new UnsupportedOperationException("Cannot alter the Session");
}
@Override
public int getIncomingBytes() {
return session.getIncomingBytes();
}
@Override
public int getOutgoingBytes() {
return session.getOutgoingBytes();
}
@Override
public Record attachments() {
return session.attachments();
}
@Override
public long getOutgoingWindow() {
return session.getOutgoingWindow();
}
@Override
public void setOutgoingWindow(long outgoingWindowSize) {
throw new UnsupportedOperationException("Cannot alter the Session");
}
@Override
public Symbol[] getDesiredCapabilities() {
return session.getDesiredCapabilities();
}
@Override
public Symbol[] getOfferedCapabilities() {
return session.getOfferedCapabilities();
}
@Override
public Map<Symbol, Object> getProperties() {
return session.getProperties();
}
@Override
public Symbol[] getRemoteDesiredCapabilities() {
return session.getRemoteDesiredCapabilities();
}
@Override
public Symbol[] getRemoteOfferedCapabilities() {
return session.getRemoteOfferedCapabilities();
}
@Override
public Map<Symbol, Object> getRemoteProperties() {
return session.getRemoteProperties();
}
@Override
public void setDesiredCapabilities(Symbol[] capabilities) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public void setOfferedCapabilities(Symbol[] capabilities) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
@Override
public void setProperties(Map<Symbol, Object> capabilities) {
throw new UnsupportedOperationException("Cannot alter the Link state");
}
}

274
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableTransport.java
View File

@ -1,274 +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.activemq.transport.amqp.client.util;
import java.nio.ByteBuffer;
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.EndpointState;
import org.apache.qpid.proton.engine.Record;
import org.apache.qpid.proton.engine.Sasl;
import org.apache.qpid.proton.engine.Ssl;
import org.apache.qpid.proton.engine.SslDomain;
import org.apache.qpid.proton.engine.SslPeerDetails;
import org.apache.qpid.proton.engine.Transport;
import org.apache.qpid.proton.engine.TransportException;
import org.apache.qpid.proton.engine.TransportResult;
/**
* Unmodifiable Transport wrapper used to prevent test code from accidentally
* modifying Transport state.
*/
public class UnmodifiableTransport implements Transport {
private final Transport transport;
public UnmodifiableTransport(Transport transport) {
this.transport = transport;
}
@Override
public void close() {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void free() {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public Object getContext() {
return null;
}
@Override
public EndpointState getLocalState() {
return transport.getLocalState();
}
@Override
public ErrorCondition getRemoteCondition() {
return transport.getRemoteCondition();
}
@Override
public EndpointState getRemoteState() {
return transport.getRemoteState();
}
@Override
public void open() {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void setCondition(ErrorCondition arg0) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void setContext(Object arg0) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void bind(Connection arg0) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public int capacity() {
return transport.capacity();
}
@Override
public void close_head() {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void close_tail() {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public int getChannelMax() {
return transport.getChannelMax();
}
@Override
public ErrorCondition getCondition() {
return transport.getCondition();
}
@Override
public int getIdleTimeout() {
return transport.getIdleTimeout();
}
@Override
public ByteBuffer getInputBuffer() {
return null;
}
@Override
public int getMaxFrameSize() {
return transport.getMaxFrameSize();
}
@Override
public ByteBuffer getOutputBuffer() {
return null;
}
@Override
public int getRemoteChannelMax() {
return transport.getRemoteChannelMax();
}
@Override
public int getRemoteIdleTimeout() {
return transport.getRemoteIdleTimeout();
}
@Override
public int getRemoteMaxFrameSize() {
return transport.getRemoteMaxFrameSize();
}
@Override
public ByteBuffer head() {
return null;
}
@Override
public int input(byte[] arg0, int arg1, int arg2) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public boolean isClosed() {
return transport.isClosed();
}
@Override
public int output(byte[] arg0, int arg1, int arg2) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void outputConsumed() {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public int pending() {
return transport.pending();
}
@Override
public void pop(int arg0) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void process() throws TransportException {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public TransportResult processInput() {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public Sasl sasl() throws IllegalStateException {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void setChannelMax(int arg0) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void setIdleTimeout(int arg0) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void setMaxFrameSize(int arg0) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public Ssl ssl(SslDomain arg0) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public Ssl ssl(SslDomain arg0, SslPeerDetails arg1) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public ByteBuffer tail() {
return null;
}
@Override
public long tick(long arg0) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void trace(int arg0) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public void unbind() {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public Record attachments() {
return transport.attachments();
}
@Override
public long getFramesInput() {
return transport.getFramesInput();
}
@Override
public long getFramesOutput() {
return transport.getFramesOutput();
}
@Override
public void setEmitFlowEventOnSend(boolean emitFlowEventOnSend) {
throw new UnsupportedOperationException("Cannot alter the Transport");
}
@Override
public boolean isEmitFlowEventOnSend() {
return transport.isEmitFlowEventOnSend();
}
}