From 7d14321fb1a64a98cef69ef00dd1aadea71f81e6 Mon Sep 17 00:00:00 2001 From: Robert Godfrey Date: Mon, 5 Oct 2009 14:34:39 +0000 Subject: Merged from java-network-refactor up to r816261 git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/java-broker-0-10@821809 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/java/broker/build.xml | 1 + .../java/org/apache/qpid/server/AMQChannel.java | 1 - .../src/main/java/org/apache/qpid/server/Main.java | 199 ++-- .../server/configuration/ServerConfiguration.java | 56 +- .../qpid/server/connection/ConnectionRegistry.java | 9 +- .../server/connection/IConnectionRegistry.java | 1 - .../server/protocol/AMQMinaProtocolSession.java | 923 ----------------- .../server/protocol/AMQPFastProtocolHandler.java | 290 ------ .../qpid/server/protocol/AMQPProtocolProvider.java | 52 - .../qpid/server/protocol/AMQProtocolEngine.java | 1040 ++++++++++++++++++++ .../server/protocol/AMQProtocolEngineFactory.java | 29 + .../qpid/server/protocol/AMQProtocolSession.java | 17 + .../server/protocol/AMQProtocolSessionMBean.java | 42 +- .../qpid/server/registry/ApplicationRegistry.java | 2 +- .../access/plugins/network/FirewallPlugin.java | 33 +- .../apache/qpid/server/transport/QpidAcceptor.java | 12 +- .../qpid/server/virtualhost/VirtualHost.java | 8 + .../configuration/ServerConfigurationTest.java | 56 +- .../protocol/AMQProtocolSessionMBeanTest.java | 2 +- .../protocol/InternalTestProtocolSession.java | 24 +- .../qpid/server/protocol/MaxChannelsTest.java | 2 +- .../apache/qpid/server/protocol/TestIoSession.java | 328 ------ .../qpid/server/queue/AMQQueueAlertTest.java | 49 +- .../access/plugins/network/FirewallPluginTest.java | 39 +- qpid/java/client/build.xml | 1 + .../qpid/client/AMQConnectionDelegate_8_0.java | 2 +- .../qpid/client/failover/FailoverHandler.java | 10 +- .../protocol/AMQIoTransportProtocolSession.java | 146 --- .../qpid/client/protocol/AMQProtocolHandler.java | 397 ++++---- .../qpid/client/protocol/AMQProtocolSession.java | 110 +-- .../qpid/client/transport/QpidThreadExecutor.java | 22 - .../transport/SocketTransportConnection.java | 71 +- .../qpid/client/transport/TransportConnection.java | 44 +- .../transport/VmPipeTransportConnection.java | 17 +- .../client/protocol/AMQProtocolHandlerTest.java | 5 +- .../org/apache/qpid/codec/AMQCodecFactory.java | 7 +- .../java/org/apache/qpid/codec/AMQDecoder.java | 79 +- .../apache/qpid/framing/AMQDataBlockDecoder.java | 39 +- .../apache/qpid/framing/AMQDataBlockEncoder.java | 2 +- .../apache/qpid/framing/ProtocolInitiation.java | 20 +- .../src/main/java/org/apache/qpid/pool/Event.java | 155 --- .../src/main/java/org/apache/qpid/pool/Job.java | 139 ++- .../java/org/apache/qpid/pool/PoolingFilter.java | 491 --------- .../org/apache/qpid/pool/ReadWriteRunnable.java | 1 - .../org/apache/qpid/pool/ReadWriteThreadModel.java | 102 -- .../org/apache/qpid/protocol/ProtocolEngine.java | 65 ++ .../qpid/protocol/ProtocolEngineFactory.java | 31 + .../org/apache/qpid/thread/QpidThreadExecutor.java | 43 + .../org/apache/qpid/transport/NetworkDriver.java | 63 ++ .../qpid/transport/NetworkDriverConfiguration.java | 44 + .../org/apache/qpid/transport/OpenException.java | 34 + .../transport/network/mina/MINANetworkDriver.java | 418 ++++++++ .../java/org/apache/qpid/codec/AMQDecoderTest.java | 130 +++ .../codec/MockAMQVersionAwareProtocolSession.java | 95 ++ .../org/apache/qpid/pool/PoolingFilterTest.java | 111 --- .../apache/qpid/transport/TestNetworkDriver.java | 122 +++ .../network/mina/MINANetworkDriverTest.java | 487 +++++++++ qpid/java/systests/build.xml | 2 +- .../qpid/server/security/acl/SimpleACLTest.java | 99 +- .../client/protocol/AMQProtocolSessionTest.java | 25 +- .../apache/qpid/test/utils/FailoverBaseCase.java | 4 +- .../org/apache/qpid/test/utils/QpidTestCase.java | 2 +- 62 files changed, 3435 insertions(+), 3415 deletions(-) delete mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java delete mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPFastProtocolHandler.java delete mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPProtocolProvider.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java create mode 100644 qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngineFactory.java delete mode 100644 qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/TestIoSession.java delete mode 100644 qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQIoTransportProtocolSession.java delete mode 100644 qpid/java/client/src/main/java/org/apache/qpid/client/transport/QpidThreadExecutor.java delete mode 100644 qpid/java/common/src/main/java/org/apache/qpid/pool/Event.java delete mode 100644 qpid/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java delete mode 100644 qpid/java/common/src/main/java/org/apache/qpid/pool/ReadWriteThreadModel.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/thread/QpidThreadExecutor.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkDriver.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkDriverConfiguration.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/transport/OpenException.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/transport/network/mina/MINANetworkDriver.java create mode 100644 qpid/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java create mode 100644 qpid/java/common/src/test/java/org/apache/qpid/codec/MockAMQVersionAwareProtocolSession.java delete mode 100644 qpid/java/common/src/test/java/org/apache/qpid/pool/PoolingFilterTest.java create mode 100644 qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkDriver.java create mode 100644 qpid/java/common/src/test/java/org/apache/qpid/transport/network/mina/MINANetworkDriverTest.java diff --git a/qpid/java/broker/build.xml b/qpid/java/broker/build.xml index 3c63c459be..ae133d1a96 100644 --- a/qpid/java/broker/build.xml +++ b/qpid/java/broker/build.xml @@ -21,6 +21,7 @@ + diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java index 4f24f7ddc0..5aaa18860f 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java @@ -41,7 +41,6 @@ import org.apache.qpid.server.exchange.NoRouteException; import org.apache.qpid.server.flow.FlowCreditManager; import org.apache.qpid.server.flow.Pre0_10CreditManager; import org.apache.qpid.server.protocol.AMQProtocolSession; -import org.apache.qpid.server.protocol.AMQMinaProtocolSession; import org.apache.qpid.server.queue.*; import org.apache.qpid.server.subscription.Subscription; import org.apache.qpid.server.subscription.SubscriptionFactoryImpl; diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java index 8a1437f3d6..d8b0f7aec7 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java @@ -20,6 +20,13 @@ */ package org.apache.qpid.server; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.util.Properties; + import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.HelpFormatter; import org.apache.commons.cli.Option; @@ -30,19 +37,12 @@ import org.apache.commons.cli.PosixParser; import org.apache.log4j.Logger; import org.apache.log4j.PropertyConfigurator; import org.apache.log4j.xml.QpidLog4JConfigurator; -import org.apache.mina.common.ByteBuffer; -import org.apache.mina.common.FixedSizeByteBufferAllocator; -import org.apache.mina.common.IoAcceptor; -import org.apache.mina.transport.socket.nio.SocketAcceptorConfig; -import org.apache.mina.transport.socket.nio.SocketSessionConfig; -import org.apache.mina.util.NewThreadExecutor; import org.apache.qpid.AMQException; import org.apache.qpid.transport.network.io.IoTransport; import org.apache.qpid.transport.network.ConnectionBinding; import org.apache.qpid.transport.*; import org.apache.qpid.common.QpidProperties; import org.apache.qpid.framing.ProtocolVersion; -import org.apache.qpid.pool.ReadWriteThreadModel; import org.apache.qpid.server.configuration.ServerConfiguration; import org.apache.qpid.server.configuration.management.ConfigurationManagementMBean; import org.apache.qpid.server.information.management.ServerInformationMBean; @@ -51,22 +51,16 @@ import org.apache.qpid.server.logging.actors.BrokerActor; import org.apache.qpid.server.logging.actors.CurrentActor; import org.apache.qpid.server.logging.management.LoggingManagementMBean; import org.apache.qpid.server.logging.messages.BrokerMessages; -import org.apache.qpid.server.protocol.AMQPFastProtocolHandler; -import org.apache.qpid.server.protocol.AMQPProtocolProvider; +import org.apache.qpid.server.protocol.AMQProtocolEngineFactory; import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry; import org.apache.qpid.server.registry.IApplicationRegistry; import org.apache.qpid.server.security.auth.manager.AuthenticationManager; import org.apache.qpid.server.transport.ServerConnection; import org.apache.qpid.server.transport.QpidAcceptor; - -import java.io.File; -import java.io.IOException; -import java.io.InputStream; -import java.net.BindException; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.util.Properties; +import org.apache.qpid.ssl.SSLContextFactory; +import org.apache.qpid.transport.NetworkDriver; +import org.apache.qpid.transport.network.mina.MINANetworkDriver; /** * Main entry point for AMQPD. @@ -307,20 +301,6 @@ public class Main _brokerLogger.info("Starting Qpid Broker " + QpidProperties.getReleaseVersion() + " build: " + QpidProperties.getBuildVersion()); - ByteBuffer.setUseDirectBuffers(serverConfig.getEnableDirectBuffers()); - - // the MINA default is currently to use the pooled allocator although this may change in future - // once more testing of the performance of the simple allocator has been done - if (!serverConfig.getEnablePooledAllocator()) - { - ByteBuffer.setAllocator(new FixedSizeByteBufferAllocator()); - } - - if (serverConfig.getUseBiasedWrites()) - { - System.setProperty("org.apache.qpid.use_write_biased_pool", "true"); - } - int port = serverConfig.getPort(); String portStr = commandLine.getOptionValue("p"); @@ -336,7 +316,54 @@ public class Main } } - bind(port, serverConfig); + String bindAddr = commandLine.getOptionValue("b"); + if (bindAddr == null) + { + bindAddr = serverConfig.getBind(); + } + InetAddress bindAddress = null; + if (bindAddr.equals("wildcard")) + { + bindAddress = new InetSocketAddress(port).getAddress(); + } + else + { + bindAddress = InetAddress.getByAddress(parseIP(bindAddr)); + } + + String keystorePath = serverConfig.getKeystorePath(); + String keystorePassword = serverConfig.getKeystorePassword(); + String certType = serverConfig.getCertType(); + SSLContextFactory sslFactory = null; + boolean isSsl = false; + + if (!serverConfig.getSSLOnly()) + { + NetworkDriver driver = new MINANetworkDriver(); + driver.bind(port, new InetAddress[]{bindAddress}, new AMQProtocolEngineFactory(), + serverConfig.getNetworkConfiguration(), null); + ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, port), + new QpidAcceptor(driver,"TCP")); + CurrentActor.get().message(BrokerMessages.BRK_1002("TCP", port)); + } + + if (serverConfig.getEnableSSL()) + { + sslFactory = new SSLContextFactory(keystorePath, keystorePassword, certType); + NetworkDriver driver = new MINANetworkDriver(); + driver.bind(serverConfig.getSSLPort(), new InetAddress[]{bindAddress}, + new AMQProtocolEngineFactory(), serverConfig.getNetworkConfiguration(), sslFactory); + ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, port), + new QpidAcceptor(driver,"TCP")); + CurrentActor.get().message(BrokerMessages.BRK_1002("TCP/SSL", serverConfig.getSSLPort())); + } + + //fixme qpid.AMQP should be using qpidproperties to get value + _brokerLogger.info("Qpid Broker Ready :" + QpidProperties.getReleaseVersion() + + " build: " + QpidProperties.getBuildVersion()); + + CurrentActor.get().message(BrokerMessages.BRK_1004()); + // TODO - Fix to use a proper binding int port_0_10 = port + 1; @@ -391,114 +418,6 @@ public class Main } } - protected void bind(int port, ServerConfiguration config) throws BindException - { - String bindAddr = commandLine.getOptionValue("b"); - if (bindAddr == null) - { - bindAddr = config.getBind(); - } - - try - { - IoAcceptor acceptor; - - if (ApplicationRegistry.getInstance().getConfiguration().getQpidNIO()) - { - _logger.warn("Using Qpid Multithreaded IO Processing"); - acceptor = new org.apache.mina.transport.socket.nio.MultiThreadSocketAcceptor(config.getProcessors(), new NewThreadExecutor()); - } - else - { - _logger.warn("Using Mina IO Processing"); - acceptor = new org.apache.mina.transport.socket.nio.SocketAcceptor(config.getProcessors(), new NewThreadExecutor()); - } - - SocketAcceptorConfig sconfig = (SocketAcceptorConfig) acceptor.getDefaultConfig(); - SocketSessionConfig sc = (SocketSessionConfig) sconfig.getSessionConfig(); - - sc.setReceiveBufferSize(config.getReceiveBufferSize()); - sc.setSendBufferSize(config.getWriteBufferSize()); - sc.setTcpNoDelay(config.getTcpNoDelay()); - - // if we do not use the executor pool threading model we get the default leader follower - // implementation provided by MINA - if (config.getEnableExecutorPool()) - { - sconfig.setThreadModel(ReadWriteThreadModel.getInstance()); - } - - if (!config.getEnableSSL() || !config.getSSLOnly()) - { - AMQPFastProtocolHandler handler = new AMQPProtocolProvider().getHandler(); - InetSocketAddress bindAddress; - if (bindAddr.equals("wildcard")) - { - bindAddress = new InetSocketAddress(port); - } - else - { - bindAddress = new InetSocketAddress(InetAddress.getByAddress(parseIP(bindAddr)), port); - } - - bind(new QpidAcceptor(acceptor,"TCP"), bindAddress, handler, sconfig); - - //fixme qpid.AMQP should be using qpidproperties to get value - _brokerLogger.info("Qpid.AMQP listening on non-SSL address " + bindAddress); - } - - if (config.getEnableSSL()) - { - AMQPFastProtocolHandler handler = new AMQPProtocolProvider().getHandler(); - try - { - - bind(new QpidAcceptor(acceptor, "TCP/SSL"), new InetSocketAddress(config.getSSLPort()), handler, sconfig); - - //fixme qpid.AMQP should be using qpidproperties to get value - _brokerLogger.info("Qpid.AMQP listening on SSL port " + config.getSSLPort()); - - } - catch (IOException e) - { - _brokerLogger.error("Unable to listen on SSL port: " + e, e); - } - } - - //fixme qpid.AMQP should be using qpidproperties to get value - _brokerLogger.info("Qpid Broker Ready :" + QpidProperties.getReleaseVersion() - + " build: " + QpidProperties.getBuildVersion()); - - CurrentActor.get().message(BrokerMessages.BRK_1004()); - - } - catch (Exception e) - { - _logger.error("Unable to bind service to registry: " + e, e); - //fixme this need tidying up - throw new BindException(e.getMessage()); - } - } - - /** - * Ensure that any bound Acceptors are recorded in the registry so they can be closed later. - * - * @param acceptor - * @param bindAddress - * @param handler - * @param sconfig - * - * @throws IOException from the acceptor.bind command - */ - private void bind(QpidAcceptor acceptor, InetSocketAddress bindAddress, AMQPFastProtocolHandler handler, SocketAcceptorConfig sconfig) throws IOException - { - acceptor.getIoAcceptor().bind(bindAddress, handler, sconfig); - - CurrentActor.get().message(BrokerMessages.BRK_1002(acceptor.toString(), bindAddress.getPort())); - - ApplicationRegistry.getInstance().addAcceptor(bindAddress, acceptor); - } - public static void main(String[] args) { //if the -Dlog4j.configuration property has not been set, enable the init override diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java index 01befbbfe8..641b44bb18 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java @@ -38,6 +38,7 @@ import org.apache.qpid.server.configuration.management.ConfigurationManagementMB import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.virtualhost.VirtualHost; import org.apache.qpid.server.virtualhost.VirtualHostRegistry; +import org.apache.qpid.transport.NetworkDriverConfiguration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -570,7 +571,7 @@ public class ServerConfiguration implements SignalHandler public boolean getSSLOnly() { - return getConfig().getBoolean("connector.ssl.sslOnly", true); + return getConfig().getBoolean("connector.ssl.sslOnly", false); } public int getSSLPort() @@ -619,4 +620,57 @@ public class ServerConfiguration implements SignalHandler getConfig().getLong("housekeeping.expiredMessageCheckPeriod", DEFAULT_HOUSEKEEPING_PERIOD)); } + + public NetworkDriverConfiguration getNetworkConfiguration() + { + return new NetworkDriverConfiguration() + { + + public Integer getTrafficClass() + { + return null; + } + + public Boolean getTcpNoDelay() + { + // Can't call parent getTcpNoDelay since it just calls this one + return getConfig().getBoolean("connector.tcpNoDelay", true); + } + + public Integer getSoTimeout() + { + return null; + } + + public Integer getSoLinger() + { + return null; + } + + public Integer getSendBufferSize() + { + return getBufferWriteLimit(); + } + + public Boolean getReuseAddress() + { + return null; + } + + public Integer getReceiveBufferSize() + { + return getBufferReadLimit(); + } + + public Boolean getOOBInline() + { + return null; + } + + public Boolean getKeepAlive() + { + return null; + } + }; + } } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java index d287595e2d..7b50a2e3ad 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java @@ -21,7 +21,6 @@ package org.apache.qpid.server.connection; import org.apache.qpid.server.virtualhost.VirtualHost; -import org.apache.qpid.server.protocol.AMQMinaProtocolSession; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.AMQException; import org.apache.qpid.AMQConnectionException; @@ -45,6 +44,14 @@ public class ConnectionRegistry implements IConnectionRegistry { } + + public void expireClosedChannels() + { + for (AMQProtocolSession connection : _registry) + { + connection.closeIfLingeringClosedChannels(); + } + } /** Close all of the currently open connections. */ public void close() throws AMQException diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java index d64fde1c20..002269bbaa 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/IConnectionRegistry.java @@ -20,7 +20,6 @@ */ package org.apache.qpid.server.connection; -import org.apache.qpid.server.protocol.AMQMinaProtocolSession; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.AMQException; diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java deleted file mode 100644 index 1235d5ffe5..0000000000 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java +++ /dev/null @@ -1,923 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ -package org.apache.qpid.server.protocol; - -import org.apache.log4j.Logger; -import org.apache.mina.common.CloseFuture; -import org.apache.mina.common.IdleStatus; -import org.apache.mina.common.IoServiceConfig; -import org.apache.mina.common.IoSession; -import org.apache.mina.transport.vmpipe.VmPipeAddress; -import org.apache.qpid.AMQChannelException; -import org.apache.qpid.AMQConnectionException; -import org.apache.qpid.AMQException; -import org.apache.qpid.codec.AMQCodecFactory; -import org.apache.qpid.codec.AMQDecoder; -import org.apache.qpid.common.ClientProperties; -import org.apache.qpid.framing.AMQBody; -import org.apache.qpid.framing.AMQDataBlock; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.AMQMethodBody; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.ChannelCloseOkBody; -import org.apache.qpid.framing.ContentBody; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.framing.HeartbeatBody; -import org.apache.qpid.framing.MethodDispatcher; -import org.apache.qpid.framing.MethodRegistry; -import org.apache.qpid.framing.ProtocolInitiation; -import org.apache.qpid.framing.ProtocolVersion; -import org.apache.qpid.pool.ReadWriteThreadModel; -import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.protocol.AMQMethodEvent; -import org.apache.qpid.protocol.AMQMethodListener; -import org.apache.qpid.server.AMQChannel; -import org.apache.qpid.server.handler.ServerMethodDispatcherImpl; -import org.apache.qpid.server.logging.actors.AMQPConnectionActor; -import org.apache.qpid.server.logging.actors.CurrentActor; -import org.apache.qpid.server.logging.subjects.ConnectionLogSubject; -import org.apache.qpid.server.logging.LogSubject; -import org.apache.qpid.server.logging.LogActor; -import org.apache.qpid.server.logging.messages.ConnectionMessages; -import org.apache.qpid.server.management.Managable; -import org.apache.qpid.server.management.ManagedObject; -import org.apache.qpid.server.output.ProtocolOutputConverter; -import org.apache.qpid.server.output.ProtocolOutputConverterRegistry; -import org.apache.qpid.server.registry.ApplicationRegistry; -import org.apache.qpid.server.state.AMQState; -import org.apache.qpid.server.state.AMQStateManager; -import org.apache.qpid.server.virtualhost.VirtualHost; -import org.apache.qpid.server.virtualhost.VirtualHostRegistry; -import org.apache.qpid.transport.Sender; - -import javax.management.JMException; -import javax.security.sasl.SaslServer; -import java.net.InetSocketAddress; -import java.net.SocketAddress; -import java.security.Principal; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.CopyOnWriteArraySet; -import java.util.concurrent.atomic.AtomicLong; - -public class AMQMinaProtocolSession implements AMQProtocolSession, Managable -{ - private static final Logger _logger = Logger.getLogger(AMQProtocolSession.class); - - private static final String CLIENT_PROPERTIES_INSTANCE = ClientProperties.instance.toString(); - - private static final AtomicLong idGenerator = new AtomicLong(0); - - // to save boxing the channelId and looking up in a map... cache in an array the low numbered - // channels. This value must be of the form 2^x - 1. - private static final int CHANNEL_CACHE_SIZE = 0xff; - - private final IoSession _minaProtocolSession; - - private AMQShortString _contextKey; - - private AMQShortString _clientVersion = null; - - private VirtualHost _virtualHost; - - private final Map _channelMap = new HashMap(); - - private final AMQChannel[] _cachedChannels = new AMQChannel[CHANNEL_CACHE_SIZE + 1]; - - private final CopyOnWriteArraySet _frameListeners = new CopyOnWriteArraySet(); - - private final AMQStateManager _stateManager; - - private AMQCodecFactory _codecFactory; - - private AMQProtocolSessionMBean _managedObject; - - private SaslServer _saslServer; - - private Object _lastReceived; - - private Object _lastSent; - - protected boolean _closed; - // maximum number of channels this session should have - private long _maxNoOfChannels = 1000; - - /* AMQP Version for this session */ - private ProtocolVersion _protocolVersion = ProtocolVersion.getLatestSupportedVersion(); - - private FieldTable _clientProperties; - private final List _taskList = new CopyOnWriteArrayList(); - - private List _closingChannelsList = new CopyOnWriteArrayList(); - private ProtocolOutputConverter _protocolOutputConverter; - private Principal _authorizedID; - private MethodDispatcher _dispatcher; - private ProtocolSessionIdentifier _sessionIdentifier; - - private static final long LAST_WRITE_FUTURE_JOIN_TIMEOUT = 60000L; - private org.apache.mina.common.WriteFuture _lastWriteFuture; - - // Create a simple ID that increments for ever new Session - private final long _sessionID = idGenerator.getAndIncrement(); - - private AMQPConnectionActor _actor; - private LogSubject _logSubject; - - public ManagedObject getManagedObject() - { - return _managedObject; - } - - public AMQMinaProtocolSession(IoSession session, VirtualHostRegistry virtualHostRegistry, AMQCodecFactory codecFactory) - throws AMQException - { - _stateManager = new AMQStateManager(virtualHostRegistry, this); - _minaProtocolSession = session; - session.setAttachment(this); - - _codecFactory = codecFactory; - - _actor = new AMQPConnectionActor(this, virtualHostRegistry.getApplicationRegistry().getRootMessageLogger()); - - _actor.message(ConnectionMessages.CON_1001(null, null, false, false)); - - try - { - IoServiceConfig config = session.getServiceConfig(); - ReadWriteThreadModel threadModel = (ReadWriteThreadModel) config.getThreadModel(); - threadModel.getAsynchronousReadFilter().createNewJobForSession(session); - threadModel.getAsynchronousWriteFilter().createNewJobForSession(session); - } - catch (RuntimeException e) - { - e.printStackTrace(); - throw e; - - } - } - - private AMQProtocolSessionMBean createMBean() throws AMQException - { - try - { - return new AMQProtocolSessionMBean(this); - } - catch (JMException ex) - { - _logger.error("AMQProtocolSession MBean creation has failed ", ex); - throw new AMQException("AMQProtocolSession MBean creation has failed ", ex); - } - } - - public IoSession getIOSession() - { - return _minaProtocolSession; - } - - public static AMQProtocolSession getAMQProtocolSession(IoSession minaProtocolSession) - { - return (AMQProtocolSession) minaProtocolSession.getAttachment(); - } - - public long getSessionID() - { - return _sessionID; - } - - public LogActor getLogActor() - { - return _actor; - } - - public void dataBlockReceived(AMQDataBlock message) throws Exception - { - _lastReceived = message; - if (message instanceof ProtocolInitiation) - { - protocolInitiationReceived((ProtocolInitiation) message); - - } - else if (message instanceof AMQFrame) - { - AMQFrame frame = (AMQFrame) message; - frameReceived(frame); - - } - else - { - throw new UnknnownMessageTypeException(message); - } - } - - private void frameReceived(AMQFrame frame) throws AMQException - { - int channelId = frame.getChannel(); - AMQBody body = frame.getBodyFrame(); - - //Look up the Channel's Actor and set that as the current actor - // If that is not available then we can use the ConnectionActor - // that is associated with this AMQMPSession. - LogActor channelActor = null; - if (_channelMap.get(channelId) != null) - { - channelActor = _channelMap.get(channelId).getLogActor(); - } - CurrentActor.set(channelActor == null ? _actor : channelActor); - - try - { - if (_logger.isDebugEnabled()) - { - _logger.debug("Frame Received: " + frame); - } - - // Check that this channel is not closing - if (channelAwaitingClosure(channelId)) - { - if ((frame.getBodyFrame() instanceof ChannelCloseOkBody)) - { - if (_logger.isInfoEnabled()) - { - _logger.info("Channel[" + channelId + "] awaiting closure - processing close-ok"); - } - } - else - { - if (_logger.isInfoEnabled()) - { - _logger.info("Channel[" + channelId + "] awaiting closure. Should close socket as client did not close-ok :" + frame); - } - - closeProtocolSession(); - return; - } - } - - try - { - body.handle(channelId, this); - } - catch (AMQException e) - { - closeChannel(channelId); - throw e; - } - } - finally - { - CurrentActor.remove(); - } - } - - private void protocolInitiationReceived(ProtocolInitiation pi) - { - // this ensures the codec never checks for a PI message again - ((AMQDecoder) _codecFactory.getDecoder()).setExpectProtocolInitiation(false); - try - { - // Log incomming protocol negotiation request - _actor.message(ConnectionMessages.CON_1001(null, pi._protocolMajor + "-" + pi._protocolMinor, false, true)); - - ProtocolVersion pv = pi.checkVersion(); // Fails if not correct - - // This sets the protocol version (and hence framing classes) for this session. - setProtocolVersion(pv); - - String mechanisms = ApplicationRegistry.getInstance().getAuthenticationManager().getMechanisms(); - - String locales = "en_US"; - - AMQMethodBody responseBody = getMethodRegistry().createConnectionStartBody((short) getProtocolMajorVersion(), - (short) getProtocolMinorVersion(), - null, - mechanisms.getBytes(), - locales.getBytes()); - _minaProtocolSession.write(responseBody.generateFrame(0)); - - } - catch (AMQException e) - { - _logger.info("Received unsupported protocol initiation for protocol version: " + getProtocolVersion()); - - _minaProtocolSession.write(new ProtocolInitiation(ProtocolVersion.getLatestSupportedVersion())); - - // TODO: Close connection (but how to wait until message is sent?) - // ritchiem 2006-12-04 will this not do? - // WriteFuture future = _minaProtocolSession.write(new ProtocolInitiation(pv[i][PROTOCOLgetProtocolMajorVersion()], pv[i][PROTOCOLgetProtocolMinorVersion()])); - // future.join(); - // close connection - - } - } - - public void methodFrameReceived(int channelId, AMQMethodBody methodBody) - { - - final AMQMethodEvent evt = new AMQMethodEvent(channelId, methodBody); - - try - { - try - { - - boolean wasAnyoneInterested = _stateManager.methodReceived(evt); - - if (!_frameListeners.isEmpty()) - { - for (AMQMethodListener listener : _frameListeners) - { - wasAnyoneInterested = listener.methodReceived(evt) || wasAnyoneInterested; - } - } - - if (!wasAnyoneInterested) - { - throw new AMQNoMethodHandlerException(evt); - } - } - catch (AMQChannelException e) - { - if (getChannel(channelId) != null) - { - if (_logger.isInfoEnabled()) - { - _logger.info("Closing channel due to: " + e.getMessage()); - } - - writeFrame(e.getCloseFrame(channelId)); - closeChannel(channelId); - } - else - { - if (_logger.isDebugEnabled()) - { - _logger.debug("ChannelException occured on non-existent channel:" + e.getMessage()); - } - - if (_logger.isInfoEnabled()) - { - _logger.info("Closing connection due to: " + e.getMessage()); - } - - AMQConnectionException ce = - evt.getMethod().getConnectionException(AMQConstant.CHANNEL_ERROR, - AMQConstant.CHANNEL_ERROR.getName().toString()); - - closeConnection(channelId, ce, false); - } - } - catch (AMQConnectionException e) - { - closeConnection(channelId, e, false); - } - } - catch (Exception e) - { - - for (AMQMethodListener listener : _frameListeners) - { - listener.error(e); - } - - _logger.error("Unexpected exception while processing frame. Closing connection.", e); - - closeProtocolSession(); - } - } - - public void contentHeaderReceived(int channelId, ContentHeaderBody body) throws AMQException - { - - AMQChannel channel = getAndAssertChannel(channelId); - - channel.publishContentHeader(body); - - } - - public void contentBodyReceived(int channelId, ContentBody body) throws AMQException - { - AMQChannel channel = getAndAssertChannel(channelId); - - channel.publishContentBody(body); - } - - public void heartbeatBodyReceived(int channelId, HeartbeatBody body) - { - // NO - OP - } - - /** - * Convenience method that writes a frame to the protocol session. Equivalent to calling - * getProtocolSession().write(). - * - * @param frame the frame to write - */ - public void writeFrame(AMQDataBlock frame) - { - _lastSent = frame; - - _lastWriteFuture = _minaProtocolSession.write(frame); - } - - public AMQShortString getContextKey() - { - return _contextKey; - } - - public void setContextKey(AMQShortString contextKey) - { - _contextKey = contextKey; - } - - public List getChannels() - { - return new ArrayList(_channelMap.values()); - } - - public AMQChannel getAndAssertChannel(int channelId) throws AMQException - { - AMQChannel channel = getChannel(channelId); - if (channel == null) - { - throw new AMQException(AMQConstant.NOT_FOUND, "Channel not found with id:" + channelId); - } - - return channel; - } - - public AMQChannel getChannel(int channelId) throws AMQException - { - final AMQChannel channel = - ((channelId & CHANNEL_CACHE_SIZE) == channelId) ? _cachedChannels[channelId] : _channelMap.get(channelId); - if ((channel == null) || channel.isClosing()) - { - return null; - } - else - { - return channel; - } - } - - public boolean channelAwaitingClosure(int channelId) - { - return !_closingChannelsList.isEmpty() && _closingChannelsList.contains(channelId); - } - - public void addChannel(AMQChannel channel) throws AMQException - { - if (_closed) - { - throw new AMQException("Session is closed"); - } - - final int channelId = channel.getChannelId(); - - if (_closingChannelsList.contains(channelId)) - { - throw new AMQException("Session is marked awaiting channel close"); - } - - if (_channelMap.size() == _maxNoOfChannels) - { - String errorMessage = - toString() + ": maximum number of channels has been reached (" + _maxNoOfChannels - + "); can't create channel"; - _logger.error(errorMessage); - throw new AMQException(AMQConstant.NOT_ALLOWED, errorMessage); - } - else - { - _channelMap.put(channel.getChannelId(), channel); - } - - if (((channelId & CHANNEL_CACHE_SIZE) == channelId)) - { - _cachedChannels[channelId] = channel; - } - - checkForNotification(); - } - - private void checkForNotification() - { - int channelsCount = _channelMap.size(); - if (channelsCount >= _maxNoOfChannels) - { - _managedObject.notifyClients("Channel count (" + channelsCount + ") has reached the threshold value"); - } - } - - public Long getMaximumNumberOfChannels() - { - return _maxNoOfChannels; - } - - public void setMaximumNumberOfChannels(Long value) - { - _maxNoOfChannels = value; - } - - public void commitTransactions(AMQChannel channel) throws AMQException - { - if ((channel != null) && channel.isTransactional()) - { - channel.commit(); - } - } - - public void rollbackTransactions(AMQChannel channel) throws AMQException - { - if ((channel != null) && channel.isTransactional()) - { - channel.rollback(); - } - } - - /** - * Close a specific channel. This will remove any resources used by the channel, including:
  • any queue - * subscriptions (this may in turn remove queues if they are auto delete
- * - * @param channelId id of the channel to close - * - * @throws AMQException if an error occurs closing the channel - * @throws IllegalArgumentException if the channel id is not valid - */ - public void closeChannel(int channelId) throws AMQException - { - final AMQChannel channel = getChannel(channelId); - if (channel == null) - { - throw new IllegalArgumentException("Unknown channel id"); - } - else - { - try - { - channel.close(); - markChannelAwaitingCloseOk(channelId); - } - finally - { - removeChannel(channelId); - } - } - } - - public void closeChannelOk(int channelId) - { - // todo QPID-847 - This is called from two lcoations ChannelCloseHandler and ChannelCloseOkHandler. - // When it is the CC_OK_Handler then it makes sence to remove the channel else we will leak memory. - // We do it from the Close Handler as we are sending the OK back to the client. - // While this is AMQP spec compliant. The Java client in the event of an IllegalArgumentException - // will send a close-ok.. Where we should call removeChannel. - // However, due to the poor exception handling on the client. The client-user will be notified of the - // InvalidArgument and if they then decide to close the session/connection then the there will be time - // for that to occur i.e. a new close method be sent before the exeption handling can mark the session closed. - //removeChannel(channelId); - _closingChannelsList.remove(new Integer(channelId)); - } - - private void markChannelAwaitingCloseOk(int channelId) - { - _closingChannelsList.add(channelId); - } - - /** - * In our current implementation this is used by the clustering code. - * - * @param channelId The channel to remove - */ - public void removeChannel(int channelId) - { - _channelMap.remove(channelId); - if ((channelId & CHANNEL_CACHE_SIZE) == channelId) - { - _cachedChannels[channelId] = null; - } - } - - /** - * Initialise heartbeats on the session. - * - * @param delay delay in seconds (not ms) - */ - public void initHeartbeats(int delay) - { - if (delay > 0) - { - _minaProtocolSession.setIdleTime(IdleStatus.WRITER_IDLE, delay); - _minaProtocolSession.setIdleTime(IdleStatus.READER_IDLE, (int) (ApplicationRegistry.getInstance().getConfiguration().getHeartBeatTimeout() * delay)); - } - } - - /** - * Closes all channels that were opened by this protocol session. This frees up all resources used by the channel. - * - * @throws AMQException if an error occurs while closing any channel - */ - private void closeAllChannels() throws AMQException - { - for (AMQChannel channel : _channelMap.values()) - { - channel.close(); - } - - _channelMap.clear(); - for (int i = 0; i <= CHANNEL_CACHE_SIZE; i++) - { - _cachedChannels[i] = null; - } - } - - /** This must be called when the session is _closed in order to free up any resources managed by the session. */ - public void closeSession() throws AMQException - { - if (!_closed) - { - if (_virtualHost != null) - { - _virtualHost.getConnectionRegistry().deregisterConnection(this); - } - - closeAllChannels(); - if (_managedObject != null) - { - _managedObject.unregister(); - } - - for (Task task : _taskList) - { - task.doTask(this); - } - - _closed = true; - - CurrentActor.get().message(_logSubject, ConnectionMessages.CON_1002()); - } - } - - public void closeConnection(int channelId, AMQConnectionException e, boolean closeProtocolSession) throws AMQException - { - if (_logger.isInfoEnabled()) - { - _logger.info("Closing connection due to: " + e.getMessage()); - } - - markChannelAwaitingCloseOk(channelId); - closeSession(); - _stateManager.changeState(AMQState.CONNECTION_CLOSING); - writeFrame(e.getCloseFrame(channelId)); - - if (closeProtocolSession) - { - closeProtocolSession(); - } - } - - public void closeProtocolSession() - { - closeProtocolSession(true); - } - - public void closeProtocolSession(boolean waitLast) - { - if (waitLast && (_lastWriteFuture != null)) - { - _logger.debug("Waiting for last write to join."); - _lastWriteFuture.join(LAST_WRITE_FUTURE_JOIN_TIMEOUT); - } - - _logger.debug("REALLY Closing protocol session:" + _minaProtocolSession); - final CloseFuture future = _minaProtocolSession.close(); - future.join(LAST_WRITE_FUTURE_JOIN_TIMEOUT); - - try - { - _stateManager.changeState(AMQState.CONNECTION_CLOSED); - } - catch (AMQException e) - { - _logger.info(e.getMessage()); - } - } - - public String toString() - { - return _minaProtocolSession.getRemoteAddress() + "(" + (getPrincipal() == null ? "?" : getPrincipal().getName() + ")"); - } - - public String dump() - { - return this + " last_sent=" + _lastSent + " last_received=" + _lastReceived; - } - - /** @return an object that can be used to identity */ - public Object getKey() - { - return _minaProtocolSession.getRemoteAddress(); - } - - /** - * Get the fully qualified domain name of the local address to which this session is bound. Since some servers may - * be bound to multiple addresses this could vary depending on the acceptor this session was created from. - * - * @return a String FQDN - */ - public String getLocalFQDN() - { - SocketAddress address = _minaProtocolSession.getLocalAddress(); - // we use the vmpipe address in some tests hence the need for this rather ugly test. The host - // information is used by SASL primary. - if (address instanceof InetSocketAddress) - { - return ((InetSocketAddress) address).getHostName(); - } - else if (address instanceof VmPipeAddress) - { - return "vmpipe:" + ((VmPipeAddress) address).getPort(); - } - else - { - throw new IllegalArgumentException("Unsupported socket address class: " + address); - } - } - - public SaslServer getSaslServer() - { - return _saslServer; - } - - public void setSaslServer(SaslServer saslServer) - { - _saslServer = saslServer; - } - - public FieldTable getClientProperties() - { - return _clientProperties; - } - - public void setClientProperties(FieldTable clientProperties) - { - _clientProperties = clientProperties; - if (_clientProperties != null) - { - if (_clientProperties.getString(CLIENT_PROPERTIES_INSTANCE) != null) - { - String clientID = _clientProperties.getString(CLIENT_PROPERTIES_INSTANCE); - setContextKey(new AMQShortString(clientID)); - - // Log the Opening of the connection for this client - _actor.message(ConnectionMessages.CON_1001(clientID, _protocolVersion.toString(), true, true)); - } - - if (_clientProperties.getString(ClientProperties.version.toString()) != null) - { - _clientVersion = new AMQShortString(_clientProperties.getString(ClientProperties.version.toString())); - } - } - _sessionIdentifier = new ProtocolSessionIdentifier(this); - } - - private void setProtocolVersion(ProtocolVersion pv) - { - _protocolVersion = pv; - - _protocolOutputConverter = ProtocolOutputConverterRegistry.getConverter(this); - _dispatcher = ServerMethodDispatcherImpl.createMethodDispatcher(_stateManager, _protocolVersion); - } - - public byte getProtocolMajorVersion() - { - return _protocolVersion.getMajorVersion(); - } - - public ProtocolVersion getProtocolVersion() - { - return _protocolVersion; - } - - public byte getProtocolMinorVersion() - { - return _protocolVersion.getMinorVersion(); - } - - public boolean isProtocolVersion(byte major, byte minor) - { - return (getProtocolMajorVersion() == major) && (getProtocolMinorVersion() == minor); - } - - public MethodRegistry getRegistry() - { - return getMethodRegistry(); - } - - public Object getClientIdentifier() - { - return (_minaProtocolSession != null) ? _minaProtocolSession.getRemoteAddress() : null; - } - - public VirtualHost getVirtualHost() - { - return _virtualHost; - } - - public void setVirtualHost(VirtualHost virtualHost) throws AMQException - { - _virtualHost = virtualHost; - - _actor.virtualHostSelected(this); - _logSubject = new ConnectionLogSubject(this); - - _virtualHost.getConnectionRegistry().registerConnection(this); - - _managedObject = createMBean(); - _managedObject.register(); - } - - public void addSessionCloseTask(Task task) - { - _taskList.add(task); - } - - public void removeSessionCloseTask(Task task) - { - _taskList.remove(task); - } - - public ProtocolOutputConverter getProtocolOutputConverter() - { - return _protocolOutputConverter; - } - - public void setAuthorizedID(Principal authorizedID) - { - _authorizedID = authorizedID; - - // Let the actor know that this connection is now Authorized - _actor.connectionAuthorized(this); - } - - public Principal getPrincipal() - { - return _authorizedID; - } - - public SocketAddress getRemoteAddress() - { - return _minaProtocolSession.getRemoteAddress(); - } - - public MethodRegistry getMethodRegistry() - { - return MethodRegistry.getMethodRegistry(getProtocolVersion()); - } - - public MethodDispatcher getMethodDispatcher() - { - return _dispatcher; - } - - public ProtocolSessionIdentifier getSessionIdentifier() - { - return _sessionIdentifier; - } - - public String getClientVersion() - { - return (_clientVersion == null) ? null : _clientVersion.toString(); - } - - public void setSender(Sender sender) - { - // No-op, interface munging between this and AMQProtocolSession - } - - public void init() - { - // No-op, interface munging between this and AMQProtocolSession - } -} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPFastProtocolHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPFastProtocolHandler.java deleted file mode 100644 index 16b85e67b3..0000000000 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPFastProtocolHandler.java +++ /dev/null @@ -1,290 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ -package org.apache.qpid.server.protocol; - -import java.io.IOException; -import java.net.InetSocketAddress; - -import org.apache.log4j.Logger; -import org.apache.mina.common.ByteBuffer; -import org.apache.mina.common.IdleStatus; -import org.apache.mina.common.IoFilterChain; -import org.apache.mina.common.IoHandlerAdapter; -import org.apache.mina.common.IoSession; -import org.apache.mina.filter.ReadThrottleFilterBuilder; -import org.apache.mina.filter.SSLFilter; -import org.apache.mina.filter.WriteBufferLimitFilterBuilder; -import org.apache.mina.filter.codec.QpidProtocolCodecFilter; -import org.apache.mina.filter.executor.ExecutorFilter; -import org.apache.mina.util.SessionUtil; -import org.apache.qpid.AMQException; -import org.apache.qpid.codec.AMQCodecFactory; -import org.apache.qpid.framing.AMQDataBlock; -import org.apache.qpid.framing.AMQProtocolHeaderException; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.ConnectionCloseBody; -import org.apache.qpid.framing.HeartbeatBody; -import org.apache.qpid.framing.MethodRegistry; -import org.apache.qpid.framing.ProtocolInitiation; -import org.apache.qpid.framing.ProtocolVersion; -import org.apache.qpid.server.configuration.ServerConfiguration; -import org.apache.qpid.server.registry.ApplicationRegistry; -import org.apache.qpid.server.registry.IApplicationRegistry; -import org.apache.qpid.server.logging.actors.CurrentActor; -import org.apache.qpid.ssl.SSLContextFactory; - -/** - * The protocol handler handles "protocol events" for all connections. The state - * associated with an individual connection is accessed through the protocol session. - * - * We delegate all frame (message) processing to the AMQProtocolSession which wraps - * the state for the connection. - */ -public class AMQPFastProtocolHandler extends IoHandlerAdapter -{ - private static final Logger _logger = Logger.getLogger(AMQPFastProtocolHandler.class); - - private final IApplicationRegistry _applicationRegistry; - - private final int BUFFER_READ_LIMIT_SIZE; - private final int BUFFER_WRITE_LIMIT_SIZE; - - public AMQPFastProtocolHandler(Integer applicationRegistryInstance) - { - this(ApplicationRegistry.getInstance(applicationRegistryInstance)); - } - - public AMQPFastProtocolHandler(IApplicationRegistry applicationRegistry) - { - _applicationRegistry = applicationRegistry; - - // Read the configuration from the application registry - BUFFER_READ_LIMIT_SIZE = _applicationRegistry.getConfiguration().getBufferReadLimit(); - BUFFER_WRITE_LIMIT_SIZE = _applicationRegistry.getConfiguration().getBufferWriteLimit(); - - _logger.debug("AMQPFastProtocolHandler created"); - } - - protected AMQPFastProtocolHandler(AMQPFastProtocolHandler handler) - { - this(handler._applicationRegistry); - } - - public void sessionCreated(IoSession protocolSession) throws Exception - { - SessionUtil.initialize(protocolSession); - final AMQCodecFactory codecFactory = new AMQCodecFactory(true); - - createSession(protocolSession, _applicationRegistry, codecFactory); - _logger.info("Protocol session created for:" + protocolSession.getRemoteAddress()); - - final QpidProtocolCodecFilter pcf = new QpidProtocolCodecFilter(codecFactory); - final ServerConfiguration config = _applicationRegistry.getConfiguration(); - - String keystorePath = config.getKeystorePath(); - String keystorePassword = config.getKeystorePassword(); - String certType = config.getCertType(); - SSLContextFactory sslContextFactory = null; - boolean isSsl = false; - if (config.getEnableSSL() && isSSLClient(config, protocolSession)) - { - sslContextFactory = new SSLContextFactory(keystorePath, keystorePassword, certType); - isSsl = true; - } - if (config.getEnableExecutorPool()) - { - if (isSsl) - { - protocolSession.getFilterChain().addAfter("AsynchronousReadFilter", "sslFilter", - new SSLFilter(sslContextFactory.buildServerContext())); - } - protocolSession.getFilterChain().addBefore("AsynchronousWriteFilter", "protocolFilter", pcf); - } - else - { - protocolSession.getFilterChain().addLast("protocolFilter", pcf); - if (isSsl) - { - protocolSession.getFilterChain().addBefore("protocolFilter", "sslFilter", - new SSLFilter(sslContextFactory.buildServerContext())); - } - } - - if (ApplicationRegistry.getInstance().getConfiguration().getProtectIOEnabled()) - { - try - { -// //Add IO Protection Filters - IoFilterChain chain = protocolSession.getFilterChain(); - - - protocolSession.getFilterChain().addLast("tempExecutorFilterForFilterBuilder", new ExecutorFilter()); - - ReadThrottleFilterBuilder readfilter = new ReadThrottleFilterBuilder(); - readfilter.setMaximumConnectionBufferSize(BUFFER_READ_LIMIT_SIZE); - readfilter.attach(chain); - - WriteBufferLimitFilterBuilder writefilter = new WriteBufferLimitFilterBuilder(); - writefilter.setMaximumConnectionBufferSize(BUFFER_WRITE_LIMIT_SIZE); - writefilter.attach(chain); - - protocolSession.getFilterChain().remove("tempExecutorFilterForFilterBuilder"); - _logger.info("Using IO Read/Write Filter Protection"); - } - catch (Exception e) - { - _logger.error("Unable to attach IO Read/Write Filter Protection :" + e.getMessage()); - } - } - } - - /** Separated into its own, protected, method to allow easier reuse */ - protected void createSession(IoSession session, IApplicationRegistry applicationRegistry, AMQCodecFactory codec) throws AMQException - { - new AMQMinaProtocolSession(session, applicationRegistry.getVirtualHostRegistry(), codec); - } - - public void sessionOpened(IoSession protocolSession) throws Exception - { - _logger.info("Session opened for:" + protocolSession.getRemoteAddress()); - } - - public void sessionClosed(IoSession protocolSession) throws Exception - { - _logger.info("Protocol Session closed for:" + protocolSession.getRemoteAddress()); - final AMQProtocolSession amqProtocolSession = AMQMinaProtocolSession.getAMQProtocolSession(protocolSession); - //fixme -- this can be null - if (amqProtocolSession != null) - { - try - { - CurrentActor.set(amqProtocolSession.getLogActor()); - amqProtocolSession.closeSession(); - } - catch (AMQException e) - { - _logger.error("Caught AMQException whilst closingSession:" + e); - } - finally - { - CurrentActor.remove(); - } - } - } - - public void sessionIdle(IoSession session, IdleStatus status) throws Exception - { - _logger.debug("Protocol Session [" + this + "] idle: " + status + " :for:" + session.getRemoteAddress()); - if (IdleStatus.WRITER_IDLE.equals(status)) - { - //write heartbeat frame: - session.write(HeartbeatBody.FRAME); - } - else if (IdleStatus.READER_IDLE.equals(status)) - { - //failover: - throw new IOException("Timed out while waiting for heartbeat from peer."); - } - - } - - public void exceptionCaught(IoSession protocolSession, Throwable throwable) throws Exception - { - AMQProtocolSession session = AMQMinaProtocolSession.getAMQProtocolSession(protocolSession); - if (throwable instanceof AMQProtocolHeaderException) - { - - protocolSession.write(new ProtocolInitiation(ProtocolVersion.getLatestSupportedVersion())); - - protocolSession.close(); - - _logger.error("Error in protocol initiation " + session + ":" + protocolSession.getRemoteAddress() + " :" + throwable.getMessage(), throwable); - } - else if (throwable instanceof IOException) - { - _logger.error("IOException caught in" + session + ", session closed implictly: " + throwable); - } - else - { - _logger.error("Exception caught in" + session + ", closing session explictly: " + throwable, throwable); - - - MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(session.getProtocolVersion()); - ConnectionCloseBody closeBody = methodRegistry.createConnectionCloseBody(200,new AMQShortString(throwable.getMessage()),0,0); - - protocolSession.write(closeBody.generateFrame(0)); - - protocolSession.close(); - } - } - - /** - * Invoked when a message is received on a particular protocol session. Note that a - * protocol session is directly tied to a particular physical connection. - * - * @param protocolSession the protocol session that received the message - * @param message the message itself (i.e. a decoded frame) - * - * @throws Exception if the message cannot be processed - */ - public void messageReceived(IoSession protocolSession, Object message) throws Exception - { - final AMQProtocolSession amqProtocolSession = AMQMinaProtocolSession.getAMQProtocolSession(protocolSession); - - if (message instanceof AMQDataBlock) - { - amqProtocolSession.dataBlockReceived((AMQDataBlock) message); - - } - else if (message instanceof ByteBuffer) - { - throw new IllegalStateException("Handed undecoded ByteBuffer buf = " + message); - } - else - { - throw new IllegalStateException("Handed unhandled message. message.class = " + message.getClass() + " message = " + message); - } - } - - /** - * Called after a message has been sent out on a particular protocol session - * - * @param protocolSession the protocol session (i.e. connection) on which this - * message was sent - * @param object the message (frame) that was encoded and sent - * - * @throws Exception if we want to indicate an error - */ - public void messageSent(IoSession protocolSession, Object object) throws Exception - { - if (_logger.isDebugEnabled()) - { - _logger.debug("Message sent: " + object); - } - } - - protected boolean isSSLClient(ServerConfiguration connectionConfig, - IoSession protocolSession) - { - InetSocketAddress addr = (InetSocketAddress) protocolSession.getLocalAddress(); - return addr.getPort() == connectionConfig.getSSLPort(); - } -} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPProtocolProvider.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPProtocolProvider.java deleted file mode 100644 index 07c153bfe8..0000000000 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPProtocolProvider.java +++ /dev/null @@ -1,52 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ -package org.apache.qpid.server.protocol; - -import org.apache.qpid.server.registry.ApplicationRegistry; -import org.apache.qpid.server.registry.IApplicationRegistry; - -/** - * The protocol provide's role is to encapsulate the initialisation of the protocol handler. - * - * The protocol handler (see AMQPFastProtocolHandler class) handles protocol events - * such as connection closing or a frame being received. It can either do this directly - * or pass off to the protocol session in the cases where state information is required to - * deal with the event. - * - */ -public class AMQPProtocolProvider -{ - /** - * Handler for protocol events - */ - private AMQPFastProtocolHandler _handler; - - public AMQPProtocolProvider() - { - IApplicationRegistry registry = ApplicationRegistry.getInstance(); - _handler = new AMQPFastProtocolHandler(registry); - } - - public AMQPFastProtocolHandler getHandler() - { - return _handler; - } -} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java new file mode 100644 index 0000000000..be23af9d82 --- /dev/null +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java @@ -0,0 +1,1040 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.server.protocol; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.nio.ByteBuffer; +import java.security.Principal; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.atomic.AtomicLong; + +import javax.management.JMException; +import javax.security.sasl.SaslServer; + +import org.apache.log4j.Logger; +import org.apache.mina.transport.vmpipe.VmPipeAddress; +import org.apache.qpid.AMQChannelException; +import org.apache.qpid.AMQConnectionException; +import org.apache.qpid.AMQException; +import org.apache.qpid.codec.AMQCodecFactory; +import org.apache.qpid.codec.AMQDecoder; +import org.apache.qpid.common.ClientProperties; +import org.apache.qpid.framing.AMQBody; +import org.apache.qpid.framing.AMQDataBlock; +import org.apache.qpid.framing.AMQFrame; +import org.apache.qpid.framing.AMQMethodBody; +import org.apache.qpid.framing.AMQProtocolHeaderException; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.ChannelCloseOkBody; +import org.apache.qpid.framing.ConnectionCloseBody; +import org.apache.qpid.framing.ContentBody; +import org.apache.qpid.framing.ContentHeaderBody; +import org.apache.qpid.framing.FieldTable; +import org.apache.qpid.framing.HeartbeatBody; +import org.apache.qpid.framing.MethodDispatcher; +import org.apache.qpid.framing.MethodRegistry; +import org.apache.qpid.framing.ProtocolInitiation; +import org.apache.qpid.framing.ProtocolVersion; +import org.apache.qpid.pool.Job; +import org.apache.qpid.pool.ReferenceCountingExecutorService; +import org.apache.qpid.protocol.AMQConstant; +import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQMethodListener; +import org.apache.qpid.protocol.ProtocolEngine; +import org.apache.qpid.server.AMQChannel; +import org.apache.qpid.server.handler.ServerMethodDispatcherImpl; +import org.apache.qpid.server.logging.LogActor; +import org.apache.qpid.server.logging.LogSubject; +import org.apache.qpid.server.logging.actors.AMQPConnectionActor; +import org.apache.qpid.server.logging.actors.CurrentActor; +import org.apache.qpid.server.logging.messages.ConnectionMessages; +import org.apache.qpid.server.logging.subjects.ConnectionLogSubject; +import org.apache.qpid.server.management.Managable; +import org.apache.qpid.server.management.ManagedObject; +import org.apache.qpid.server.output.ProtocolOutputConverter; +import org.apache.qpid.server.output.ProtocolOutputConverterRegistry; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.server.state.AMQState; +import org.apache.qpid.server.state.AMQStateManager; +import org.apache.qpid.server.virtualhost.VirtualHost; +import org.apache.qpid.server.virtualhost.VirtualHostRegistry; +import org.apache.qpid.transport.NetworkDriver; +import org.apache.qpid.transport.Sender; + +public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocolSession +{ + private static final Logger _logger = Logger.getLogger(AMQProtocolEngine.class); + + private static final String CLIENT_PROPERTIES_INSTANCE = ClientProperties.instance.toString(); + + private static final AtomicLong idGenerator = new AtomicLong(0); + + // to save boxing the channelId and looking up in a map... cache in an array the low numbered + // channels. This value must be of the form 2^x - 1. + private static final int CHANNEL_CACHE_SIZE = 0xff; + + private AMQShortString _contextKey; + + private AMQShortString _clientVersion = null; + + private VirtualHost _virtualHost; + + private final Map _channelMap = new HashMap(); + + private final AMQChannel[] _cachedChannels = new AMQChannel[CHANNEL_CACHE_SIZE + 1]; + + private final CopyOnWriteArraySet _frameListeners = new CopyOnWriteArraySet(); + + private final AMQStateManager _stateManager; + + private AMQCodecFactory _codecFactory; + + private AMQProtocolSessionMBean _managedObject; + + private SaslServer _saslServer; + + private Object _lastReceived; + + private Object _lastSent; + + protected boolean _closed; + // maximum number of channels this session should have + private long _maxNoOfChannels = 1000; + + /* AMQP Version for this session */ + private ProtocolVersion _protocolVersion = ProtocolVersion.getLatestSupportedVersion(); + + private FieldTable _clientProperties; + private final List _taskList = new CopyOnWriteArrayList(); + + private Map _closingChannelsList = new ConcurrentHashMap(); + private ProtocolOutputConverter _protocolOutputConverter; + private Principal _authorizedID; + private MethodDispatcher _dispatcher; + private ProtocolSessionIdentifier _sessionIdentifier; + + // Create a simple ID that increments for ever new Session + private final long _sessionID = idGenerator.getAndIncrement(); + + private AMQPConnectionActor _actor; + private LogSubject _logSubject; + + private NetworkDriver _networkDriver; + + private long _lastIoTime; + + private long _writtenBytes; + private long _readBytes; + + private Job _readJob; + private Job _writeJob; + + private ReferenceCountingExecutorService _poolReference = ReferenceCountingExecutorService.getInstance(); + + public ManagedObject getManagedObject() + { + return _managedObject; + } + + public AMQProtocolEngine(VirtualHostRegistry virtualHostRegistry, NetworkDriver driver) + { + _stateManager = new AMQStateManager(virtualHostRegistry, this); + _networkDriver = driver; + + _codecFactory = new AMQCodecFactory(true, this); + _poolReference.acquireExecutorService(); + _readJob = new Job(_poolReference, Job.MAX_JOB_EVENTS, true); + _writeJob = new Job(_poolReference, Job.MAX_JOB_EVENTS, false); + + _actor = new AMQPConnectionActor(this, virtualHostRegistry.getApplicationRegistry().getRootMessageLogger()); + _actor.message(ConnectionMessages.CON_1001(null, null, false, false)); + + } + + private AMQProtocolSessionMBean createMBean() throws AMQException + { + try + { + return new AMQProtocolSessionMBean(this); + } + catch (JMException ex) + { + _logger.error("AMQProtocolSession MBean creation has failed ", ex); + throw new AMQException("AMQProtocolSession MBean creation has failed ", ex); + } + } + + public long getSessionID() + { + return _sessionID; + } + + public LogActor getLogActor() + { + return _actor; + } + + @Override + public void received(final ByteBuffer msg) + { + _lastIoTime = System.currentTimeMillis(); + try + { + final ArrayList dataBlocks = _codecFactory.getDecoder().decodeBuffer(msg); + Job.fireAsynchEvent(_poolReference.getPool(), _readJob, new Runnable() + { + @Override + public void run() + { + // Decode buffer + + for (AMQDataBlock dataBlock : dataBlocks) + { + try + { + dataBlockReceived(dataBlock); + } + catch (Exception e) + { + _logger.error("Unexpected exception when processing datablock", e); + closeProtocolSession(); + } + } + } + }); + } + catch (Exception e) + { + _logger.error("Unexpected exception when processing datablock", e); + closeProtocolSession(); + } + } + + public void dataBlockReceived(AMQDataBlock message) throws Exception + { + _lastReceived = message; + if (message instanceof ProtocolInitiation) + { + protocolInitiationReceived((ProtocolInitiation) message); + + } + else if (message instanceof AMQFrame) + { + AMQFrame frame = (AMQFrame) message; + frameReceived(frame); + + } + else + { + throw new UnknnownMessageTypeException(message); + } + } + + private void frameReceived(AMQFrame frame) throws AMQException + { + int channelId = frame.getChannel(); + AMQBody body = frame.getBodyFrame(); + + //Look up the Channel's Actor and set that as the current actor + // If that is not available then we can use the ConnectionActor + // that is associated with this AMQMPSession. + LogActor channelActor = null; + if (_channelMap.get(channelId) != null) + { + channelActor = _channelMap.get(channelId).getLogActor(); + } + CurrentActor.set(channelActor == null ? _actor : channelActor); + + try + { + if (_logger.isDebugEnabled()) + { + _logger.debug("Frame Received: " + frame); + } + + // Check that this channel is not closing + if (channelAwaitingClosure(channelId)) + { + if ((frame.getBodyFrame() instanceof ChannelCloseOkBody)) + { + if (_logger.isInfoEnabled()) + { + _logger.info("Channel[" + channelId + "] awaiting closure - processing close-ok"); + } + } + else + { + // The channel has been told to close, we don't process any more frames until + // it's closed. + return; + } + } + + try + { + body.handle(channelId, this); + } + catch (AMQException e) + { + closeChannel(channelId); + throw e; + } + } + finally + { + CurrentActor.remove(); + } + } + + private void protocolInitiationReceived(ProtocolInitiation pi) + { + // this ensures the codec never checks for a PI message again + ((AMQDecoder) _codecFactory.getDecoder()).setExpectProtocolInitiation(false); + try + { + // Log incomming protocol negotiation request + _actor.message(ConnectionMessages.CON_1001(null, pi._protocolMajor + "-" + pi._protocolMinor, false, true)); + + ProtocolVersion pv = pi.checkVersion(); // Fails if not correct + + // This sets the protocol version (and hence framing classes) for this session. + setProtocolVersion(pv); + + String mechanisms = ApplicationRegistry.getInstance().getAuthenticationManager().getMechanisms(); + + String locales = "en_US"; + + AMQMethodBody responseBody = getMethodRegistry().createConnectionStartBody((short) getProtocolMajorVersion(), + (short) getProtocolMinorVersion(), + null, + mechanisms.getBytes(), + locales.getBytes()); + _networkDriver.send(responseBody.generateFrame(0).toNioByteBuffer()); + + } + catch (AMQException e) + { + _logger.info("Received unsupported protocol initiation for protocol version: " + getProtocolVersion()); + + _networkDriver.send(new ProtocolInitiation(ProtocolVersion.getLatestSupportedVersion()).toNioByteBuffer()); + } + } + + public void methodFrameReceived(int channelId, AMQMethodBody methodBody) + { + + final AMQMethodEvent evt = new AMQMethodEvent(channelId, methodBody); + + try + { + try + { + + boolean wasAnyoneInterested = _stateManager.methodReceived(evt); + + if (!_frameListeners.isEmpty()) + { + for (AMQMethodListener listener : _frameListeners) + { + wasAnyoneInterested = listener.methodReceived(evt) || wasAnyoneInterested; + } + } + + if (!wasAnyoneInterested) + { + throw new AMQNoMethodHandlerException(evt); + } + } + catch (AMQChannelException e) + { + if (getChannel(channelId) != null) + { + if (_logger.isInfoEnabled()) + { + _logger.info("Closing channel due to: " + e.getMessage()); + } + + writeFrame(e.getCloseFrame(channelId)); + closeChannel(channelId); + } + else + { + if (_logger.isDebugEnabled()) + { + _logger.debug("ChannelException occured on non-existent channel:" + e.getMessage()); + } + + if (_logger.isInfoEnabled()) + { + _logger.info("Closing connection due to: " + e.getMessage()); + } + + AMQConnectionException ce = + evt.getMethod().getConnectionException(AMQConstant.CHANNEL_ERROR, + AMQConstant.CHANNEL_ERROR.getName().toString()); + + closeConnection(channelId, ce, false); + } + } + catch (AMQConnectionException e) + { + closeConnection(channelId, e, false); + } + } + catch (Exception e) + { + + for (AMQMethodListener listener : _frameListeners) + { + listener.error(e); + } + + _logger.error("Unexpected exception while processing frame. Closing connection.", e); + + closeProtocolSession(); + } + } + + public void contentHeaderReceived(int channelId, ContentHeaderBody body) throws AMQException + { + + AMQChannel channel = getAndAssertChannel(channelId); + + channel.publishContentHeader(body); + + } + + public void contentBodyReceived(int channelId, ContentBody body) throws AMQException + { + AMQChannel channel = getAndAssertChannel(channelId); + + channel.publishContentBody(body); + } + + public void heartbeatBodyReceived(int channelId, HeartbeatBody body) + { + // NO - OP + } + + /** + * Convenience method that writes a frame to the protocol session. Equivalent to calling + * getProtocolSession().write(). + * + * @param frame the frame to write + */ + public void writeFrame(AMQDataBlock frame) + { + _lastSent = frame; + final ByteBuffer buf = frame.toNioByteBuffer(); + _lastIoTime = System.currentTimeMillis(); + _writtenBytes += buf.remaining(); + Job.fireAsynchEvent(_poolReference.getPool(), _writeJob, new Runnable() + { + @Override + public void run() + { + _networkDriver.send(buf); + } + }); + } + + public AMQShortString getContextKey() + { + return _contextKey; + } + + public void setContextKey(AMQShortString contextKey) + { + _contextKey = contextKey; + } + + public List getChannels() + { + return new ArrayList(_channelMap.values()); + } + + public AMQChannel getAndAssertChannel(int channelId) throws AMQException + { + AMQChannel channel = getChannel(channelId); + if (channel == null) + { + throw new AMQException(AMQConstant.NOT_FOUND, "Channel not found with id:" + channelId); + } + + return channel; + } + + public AMQChannel getChannel(int channelId) throws AMQException + { + final AMQChannel channel = + ((channelId & CHANNEL_CACHE_SIZE) == channelId) ? _cachedChannels[channelId] : _channelMap.get(channelId); + if ((channel == null) || channel.isClosing()) + { + return null; + } + else + { + return channel; + } + } + + public boolean channelAwaitingClosure(int channelId) + { + return !_closingChannelsList.isEmpty() && _closingChannelsList.containsKey(channelId); + } + + public void addChannel(AMQChannel channel) throws AMQException + { + if (_closed) + { + throw new AMQException("Session is closed"); + } + + final int channelId = channel.getChannelId(); + + if (_closingChannelsList.containsKey(channelId)) + { + throw new AMQException("Session is marked awaiting channel close"); + } + + if (_channelMap.size() == _maxNoOfChannels) + { + String errorMessage = + toString() + ": maximum number of channels has been reached (" + _maxNoOfChannels + + "); can't create channel"; + _logger.error(errorMessage); + throw new AMQException(AMQConstant.NOT_ALLOWED, errorMessage); + } + else + { + _channelMap.put(channel.getChannelId(), channel); + } + + if (((channelId & CHANNEL_CACHE_SIZE) == channelId)) + { + _cachedChannels[channelId] = channel; + } + + checkForNotification(); + } + + private void checkForNotification() + { + int channelsCount = _channelMap.size(); + if (channelsCount >= _maxNoOfChannels) + { + _managedObject.notifyClients("Channel count (" + channelsCount + ") has reached the threshold value"); + } + } + + public Long getMaximumNumberOfChannels() + { + return _maxNoOfChannels; + } + + public void setMaximumNumberOfChannels(Long value) + { + _maxNoOfChannels = value; + } + + public void commitTransactions(AMQChannel channel) throws AMQException + { + if ((channel != null) && channel.isTransactional()) + { + channel.commit(); + } + } + + public void rollbackTransactions(AMQChannel channel) throws AMQException + { + if ((channel != null) && channel.isTransactional()) + { + channel.rollback(); + } + } + + /** + * Close a specific channel. This will remove any resources used by the channel, including:
  • any queue + * subscriptions (this may in turn remove queues if they are auto delete
+ * + * @param channelId id of the channel to close + * + * @throws AMQException if an error occurs closing the channel + * @throws IllegalArgumentException if the channel id is not valid + */ + public void closeChannel(int channelId) throws AMQException + { + final AMQChannel channel = getChannel(channelId); + if (channel == null) + { + throw new IllegalArgumentException("Unknown channel id"); + } + else + { + try + { + channel.close(); + markChannelAwaitingCloseOk(channelId); + } + finally + { + removeChannel(channelId); + } + } + } + + public void closeChannelOk(int channelId) + { + // todo QPID-847 - This is called from two lcoations ChannelCloseHandler and ChannelCloseOkHandler. + // When it is the CC_OK_Handler then it makes sence to remove the channel else we will leak memory. + // We do it from the Close Handler as we are sending the OK back to the client. + // While this is AMQP spec compliant. The Java client in the event of an IllegalArgumentException + // will send a close-ok.. Where we should call removeChannel. + // However, due to the poor exception handling on the client. The client-user will be notified of the + // InvalidArgument and if they then decide to close the session/connection then the there will be time + // for that to occur i.e. a new close method be sent before the exeption handling can mark the session closed. + //removeChannel(channelId); + _closingChannelsList.remove(new Integer(channelId)); + } + + private void markChannelAwaitingCloseOk(int channelId) + { + _closingChannelsList.put(channelId, System.currentTimeMillis()); + } + + /** + * In our current implementation this is used by the clustering code. + * + * @param channelId The channel to remove + */ + public void removeChannel(int channelId) + { + _channelMap.remove(channelId); + if ((channelId & CHANNEL_CACHE_SIZE) == channelId) + { + _cachedChannels[channelId] = null; + } + } + + /** + * Initialise heartbeats on the session. + * + * @param delay delay in seconds (not ms) + */ + public void initHeartbeats(int delay) + { + if (delay > 0) + { + _networkDriver.setMaxWriteIdle(delay); + _networkDriver.setMaxReadIdle((int) (ApplicationRegistry.getInstance().getConfiguration().getHeartBeatTimeout() * delay)); + } + } + + /** + * Closes all channels that were opened by this protocol session. This frees up all resources used by the channel. + * + * @throws AMQException if an error occurs while closing any channel + */ + private void closeAllChannels() throws AMQException + { + for (AMQChannel channel : _channelMap.values()) + { + channel.close(); + } + + _channelMap.clear(); + for (int i = 0; i <= CHANNEL_CACHE_SIZE; i++) + { + _cachedChannels[i] = null; + } + } + + /** This must be called when the session is _closed in order to free up any resources managed by the session. */ + public void closeSession() throws AMQException + { + if (CurrentActor.get() == null) + { + CurrentActor.set(_actor); + } + if (!_closed) + { + if (_virtualHost != null) + { + _virtualHost.getConnectionRegistry().deregisterConnection(this); + } + + closeAllChannels(); + if (_managedObject != null) + { + _managedObject.unregister(); + } + + for (Task task : _taskList) + { + task.doTask(this); + } + + _closed = true; + _poolReference.releaseExecutorService(); + CurrentActor.get().message(_logSubject, ConnectionMessages.CON_1002()); + } + } + + public void closeConnection(int channelId, AMQConnectionException e, boolean closeProtocolSession) throws AMQException + { + if (_logger.isInfoEnabled()) + { + _logger.info("Closing connection due to: " + e.getMessage()); + } + + markChannelAwaitingCloseOk(channelId); + closeSession(); + _stateManager.changeState(AMQState.CONNECTION_CLOSING); + writeFrame(e.getCloseFrame(channelId)); + + if (closeProtocolSession) + { + closeProtocolSession(); + } + } + + public void closeProtocolSession() + { + _networkDriver.close(); + try + { + _stateManager.changeState(AMQState.CONNECTION_CLOSED); + } + catch (AMQException e) + { + _logger.info(e.getMessage()); + } + } + + public String toString() + { + return getRemoteAddress() + "(" + (getAuthorizedID() == null ? "?" : getAuthorizedID().getName() + ")"); + } + + public String dump() + { + return this + " last_sent=" + _lastSent + " last_received=" + _lastReceived; + } + + /** @return an object that can be used to identity */ + public Object getKey() + { + return getRemoteAddress(); + } + + /** + * Get the fully qualified domain name of the local address to which this session is bound. Since some servers may + * be bound to multiple addresses this could vary depending on the acceptor this session was created from. + * + * @return a String FQDN + */ + public String getLocalFQDN() + { + SocketAddress address = _networkDriver.getLocalAddress(); + // we use the vmpipe address in some tests hence the need for this rather ugly test. The host + // information is used by SASL primary. + if (address instanceof InetSocketAddress) + { + return ((InetSocketAddress) address).getHostName(); + } + else if (address instanceof VmPipeAddress) + { + return "vmpipe:" + ((VmPipeAddress) address).getPort(); + } + else + { + throw new IllegalArgumentException("Unsupported socket address class: " + address); + } + } + + public SaslServer getSaslServer() + { + return _saslServer; + } + + public void setSaslServer(SaslServer saslServer) + { + _saslServer = saslServer; + } + + public FieldTable getClientProperties() + { + return _clientProperties; + } + + public void setClientProperties(FieldTable clientProperties) + { + _clientProperties = clientProperties; + if (_clientProperties != null) + { + if (_clientProperties.getString(CLIENT_PROPERTIES_INSTANCE) != null) + { + String clientID = _clientProperties.getString(CLIENT_PROPERTIES_INSTANCE); + setContextKey(new AMQShortString(clientID)); + + // Log the Opening of the connection for this client + _actor.message(ConnectionMessages.CON_1001(clientID, _protocolVersion.toString(), true, true)); + } + + if (_clientProperties.getString(ClientProperties.version.toString()) != null) + { + _clientVersion = new AMQShortString(_clientProperties.getString(ClientProperties.version.toString())); + } + } + _sessionIdentifier = new ProtocolSessionIdentifier(this); + } + + private void setProtocolVersion(ProtocolVersion pv) + { + _protocolVersion = pv; + + _protocolOutputConverter = ProtocolOutputConverterRegistry.getConverter(this); + _dispatcher = ServerMethodDispatcherImpl.createMethodDispatcher(_stateManager, _protocolVersion); + } + + public byte getProtocolMajorVersion() + { + return _protocolVersion.getMajorVersion(); + } + + public ProtocolVersion getProtocolVersion() + { + return _protocolVersion; + } + + public byte getProtocolMinorVersion() + { + return _protocolVersion.getMinorVersion(); + } + + public boolean isProtocolVersion(byte major, byte minor) + { + return (getProtocolMajorVersion() == major) && (getProtocolMinorVersion() == minor); + } + + public MethodRegistry getRegistry() + { + return getMethodRegistry(); + } + + public Object getClientIdentifier() + { + return (_networkDriver != null) ? _networkDriver.getRemoteAddress() : null; + } + + public VirtualHost getVirtualHost() + { + return _virtualHost; + } + + public void setVirtualHost(VirtualHost virtualHost) throws AMQException + { + _virtualHost = virtualHost; + + _actor.virtualHostSelected(this); + _logSubject = new ConnectionLogSubject(this); + + _virtualHost.getConnectionRegistry().registerConnection(this); + + _managedObject = createMBean(); + _managedObject.register(); + } + + public void addSessionCloseTask(Task task) + { + _taskList.add(task); + } + + public void removeSessionCloseTask(Task task) + { + _taskList.remove(task); + } + + public ProtocolOutputConverter getProtocolOutputConverter() + { + return _protocolOutputConverter; + } + + public void setAuthorizedID(Principal authorizedID) + { + _authorizedID = authorizedID; + + // Let the actor know that this connection is now Authorized + _actor.connectionAuthorized(this); + } + + public Principal getAuthorizedID() + { + return _authorizedID; + } + + public Principal getPrincipal() + { + return _authorizedID; + } + + public SocketAddress getRemoteAddress() + { + return _networkDriver.getRemoteAddress(); + } + + public SocketAddress getLocalAddress() + { + return _networkDriver.getLocalAddress(); + } + + public MethodRegistry getMethodRegistry() + { + return MethodRegistry.getMethodRegistry(getProtocolVersion()); + } + + public MethodDispatcher getMethodDispatcher() + { + return _dispatcher; + } + + @Override + public void closed() + { + try + { + closeSession(); + } + catch (AMQException e) + { + _logger.error("Could not close protocol engine", e); + } + } + + @Override + public void readerIdle() + { + // Nothing + } + + @Override + public void setNetworkDriver(NetworkDriver driver) + { + _networkDriver = driver; + } + + @Override + public void writerIdle() + { + _networkDriver.send(HeartbeatBody.FRAME.toNioByteBuffer()); + } + + @Override + public void exception(Throwable throwable) + { + if (throwable instanceof AMQProtocolHeaderException) + { + + writeFrame(new ProtocolInitiation(ProtocolVersion.getLatestSupportedVersion())); + _networkDriver.close(); + + _logger.error("Error in protocol initiation " + this + ":" + getRemoteAddress() + " :" + throwable.getMessage(), throwable); + } + else if (throwable instanceof IOException) + { + _logger.error("IOException caught in" + this + ", session closed implictly: " + throwable); + } + else + { + _logger.error("Exception caught in" + this + ", closing session explictly: " + throwable, throwable); + + + MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(getProtocolVersion()); + ConnectionCloseBody closeBody = methodRegistry.createConnectionCloseBody(200,new AMQShortString(throwable.getMessage()),0,0); + + writeFrame(closeBody.generateFrame(0)); + + _networkDriver.close(); + } + } + + @Override + public void init() + { + // Do nothing + } + + @Override + public void setSender(Sender sender) + { + // Do nothing + } + + @Override + public long getReadBytes() + { + return _readBytes; + } + + public long getWrittenBytes() + { + return _writtenBytes; + } + + public long getLastIoTime() + { + return _lastIoTime; + } + + public ProtocolSessionIdentifier getSessionIdentifier() + { + return _sessionIdentifier; + } + + public String getClientVersion() + { + return (_clientVersion == null) ? null : _clientVersion.toString(); + } + + @Override + public void closeIfLingeringClosedChannels() + { + for (Entryid : _closingChannelsList.entrySet()) + { + if (id.getValue() + 30000 > System.currentTimeMillis()) + { + // We have a channel that we closed 30 seconds ago. Client's dead, kill the connection + _logger.error("Closing connection as channel was closed more than 30 seconds ago and no ChannelCloseOk has been processed"); + closeProtocolSession(); + } + } + } + +} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngineFactory.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngineFactory.java new file mode 100644 index 0000000000..ff0c007a60 --- /dev/null +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngineFactory.java @@ -0,0 +1,29 @@ +package org.apache.qpid.server.protocol; + +import org.apache.qpid.protocol.ProtocolEngine; +import org.apache.qpid.protocol.ProtocolEngineFactory; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.server.virtualhost.VirtualHostRegistry; +import org.apache.qpid.transport.NetworkDriver; + +public class AMQProtocolEngineFactory implements ProtocolEngineFactory +{ + private VirtualHostRegistry _vhosts; + + public AMQProtocolEngineFactory() + { + this(1); + } + + public AMQProtocolEngineFactory(Integer port) + { + _vhosts = ApplicationRegistry.getInstance(port).getVirtualHostRegistry(); + } + + + public ProtocolEngine newProtocolEngine(NetworkDriver networkDriver) + { + return new AMQProtocolEngine(_vhosts, networkDriver); + } + +} diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java index 21fd1ce6b4..aecf7a6839 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java @@ -35,6 +35,7 @@ import org.apache.qpid.server.output.ProtocolOutputConverter; import org.apache.qpid.server.virtualhost.VirtualHost; import java.security.Principal; +import java.util.List; public interface AMQProtocolSession extends AMQVersionAwareProtocolSession, PrincipalHolder @@ -208,5 +209,21 @@ public interface AMQProtocolSession extends AMQVersionAwareProtocolSession, Prin public MethodDispatcher getMethodDispatcher(); public ProtocolSessionIdentifier getSessionIdentifier(); + + String getClientVersion(); + + long getLastIoTime(); + + long getWrittenBytes(); + + Long getMaximumNumberOfChannels(); + + void setMaximumNumberOfChannels(Long value); + + void commitTransactions(AMQChannel channel) throws AMQException; + + List getChannels(); + + void closeIfLingeringClosedChannels(); } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java index 5415f69dc1..393d4ab4cb 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java @@ -79,7 +79,7 @@ import org.apache.qpid.server.management.ManagedObject; @MBeanDescription("Management Bean for an AMQ Broker Connection") public class AMQProtocolSessionMBean extends AMQManagedObject implements ManagedConnection { - private AMQMinaProtocolSession _session = null; + private AMQProtocolSession _protocolSession = null; private String _name = null; // openmbean data types for representing the channel attributes @@ -92,10 +92,10 @@ public class AMQProtocolSessionMBean extends AMQManagedObject implements Managed new AMQShortString("Broker Management Console has closed the connection."); @MBeanConstructor("Creates an MBean exposing an AMQ Broker Connection") - public AMQProtocolSessionMBean(AMQMinaProtocolSession session) throws NotCompliantMBeanException, OpenDataException + public AMQProtocolSessionMBean(AMQProtocolSession amqProtocolSession) throws NotCompliantMBeanException, OpenDataException { super(ManagedConnection.class, ManagedConnection.TYPE, ManagedConnection.VERSION); - _session = session; + _protocolSession = amqProtocolSession; String remote = getRemoteAddress(); remote = "anonymous".equals(remote) ? (remote + hashCode()) : remote; _name = jmxEncode(new StringBuffer(remote), 0).toString(); @@ -128,52 +128,52 @@ public class AMQProtocolSessionMBean extends AMQManagedObject implements Managed public String getClientId() { - return (_session.getContextKey() == null) ? null : _session.getContextKey().toString(); + return (_protocolSession.getContextKey() == null) ? null : _protocolSession.getContextKey().toString(); } public String getAuthorizedId() { - return (_session.getPrincipal() != null ) ? _session.getPrincipal().getName() : null; + return (_protocolSession.getPrincipal() != null ) ? _protocolSession.getPrincipal().getName() : null; } public String getVersion() { - return (_session.getClientVersion() == null) ? null : _session.getClientVersion().toString(); + return (_protocolSession.getClientVersion() == null) ? null : _protocolSession.getClientVersion().toString(); } public Date getLastIoTime() { - return new Date(_session.getIOSession().getLastIoTime()); + return new Date(_protocolSession.getLastIoTime()); } public String getRemoteAddress() { - return _session.getIOSession().getRemoteAddress().toString(); + return _protocolSession.getRemoteAddress().toString(); } public ManagedObject getParentObject() { - return _session.getVirtualHost().getManagedObject(); + return _protocolSession.getVirtualHost().getManagedObject(); } public Long getWrittenBytes() { - return _session.getIOSession().getWrittenBytes(); + return _protocolSession.getWrittenBytes(); } public Long getReadBytes() { - return _session.getIOSession().getReadBytes(); + return _protocolSession.getWrittenBytes(); } public Long getMaximumNumberOfChannels() { - return _session.getMaximumNumberOfChannels(); + return _protocolSession.getMaximumNumberOfChannels(); } public void setMaximumNumberOfChannels(Long value) { - _session.setMaximumNumberOfChannels(value); + _protocolSession.setMaximumNumberOfChannels(value); } public String getObjectInstanceName() @@ -192,13 +192,13 @@ public class AMQProtocolSessionMBean extends AMQManagedObject implements Managed CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger())); try { - AMQChannel channel = _session.getChannel(channelId); + AMQChannel channel = _protocolSession.getChannel(channelId); if (channel == null) { throw new JMException("The channel (channel Id = " + channelId + ") does not exist"); } - _session.commitTransactions(channel); + _protocolSession.commitTransactions(channel); } catch (AMQException ex) { @@ -221,13 +221,13 @@ public class AMQProtocolSessionMBean extends AMQManagedObject implements Managed CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger())); try { - AMQChannel channel = _session.getChannel(channelId); + AMQChannel channel = _protocolSession.getChannel(channelId); if (channel == null) { throw new JMException("The channel (channel Id = " + channelId + ") does not exist"); } - _session.rollbackTransactions(channel); + _protocolSession.commitTransactions(channel); } catch (AMQException ex) { @@ -248,7 +248,7 @@ public class AMQProtocolSessionMBean extends AMQManagedObject implements Managed public TabularData channels() throws OpenDataException { TabularDataSupport channelsList = new TabularDataSupport(_channelsType); - List list = _session.getChannels(); + List list = _protocolSession.getChannels(); for (AMQChannel channel : list) { @@ -274,7 +274,7 @@ public class AMQProtocolSessionMBean extends AMQManagedObject implements Managed public void closeConnection() throws JMException { - MethodRegistry methodRegistry = _session.getMethodRegistry(); + MethodRegistry methodRegistry = _protocolSession.getMethodRegistry(); ConnectionCloseBody responseBody = methodRegistry.createConnectionCloseBody(AMQConstant.REPLY_SUCCESS.getCode(), // replyCode @@ -301,12 +301,12 @@ public class AMQProtocolSessionMBean extends AMQManagedObject implements Managed try { - _session.writeFrame(responseBody.generateFrame(0)); + _protocolSession.writeFrame(responseBody.generateFrame(0)); try { - _session.closeSession(); + _protocolSession.closeSession(); } catch (AMQException ex) { diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java index 7511b5c818..1affdd6590 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java @@ -258,7 +258,7 @@ public abstract class ApplicationRegistry implements IApplicationRegistry for (InetSocketAddress bindAddress : _acceptors.keySet()) { QpidAcceptor acceptor = _acceptors.get(bindAddress); - acceptor.getIoAcceptor().unbind(bindAddress); + acceptor.getNetworkDriver().close(); CurrentActor.get().message(BrokerMessages.BRK_1003(acceptor.toString(), bindAddress.getPort())); } } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/network/FirewallPlugin.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/network/FirewallPlugin.java index 438bd696c2..393e792275 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/network/FirewallPlugin.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/network/FirewallPlugin.java @@ -23,7 +23,6 @@ package org.apache.qpid.server.security.access.plugins.network; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.SocketAddress; -import java.util.Iterator; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; import java.util.regex.Pattern; @@ -32,7 +31,7 @@ import org.apache.commons.configuration.CompositeConfiguration; import org.apache.commons.configuration.Configuration; import org.apache.commons.configuration.ConfigurationException; import org.apache.commons.configuration.XMLConfiguration; -import org.apache.qpid.server.protocol.AMQMinaProtocolSession; +import org.apache.qpid.protocol.ProtocolEngine; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.security.access.ACLPlugin; import org.apache.qpid.server.security.access.ACLPluginFactory; @@ -179,15 +178,22 @@ public class FirewallPlugin extends AbstractACLPlugin private FirewallRule[] _rules; @Override - public AuthzResult authoriseConnect(PrincipalHolder session, VirtualHost virtualHost) + public AuthzResult authoriseConnect(PrincipalHolder principalHolder, VirtualHost virtualHost) { - if (!(session instanceof AMQMinaProtocolSession)) + if(!(principalHolder instanceof ProtocolEngine)) { - return AuthzResult.ABSTAIN; // We only deal with tcp sessions, which - // mean MINA right now + return AuthzResult.ABSTAIN; // We only deal with tcp sessions + } + + ProtocolEngine session = (ProtocolEngine) principalHolder; + + SocketAddress sockAddr = session.getRemoteAddress(); + if (!(sockAddr instanceof InetSocketAddress)) + { + return AuthzResult.ABSTAIN; // We only deal with tcp sessions } - InetAddress addr = getInetAdressFromMinaSession((AMQMinaProtocolSession) session); + InetAddress addr = ((InetSocketAddress) sockAddr).getAddress(); if (addr == null) { @@ -214,19 +220,6 @@ public class FirewallPlugin extends AbstractACLPlugin } - private InetAddress getInetAdressFromMinaSession(AMQMinaProtocolSession session) - { - SocketAddress remote = session.getIOSession().getRemoteAddress(); - if (remote instanceof InetSocketAddress) - { - return ((InetSocketAddress) remote).getAddress(); - } - else - { - return null; - } - } - public void setConfiguration(Configuration config) throws ConfigurationException { // Get default action diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/QpidAcceptor.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/QpidAcceptor.java index 61cc7cdeb6..3ca22b60c8 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/QpidAcceptor.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/QpidAcceptor.java @@ -20,21 +20,21 @@ */ package org.apache.qpid.server.transport; -import org.apache.mina.common.IoAcceptor; +import org.apache.qpid.transport.NetworkDriver; public class QpidAcceptor { - IoAcceptor _acceptor; + NetworkDriver _driver; String _protocol; - public QpidAcceptor(IoAcceptor acceptor, String protocol) + public QpidAcceptor(NetworkDriver driver, String protocol) { - _acceptor = acceptor; + _driver = driver; _protocol = protocol; } - public IoAcceptor getIoAcceptor() + public NetworkDriver getNetworkDriver() { - return _acceptor; + return _driver; } public String toString() diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java index 9ab3592628..3b776a62b4 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java @@ -279,6 +279,14 @@ public class VirtualHost implements Accessable _houseKeepingTimer.scheduleAtFixedRate(new RemoveExpiredMessagesTask(), period / 2, period); + + class ForceChannelClosuresTask extends TimerTask + { + public void run() + { + _connectionRegistry.expireClosedChannels(); + } + } } } diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java index ebfa80d139..93e7e756e6 100644 --- a/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java +++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/ServerConfigurationTest.java @@ -20,26 +20,26 @@ */ package org.apache.qpid.server.configuration; +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.util.Iterator; +import java.util.List; +import java.util.Locale; + import junit.framework.TestCase; + import org.apache.commons.configuration.Configuration; import org.apache.commons.configuration.ConfigurationException; import org.apache.commons.configuration.XMLConfiguration; -import org.apache.qpid.codec.AMQCodecFactory; -import org.apache.qpid.server.protocol.AMQMinaProtocolSession; +import org.apache.qpid.server.protocol.AMQProtocolEngine; import org.apache.qpid.server.protocol.AMQProtocolSession; -import org.apache.qpid.server.protocol.TestIoSession; import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry; import org.apache.qpid.server.virtualhost.VirtualHost; import org.apache.qpid.server.virtualhost.VirtualHostRegistry; - -import java.io.File; -import java.io.FileWriter; -import java.io.IOException; -import java.io.RandomAccessFile; -import java.util.Iterator; -import java.util.List; -import java.util.Locale; +import org.apache.qpid.transport.TestNetworkDriver; public class ServerConfigurationTest extends TestCase { @@ -589,12 +589,12 @@ public class ServerConfigurationTest extends TestCase { // Check default ServerConfiguration serverConfig = new ServerConfiguration(_config); - assertEquals(true, serverConfig.getSSLOnly()); + assertEquals(false, serverConfig.getSSLOnly()); // Check value we set - _config.setProperty("connector.ssl.sslOnly", false); + _config.setProperty("connector.ssl.sslOnly", true); serverConfig = new ServerConfiguration(_config); - assertEquals(false, serverConfig.getSSLOnly()); + assertEquals(true, serverConfig.getSSLOnly()); } public void testGetSSLPort() throws ConfigurationException @@ -791,16 +791,15 @@ public class ServerConfigurationTest extends TestCase // Test config VirtualHostRegistry virtualHostRegistry = reg.getVirtualHostRegistry(); VirtualHost virtualHost = virtualHostRegistry.getVirtualHost("test"); - AMQCodecFactory codecFactory = new AMQCodecFactory(true); - TestIoSession iosession = new TestIoSession(); - iosession.setAddress("127.0.0.1"); - - AMQProtocolSession session = new AMQMinaProtocolSession(iosession, virtualHostRegistry, codecFactory); + TestNetworkDriver testDriver = new TestNetworkDriver(); + testDriver.setRemoteAddress("127.0.0.1"); + + AMQProtocolEngine session = new AMQProtocolEngine(virtualHostRegistry, testDriver); assertFalse(reg.getAccessManager().authoriseConnect(session, virtualHost)); - iosession.setAddress("127.1.2.3"); - session = new AMQMinaProtocolSession(iosession, virtualHostRegistry, codecFactory); + testDriver.setRemoteAddress("127.1.2.3"); + session = new AMQProtocolEngine(virtualHostRegistry, testDriver); assertTrue(reg.getAccessManager().authoriseConnect(session, virtualHost)); } @@ -866,12 +865,12 @@ public class ServerConfigurationTest extends TestCase // Test config VirtualHostRegistry virtualHostRegistry = reg.getVirtualHostRegistry(); VirtualHost virtualHost = virtualHostRegistry.getVirtualHost("test"); - AMQCodecFactory codecFactory = new AMQCodecFactory(true); - TestIoSession iosession = new TestIoSession(); - iosession.setAddress("127.0.0.1"); + TestNetworkDriver testDriver = new TestNetworkDriver(); + testDriver.setRemoteAddress("127.0.0.1"); - AMQProtocolSession session = new AMQMinaProtocolSession(iosession, virtualHostRegistry, codecFactory); + AMQProtocolEngine session = new AMQProtocolEngine(virtualHostRegistry, testDriver); + session.setNetworkDriver(testDriver); assertFalse(reg.getAccessManager().authoriseConnect(session, virtualHost)); } @@ -935,12 +934,11 @@ public class ServerConfigurationTest extends TestCase ApplicationRegistry.initialise(reg, 1); // Test config - TestIoSession iosession = new TestIoSession(); - iosession.setAddress("127.0.0.1"); + TestNetworkDriver testDriver = new TestNetworkDriver(); + testDriver.setRemoteAddress("127.0.0.1"); VirtualHostRegistry virtualHostRegistry = reg.getVirtualHostRegistry(); VirtualHost virtualHost = virtualHostRegistry.getVirtualHost("test"); - AMQCodecFactory codecFactory = new AMQCodecFactory(true); - AMQProtocolSession session = new AMQMinaProtocolSession(iosession, virtualHostRegistry, codecFactory); + AMQProtocolSession session = new AMQProtocolEngine(virtualHostRegistry, testDriver); assertFalse(reg.getAccessManager().authoriseConnect(session, virtualHost)); RandomAccessFile fileBRandom = new RandomAccessFile(fileB, "rw"); diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java index e199255f50..bc36c61382 100644 --- a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java +++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java @@ -44,7 +44,7 @@ public class AMQProtocolSessionMBeanTest extends TestCase private static final Logger log = Logger.getLogger(AMQProtocolSessionMBeanTest.class); private MessageStore _messageStore = new SkeletonMessageStore(); - private AMQMinaProtocolSession _protocolSession; + private AMQProtocolEngine _protocolSession; private AMQChannel _channel; private AMQProtocolSessionMBean _mbean; diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java index 37dfead2e5..ec7bf1cb72 100644 --- a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java +++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java @@ -20,23 +20,23 @@ */ package org.apache.qpid.server.protocol; -import org.apache.qpid.AMQException; -import org.apache.qpid.codec.AMQCodecFactory; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.server.output.ProtocolOutputConverter; -import org.apache.qpid.server.queue.AMQMessage; -import org.apache.qpid.server.registry.ApplicationRegistry; -import org.apache.qpid.server.virtualhost.VirtualHost; - +import java.security.Principal; import java.util.ArrayList; import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; -import java.security.Principal; -public class InternalTestProtocolSession extends AMQMinaProtocolSession implements ProtocolOutputConverter +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.server.output.ProtocolOutputConverter; +import org.apache.qpid.server.queue.AMQMessage; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.server.virtualhost.VirtualHost; +import org.apache.qpid.transport.TestNetworkDriver; + +public class InternalTestProtocolSession extends AMQProtocolEngine implements ProtocolOutputConverter { // ChannelID(LIST) -> LinkedList final Map>> _channelDelivers; @@ -44,9 +44,7 @@ public class InternalTestProtocolSession extends AMQMinaProtocolSession implemen public InternalTestProtocolSession(VirtualHost virtualHost) throws AMQException { - super(new TestIoSession(), - ApplicationRegistry.getInstance().getVirtualHostRegistry(), - new AMQCodecFactory(true)); + super(ApplicationRegistry.getInstance().getVirtualHostRegistry(), new TestNetworkDriver()); _channelDelivers = new HashMap>>(); diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/MaxChannelsTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/MaxChannelsTest.java index 8597fc5863..e37492bcb0 100644 --- a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/MaxChannelsTest.java +++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/MaxChannelsTest.java @@ -37,7 +37,7 @@ import java.security.Principal; /** Test class to test MBean operations for AMQMinaProtocolSession. */ public class MaxChannelsTest extends TestCase { - private AMQMinaProtocolSession _session; + private AMQProtocolEngine _session; public void testChannels() throws Exception { diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/TestIoSession.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/TestIoSession.java deleted file mode 100644 index 211f491867..0000000000 --- a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/TestIoSession.java +++ /dev/null @@ -1,328 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ -package org.apache.qpid.server.protocol; - -import java.net.InetSocketAddress; -import java.net.SocketAddress; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; - -import org.apache.mina.common.CloseFuture; -import org.apache.mina.common.IdleStatus; -import org.apache.mina.common.IoFilterChain; -import org.apache.mina.common.IoHandler; -import org.apache.mina.common.IoService; -import org.apache.mina.common.IoServiceConfig; -import org.apache.mina.common.IoSession; -import org.apache.mina.common.IoSessionConfig; -import org.apache.mina.common.ThreadModel; -import org.apache.mina.common.TrafficMask; -import org.apache.mina.common.TransportType; -import org.apache.mina.common.WriteFuture; -import org.apache.mina.transport.socket.nio.SocketAcceptorConfig; -import org.apache.qpid.pool.ReadWriteThreadModel; - -/** - * Test implementation of IoSession, which is required for some tests. Methods not being used are not implemented, - * so if this class is being used and some methods are to be used, then please update those. - */ -public class TestIoSession implements IoSession -{ - private final ConcurrentMap attributes = new ConcurrentHashMap(); - private String _address = "127.0.0.1"; - private int _port = 1; - - public TestIoSession() - { - } - - public IoService getService() - { - return null; - } - - public IoServiceConfig getServiceConfig() - { - return new TestIoConfig(); - } - - public IoHandler getHandler() - { - return null; - } - - public IoSessionConfig getConfig() - { - return null; - } - - public IoFilterChain getFilterChain() - { - return null; - } - - public WriteFuture write(Object message) - { - return null; - } - - public CloseFuture close() - { - return null; - } - - public Object getAttachment() - { - return getAttribute(""); - } - - public Object setAttachment(Object attachment) - { - return setAttribute("",attachment); - } - - public Object getAttribute(String key) - { - return attributes.get(key); - } - - public Object setAttribute(String key, Object value) - { - return attributes.put(key,value); - } - - public Object setAttribute(String key) - { - return attributes.put(key, Boolean.TRUE); - } - - public Object removeAttribute(String key) - { - return attributes.remove(key); - } - - public boolean containsAttribute(String key) - { - return attributes.containsKey(key); - } - - public Set getAttributeKeys() - { - return attributes.keySet(); - } - - public TransportType getTransportType() - { - return null; - } - - public boolean isConnected() - { - return false; - } - - public boolean isClosing() - { - return false; - } - - public CloseFuture getCloseFuture() - { - return null; - } - - public SocketAddress getRemoteAddress() - { - return new InetSocketAddress(getAddress(), getPort()); - } - - public SocketAddress getLocalAddress() - { - return null; - } - - public SocketAddress getServiceAddress() - { - return null; - } - - public int getIdleTime(IdleStatus status) - { - return 0; - } - - public long getIdleTimeInMillis(IdleStatus status) - { - return 0; - } - - public void setIdleTime(IdleStatus status, int idleTime) - { - - } - - public int getWriteTimeout() - { - return 0; - } - - public long getWriteTimeoutInMillis() - { - return 0; - } - - public void setWriteTimeout(int writeTimeout) - { - - } - - public TrafficMask getTrafficMask() - { - return null; - } - - public void setTrafficMask(TrafficMask trafficMask) - { - - } - - public void suspendRead() - { - - } - - public void suspendWrite() - { - - } - - public void resumeRead() - { - - } - - public void resumeWrite() - { - - } - - public long getReadBytes() - { - return 0; - } - - public long getWrittenBytes() - { - return 0; - } - - public long getReadMessages() - { - return 0; - } - - public long getWrittenMessages() - { - return 0; - } - - public long getWrittenWriteRequests() - { - return 0; - } - - public int getScheduledWriteRequests() - { - return 0; - } - - public int getScheduledWriteBytes() - { - return 0; - } - - public long getCreationTime() - { - return 0; - } - - public long getLastIoTime() - { - return 0; - } - - public long getLastReadTime() - { - return 0; - } - - public long getLastWriteTime() - { - return 0; - } - - public boolean isIdle(IdleStatus status) - { - return false; - } - - public int getIdleCount(IdleStatus status) - { - return 0; - } - - public long getLastIdleTime(IdleStatus status) - { - return 0; - } - - public void setAddress(String string) - { - this._address = string; - } - - public String getAddress() - { - return _address; - } - - public void setPort(int _port) - { - this._port = _port; - } - - public int getPort() - { - return _port; - } - - /** - * Test implementation of IoServiceConfig - */ - private class TestIoConfig extends SocketAcceptorConfig - { - public ThreadModel getThreadModel() - { - return ReadWriteThreadModel.getInstance(); - } - } -} diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java index 8c6260ca9e..19470e6226 100644 --- a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java +++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java @@ -20,37 +20,34 @@ */ package org.apache.qpid.server.queue; +import java.util.ArrayList; +import java.util.LinkedList; + +import javax.management.Notification; + import junit.framework.TestCase; + +import org.apache.mina.common.ByteBuffer; import org.apache.qpid.AMQException; -import org.apache.qpid.server.store.MessageStore; -import org.apache.qpid.server.store.MemoryMessageStore; -import org.apache.qpid.server.store.StoreContext; -import org.apache.qpid.server.virtualhost.VirtualHost; -import org.apache.qpid.server.registry.IApplicationRegistry; -import org.apache.qpid.server.registry.ApplicationRegistry; -import org.apache.qpid.server.txn.TransactionalContext; -import org.apache.qpid.server.txn.NonTransactionalContext; -import org.apache.qpid.server.RequiredDeliveryException; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.ContentHeaderBody; +import org.apache.qpid.framing.abstraction.ContentChunk; +import org.apache.qpid.framing.abstraction.MessagePublishInfo; import org.apache.qpid.server.AMQChannel; +import org.apache.qpid.server.RequiredDeliveryException; import org.apache.qpid.server.logging.actors.CurrentActor; +import org.apache.qpid.server.protocol.AMQProtocolEngine; +import org.apache.qpid.server.protocol.InternalTestProtocolSession; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.server.registry.IApplicationRegistry; +import org.apache.qpid.server.store.MemoryMessageStore; +import org.apache.qpid.server.store.MessageStore; +import org.apache.qpid.server.store.StoreContext; import org.apache.qpid.server.subscription.Subscription; import org.apache.qpid.server.subscription.SubscriptionFactoryImpl; -import org.apache.qpid.server.protocol.AMQMinaProtocolSession; -import org.apache.qpid.server.protocol.InternalTestProtocolSession; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.abstraction.MessagePublishInfo; -import org.apache.qpid.framing.abstraction.ContentChunk; -import org.apache.commons.configuration.CompositeConfiguration; -import org.apache.mina.common.ByteBuffer; - -import javax.management.Notification; - -import java.util.ArrayList; -import java.util.LinkedList; -import java.util.Collections; -import java.util.Set; -import java.security.Principal; +import org.apache.qpid.server.txn.NonTransactionalContext; +import org.apache.qpid.server.txn.TransactionalContext; +import org.apache.qpid.server.virtualhost.VirtualHost; /** This class tests all the alerts an AMQQueue can throw based on threshold values of different parameters */ public class AMQQueueAlertTest extends TestCase @@ -62,7 +59,7 @@ public class AMQQueueAlertTest extends TestCase private AMQQueue _queue; private AMQQueueMBean _queueMBean; private VirtualHost _virtualHost; - private AMQMinaProtocolSession _protocolSession; + private AMQProtocolEngine _protocolSession; private MessageStore _messageStore = new MemoryMessageStore(); private StoreContext _storeContext = new StoreContext(); private TransactionalContext _transactionalContext = new NonTransactionalContext(_messageStore, _storeContext, diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/plugins/network/FirewallPluginTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/plugins/network/FirewallPluginTest.java index a497365b06..5d3335c001 100644 --- a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/plugins/network/FirewallPluginTest.java +++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/plugins/network/FirewallPluginTest.java @@ -30,17 +30,14 @@ import java.net.InetSocketAddress; import junit.framework.TestCase; import org.apache.commons.configuration.ConfigurationException; -import org.apache.commons.configuration.PropertiesConfiguration; import org.apache.commons.configuration.XMLConfiguration; -import org.apache.qpid.codec.AMQCodecFactory; -import org.apache.qpid.server.configuration.VirtualHostConfiguration; -import org.apache.qpid.server.protocol.AMQMinaProtocolSession; -import org.apache.qpid.server.protocol.TestIoSession; +import org.apache.qpid.server.protocol.AMQProtocolEngine; +import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.security.access.ACLPlugin.AuthzResult; import org.apache.qpid.server.store.TestableMemoryMessageStore; import org.apache.qpid.server.virtualhost.VirtualHost; import org.apache.qpid.server.virtualhost.VirtualHostRegistry; -import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.transport.TestNetworkDriver; public class FirewallPluginTest extends TestCase { @@ -84,22 +81,22 @@ public class FirewallPluginTest extends TestCase private TestableMemoryMessageStore _store; private VirtualHost _virtualHost; - private AMQMinaProtocolSession _session; + private AMQProtocolEngine _session; + private TestNetworkDriver _testDriver; @Override public void setUp() throws Exception { super.setUp(); _store = new TestableMemoryMessageStore(); - TestIoSession iosession = new TestIoSession(); - iosession.setAddress("127.0.0.1"); + _testDriver = new TestNetworkDriver(); + _testDriver.setRemoteAddress("127.0.0.1"); // Retreive VirtualHost from the Registry VirtualHostRegistry virtualHostRegistry = ApplicationRegistry.getInstance().getVirtualHostRegistry(); _virtualHost = virtualHostRegistry.getVirtualHost("test"); - AMQCodecFactory codecFactory = new AMQCodecFactory(true); - _session = new AMQMinaProtocolSession(iosession, virtualHostRegistry, codecFactory); + _session = new AMQProtocolEngine(virtualHostRegistry, _testDriver); } public void tearDown() throws Exception @@ -170,7 +167,7 @@ public class FirewallPluginTest extends TestCase assertEquals(AuthzResult.DENIED, plugin.authoriseConnect(_session, _virtualHost)); // Set session IP so that we're connected from the right address - ((TestIoSession) _session.getIOSession()).setAddress("192.168.23.23"); + _testDriver.setRemoteAddress("192.168.23.23"); assertEquals(AuthzResult.ALLOWED, plugin.authoriseConnect(_session, _virtualHost)); } @@ -185,7 +182,7 @@ public class FirewallPluginTest extends TestCase assertEquals(AuthzResult.DENIED, plugin.authoriseConnect(_session, _virtualHost)); // Set session IP so that we're connected from the right address - ((TestIoSession) _session.getIOSession()).setAddress("192.168.23.23"); + _testDriver.setRemoteAddress("192.168.23.23"); assertEquals(AuthzResult.ALLOWED, plugin.authoriseConnect(_session, _virtualHost)); } @@ -198,7 +195,7 @@ public class FirewallPluginTest extends TestCase FirewallPlugin plugin = initialisePlugin("deny", new RuleInfo[]{rule}); // Set session IP so that we're connected from the right address - ((TestIoSession) _session.getIOSession()).setAddress("127.0.0.1"); + _testDriver.setRemoteAddress("127.0.0.1"); assertEquals(AuthzResult.ALLOWED, plugin.authoriseConnect(_session, _virtualHost)); } @@ -211,7 +208,7 @@ public class FirewallPluginTest extends TestCase FirewallPlugin plugin = initialisePlugin("deny", new RuleInfo[]{rule}); // Set session IP so that we're connected from the right address - ((TestIoSession) _session.getIOSession()).setAddress("127.0.0.1"); + _testDriver.setRemoteAddress("127.0.0.1"); assertEquals(AuthzResult.ALLOWED, plugin.authoriseConnect(_session, _virtualHost)); } @@ -234,7 +231,7 @@ public class FirewallPluginTest extends TestCase assertEquals(AuthzResult.DENIED, plugin.authoriseConnect(_session, _virtualHost)); // Set session IP so that we're connected from the right address - ((TestIoSession) _session.getIOSession()).setAddress("192.168.23.23"); + _testDriver.setRemoteAddress("192.168.23.23"); assertEquals(AuthzResult.ALLOWED, plugin.authoriseConnect(_session, _virtualHost)); } @@ -257,7 +254,7 @@ public class FirewallPluginTest extends TestCase assertEquals(AuthzResult.DENIED, plugin.authoriseConnect(_session, _virtualHost)); // Set session IP so that we're connected from the right address - ((TestIoSession) _session.getIOSession()).setAddress("192.168.23.23"); + _testDriver.setRemoteAddress("192.168.23.23"); assertEquals(AuthzResult.ALLOWED, plugin.authoriseConnect(_session, _virtualHost)); } @@ -271,7 +268,7 @@ public class FirewallPluginTest extends TestCase assertEquals(AuthzResult.DENIED, plugin.authoriseConnect(_session, _virtualHost)); // Set session IP so that we're connected from the right address - ((TestIoSession) _session.getIOSession()).setAddress("192.168.23.23"); + _testDriver.setRemoteAddress("192.168.23.23"); assertEquals(AuthzResult.ALLOWED, plugin.authoriseConnect(_session, _virtualHost)); } @@ -285,7 +282,7 @@ public class FirewallPluginTest extends TestCase assertEquals(AuthzResult.DENIED, plugin.authoriseConnect(_session, _virtualHost)); // Set session IP so that we're connected from the right address - ((TestIoSession) _session.getIOSession()).setAddress("192.168.23.23"); + _testDriver.setRemoteAddress("192.168.23.23"); assertEquals(AuthzResult.ALLOWED, plugin.authoriseConnect(_session, _virtualHost)); } @@ -295,11 +292,11 @@ public class FirewallPluginTest extends TestCase firstRule.setAccess("allow"); firstRule.setHostname("foo, bar, "+new InetSocketAddress("127.0.0.1", 5672).getHostName()); FirewallPlugin plugin = initialisePlugin("deny", new RuleInfo[]{firstRule}); - ((TestIoSession) _session.getIOSession()).setAddress("10.0.0.1"); + _testDriver.setRemoteAddress("10.0.0.1"); assertEquals(AuthzResult.DENIED, plugin.authoriseConnect(_session, _virtualHost)); // Set session IP so that we're connected from the right address - ((TestIoSession) _session.getIOSession()).setAddress("127.0.0.1"); + _testDriver.setRemoteAddress("127.0.0.1"); assertEquals(AuthzResult.ALLOWED, plugin.authoriseConnect(_session, _virtualHost)); } diff --git a/qpid/java/client/build.xml b/qpid/java/client/build.xml index 321e613d94..3c6132dc5b 100644 --- a/qpid/java/client/build.xml +++ b/qpid/java/client/build.xml @@ -21,6 +21,7 @@ + diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java index a0b69b5493..9876393d4c 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java @@ -97,7 +97,7 @@ public class AMQConnectionDelegate_8_0 implements AMQConnectionDelegate { _conn.getProtocolHandler().createIoTransportSession(brokerDetail); } - + _conn._protocolHandler.getProtocolSession().init(); // this blocks until the connection has been set up or when an error // has prevented the connection being set up diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java index 927f660932..8223cd5394 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java @@ -20,8 +20,6 @@ */ package org.apache.qpid.client.failover; -import org.apache.mina.common.IoSession; - import org.apache.qpid.AMQDisconnectedException; import org.apache.qpid.client.protocol.AMQProtocolHandler; import org.apache.qpid.client.state.AMQStateManager; @@ -81,9 +79,6 @@ public class FailoverHandler implements Runnable /** Used for debugging. */ private static final Logger _logger = LoggerFactory.getLogger(FailoverHandler.class); - /** Holds the MINA session for the connection that has failed, not the connection that is being failed onto. */ - private final IoSession _session; - /** Holds the protocol handler for the failed connection, upon which the new connection is to be set up. */ private AMQProtocolHandler _amqProtocolHandler; @@ -99,10 +94,9 @@ public class FailoverHandler implements Runnable * @param amqProtocolHandler The protocol handler that spans the failover. * @param session The MINA session, for the failing connection. */ - public FailoverHandler(AMQProtocolHandler amqProtocolHandler, IoSession session) + public FailoverHandler(AMQProtocolHandler amqProtocolHandler) { _amqProtocolHandler = amqProtocolHandler; - _session = session; } /** @@ -221,7 +215,7 @@ public class FailoverHandler implements Runnable _amqProtocolHandler.setFailoverState(FailoverState.FAILED); /*try {*/ - _amqProtocolHandler.exceptionCaught(_session, e); + _amqProtocolHandler.exception(e); /*} catch (Exception ex) { diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQIoTransportProtocolSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQIoTransportProtocolSession.java deleted file mode 100644 index 8782e00a12..0000000000 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQIoTransportProtocolSession.java +++ /dev/null @@ -1,146 +0,0 @@ -package org.apache.qpid.client.protocol; -/* - * - * 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. - * - */ - - -import java.util.UUID; - -import javax.security.sasl.SaslClient; - -import org.apache.commons.lang.StringUtils; -import org.apache.mina.common.IdleStatus; -import org.apache.qpid.AMQException; -import org.apache.qpid.client.AMQConnection; -import org.apache.qpid.client.ConnectionTuneParameters; -import org.apache.qpid.client.handler.ClientMethodDispatcherImpl; -import org.apache.qpid.client.state.AMQState; -import org.apache.qpid.framing.AMQDataBlock; -import org.apache.qpid.framing.AMQMethodBody; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.ProtocolInitiation; -import org.apache.qpid.framing.ProtocolVersion; -import org.apache.qpid.transport.Sender; - -public class AMQIoTransportProtocolSession extends AMQProtocolSession -{ - - protected Sender _ioSender; - private SaslClient _saslClient; - private ConnectionTuneParameters _connectionTuneParameters; - - public AMQIoTransportProtocolSession(AMQProtocolHandler protocolHandler, AMQConnection connection) - { - super(protocolHandler, connection); - } - - @Override - public void closeProtocolSession(boolean waitLast) - { - _ioSender.close(); - _protocolHandler.getStateManager().changeState(AMQState.CONNECTION_CLOSED); - } - - @Override - public void init() - { - _ioSender.send(new ProtocolInitiation(_connection.getProtocolVersion()).toNioByteBuffer()); - _ioSender.flush(); - } - - @Override - protected AMQShortString generateQueueName() - { - int id; - synchronized (_queueIdLock) - { - id = _queueId++; - } - return new AMQShortString("tmp_" + UUID.randomUUID() + "_" + id); - } - - @Override - public AMQConnection getAMQConnection() - { - return _connection; - } - - @Override - public SaslClient getSaslClient() - { - return _saslClient; - } - - @Override - public void setSaslClient(SaslClient client) - { - _saslClient = client; - } - - /** @param delay delay in seconds (not ms) */ - @Override - void initHeartbeats(int delay) - { - if (delay > 0) - { - // FIXME: actually do something here - HeartbeatDiagnostics.init(delay, HeartbeatConfig.CONFIG.getTimeout(delay)); - } - } - - @Override - public void methodFrameReceived(final int channel, final AMQMethodBody amqMethodBody) throws AMQException - { - // FIXME? - _protocolHandler.methodBodyReceived(channel, amqMethodBody, null); - } - - @Override - public void writeFrame(AMQDataBlock frame, boolean wait) - { - _ioSender.send(frame.toNioByteBuffer()); - if (wait) - { - _ioSender.flush(); - } - } - - @Override - public void setSender(Sender sender) - { - _ioSender = sender; - } - - @Override - public ConnectionTuneParameters getConnectionTuneParameters() - { - return _connectionTuneParameters; - } - - @Override - public void setConnectionTuneParameters(ConnectionTuneParameters params) - { - _connectionTuneParameters = params; - AMQConnection con = getAMQConnection(); - con.setMaximumChannelCount(params.getChannelMax()); - con.setMaximumFrameSize(params.getFrameMax()); - initHeartbeats((int) params.getHeartbeat()); - } -} diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java index 6f62070fd0..be75fc150f 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java @@ -20,24 +20,22 @@ */ package org.apache.qpid.client.protocol; -import org.apache.mina.common.IdleStatus; -import org.apache.mina.common.IoFilterChain; -import org.apache.mina.common.IoHandlerAdapter; -import org.apache.mina.common.IoSession; -import org.apache.mina.filter.ReadThrottleFilterBuilder; -import org.apache.mina.filter.SSLFilter; -import org.apache.mina.filter.WriteBufferLimitFilterBuilder; +import java.io.IOException; +import java.net.SocketAddress; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.CountDownLatch; + import org.apache.mina.filter.codec.ProtocolCodecException; -import org.apache.mina.filter.codec.ProtocolCodecFilter; -import org.apache.mina.filter.executor.ExecutorFilter; import org.apache.qpid.AMQConnectionClosedException; import org.apache.qpid.AMQDisconnectedException; import org.apache.qpid.AMQException; import org.apache.qpid.AMQTimeoutException; import org.apache.qpid.client.AMQConnection; import org.apache.qpid.client.AMQSession; -import org.apache.qpid.client.SSLConfiguration; -import org.apache.qpid.client.configuration.ClientProperties; import org.apache.qpid.client.failover.FailoverException; import org.apache.qpid.client.failover.FailoverHandler; import org.apache.qpid.client.failover.FailoverState; @@ -46,23 +44,29 @@ import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateWaiter; import org.apache.qpid.client.state.listener.SpecificMethodFrameListener; import org.apache.qpid.codec.AMQCodecFactory; -import org.apache.qpid.framing.*; +import org.apache.qpid.framing.AMQBody; +import org.apache.qpid.framing.AMQDataBlock; +import org.apache.qpid.framing.AMQFrame; +import org.apache.qpid.framing.AMQMethodBody; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.ConnectionCloseBody; +import org.apache.qpid.framing.ConnectionCloseOkBody; +import org.apache.qpid.framing.HeartbeatBody; +import org.apache.qpid.framing.MethodRegistry; +import org.apache.qpid.framing.ProtocolInitiation; +import org.apache.qpid.framing.ProtocolVersion; import org.apache.qpid.jms.BrokerDetails; -import org.apache.qpid.pool.ReadWriteThreadModel; +import org.apache.qpid.pool.Job; +import org.apache.qpid.pool.ReferenceCountingExecutorService; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.protocol.AMQMethodListener; -import org.apache.qpid.ssl.SSLContextFactory; +import org.apache.qpid.protocol.ProtocolEngine; +import org.apache.qpid.transport.NetworkDriver; import org.apache.qpid.transport.network.io.IoTransport; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.util.Iterator; -import java.util.Set; -import java.util.concurrent.CopyOnWriteArraySet; -import java.util.concurrent.CountDownLatch; - /** * AMQProtocolHandler is the client side protocol handler for AMQP, it handles all protocol events received from the * network by MINA. The primary purpose of AMQProtocolHandler is to translate the generic event model of MINA into the @@ -102,9 +106,6 @@ import java.util.concurrent.CountDownLatch; * *

*
CRC Card
Responsibilities Collaborations - *
Create the filter chain to filter this handlers events. - * {@link ProtocolCodecFilter}, {@link SSLContextFactory}, {@link SSLFilter}, {@link ReadWriteThreadModel}. - * *
Maintain fail-over state. *
*
@@ -120,7 +121,7 @@ import java.util.concurrent.CountDownLatch; * held per protocol handler, per protocol session, per network connection, per channel, in seperate classes, so * that lifecycles of the fields match lifecycles of their containing objects. */ -public class AMQProtocolHandler extends IoHandlerAdapter +public class AMQProtocolHandler implements ProtocolEngine { /** Used for debugging. */ private static final Logger _logger = LoggerFactory.getLogger(AMQProtocolHandler.class); @@ -137,7 +138,7 @@ public class AMQProtocolHandler extends IoHandlerAdapter private volatile AMQProtocolSession _protocolSession; /** Holds the state of the protocol session. */ - private AMQStateManager _stateManager = new AMQStateManager(); + private AMQStateManager _stateManager; /** Holds the method listeners, */ private final CopyOnWriteArraySet _frameListeners = new CopyOnWriteArraySet(); @@ -166,7 +167,15 @@ public class AMQProtocolHandler extends IoHandlerAdapter /** Object to lock on when changing the latch */ private Object _failoverLatchChange = new Object(); - + private AMQCodecFactory _codecFactory; + private Job _readJob; + private Job _writeJob; + private ReferenceCountingExecutorService _poolReference = ReferenceCountingExecutorService.getInstance(); + private NetworkDriver _networkDriver; + + private long _writtenBytes; + private long _readBytes; + /** * Creates a new protocol handler, associated with the specified client connection instance. * @@ -175,86 +184,13 @@ public class AMQProtocolHandler extends IoHandlerAdapter public AMQProtocolHandler(AMQConnection con) { _connection = con; - } - - /** - * Invoked by MINA when a MINA session for a new connection is created. This method sets up the filter chain on the - * session, which filters the events handled by this handler. The filter chain consists of, handing off events - * to an asynchronous thread pool, optionally encoding/decoding ssl, encoding/decoding AMQP. - * - * @param session The MINA session. - * - * @throws Exception Any underlying exceptions are allowed to fall through to MINA. - */ - public void sessionCreated(IoSession session) throws Exception - { - _logger.debug("Protocol session created for session " + System.identityHashCode(session)); - _failoverHandler = new FailoverHandler(this, session); - - final ProtocolCodecFilter pcf = new ProtocolCodecFilter(new AMQCodecFactory(false)); - - if (Boolean.getBoolean("amqj.shared_read_write_pool")) - { - session.getFilterChain().addBefore("AsynchronousWriteFilter", "protocolFilter", pcf); - } - else - { - session.getFilterChain().addLast("protocolFilter", pcf); - } - // we only add the SSL filter where we have an SSL connection - if (_connection.getSSLConfiguration() != null) - { - SSLConfiguration sslConfig = _connection.getSSLConfiguration(); - SSLContextFactory sslFactory = - new SSLContextFactory(sslConfig.getKeystorePath(), sslConfig.getKeystorePassword(), sslConfig.getCertType()); - SSLFilter sslFilter = new SSLFilter(sslFactory.buildClientContext()); - sslFilter.setUseClientMode(true); - session.getFilterChain().addBefore("protocolFilter", "ssl", sslFilter); - } - - try - { - ReadWriteThreadModel threadModel = ReadWriteThreadModel.getInstance(); - threadModel.getAsynchronousReadFilter().createNewJobForSession(session); - threadModel.getAsynchronousWriteFilter().createNewJobForSession(session); - } - catch (RuntimeException e) - { - _logger.error(e.getMessage(), e); - } - - if (Boolean.getBoolean(ClientProperties.PROTECTIO_PROP_NAME)) - { - try - { - //Add IO Protection Filters - IoFilterChain chain = session.getFilterChain(); - - session.getFilterChain().addLast("tempExecutorFilterForFilterBuilder", new ExecutorFilter()); - - ReadThrottleFilterBuilder readfilter = new ReadThrottleFilterBuilder(); - readfilter.setMaximumConnectionBufferSize(Integer.parseInt(System.getProperty( - ClientProperties.READ_BUFFER_LIMIT_PROP_NAME, ClientProperties.READ_BUFFER_LIMIT_DEFAULT))); - readfilter.attach(chain); - - WriteBufferLimitFilterBuilder writefilter = new WriteBufferLimitFilterBuilder(); - writefilter.setMaximumConnectionBufferSize(Integer.parseInt(System.getProperty( - ClientProperties.WRITE_BUFFER_LIMIT_PROP_NAME, ClientProperties.WRITE_BUFFER_LIMIT_DEFAULT))); - writefilter.attach(chain); - session.getFilterChain().remove("tempExecutorFilterForFilterBuilder"); - - _logger.info("Using IO Read/Write Filter Protection"); - } - catch (Exception e) - { - _logger.error("Unable to attach IO Read/Write Filter Protection :" + e.getMessage()); - } - } - _protocolSession = new AMQProtocolSession(this, session, _connection); - - _stateManager.setProtocolSession(_protocolSession); - - _protocolSession.init(); + _protocolSession = new AMQProtocolSession(this, _connection); + _stateManager = new AMQStateManager(_protocolSession); + _codecFactory = new AMQCodecFactory(false, _protocolSession); + _readJob = new Job(_poolReference, Job.MAX_JOB_EVENTS, true); + _writeJob = new Job(_poolReference, Job.MAX_JOB_EVENTS, false); + _poolReference.acquireExecutorService(); + _failoverHandler = new FailoverHandler(this); } /** @@ -283,12 +219,10 @@ public class AMQProtocolHandler extends IoHandlerAdapter * may be called first followed by this method. This depends on whether the client was trying to send data at the * time of the failure. * - * @param session The MINA session. - * * @todo Clarify: presumably exceptionCaught is called when the client is sending during a connection failure and * not otherwise? The above comment doesn't make that clear. */ - public void sessionClosed(IoSession session) + public void closed() { if (_connection.isClosed()) { @@ -327,7 +261,8 @@ public class AMQProtocolHandler extends IoHandlerAdapter { _logger.debug("sessionClose() not allowed to failover"); _connection.exceptionReceived(new AMQDisconnectedException( - "Server closed connection and reconnection " + "not permitted.", null)); + "Server closed connection and reconnection " + "not permitted.", + _stateManager.getLastException())); } else { @@ -350,43 +285,39 @@ public class AMQProtocolHandler extends IoHandlerAdapter failoverThread.start(); } - public void sessionIdle(IoSession session, IdleStatus status) throws Exception + @Override + public void readerIdle() { - _logger.debug("Protocol Session [" + this + ":" + session + "] idle: " + status); - if (IdleStatus.WRITER_IDLE.equals(status)) - { - // write heartbeat frame: - _logger.debug("Sent heartbeat"); - session.write(HeartbeatBody.FRAME); - HeartbeatDiagnostics.sent(); - } - else if (IdleStatus.READER_IDLE.equals(status)) - { - // failover: - HeartbeatDiagnostics.timeout(); - _logger.warn("Timed out while waiting for heartbeat from peer."); - session.close(); - } + _logger.debug("Protocol Session [" + this + "] idle: reader"); + // failover: + HeartbeatDiagnostics.timeout(); + _logger.warn("Timed out while waiting for heartbeat from peer."); + _networkDriver.close(); + } + + @Override + public void writerIdle() + { + _logger.debug("Protocol Session [" + this + "] idle: reader"); + writeFrame(HeartbeatBody.FRAME); + HeartbeatDiagnostics.sent(); } /** - * Invoked when any exception is thrown by a user IoHandler implementation or by MINA. If the cause is an - * IOException, MINA will close the connection automatically. - * - * @param session The MINA session. - * @param cause The exception that triggered this event. + * Invoked when any exception is thrown by the NetworkDriver */ - public void exceptionCaught(IoSession session, Throwable cause) + public void exception(Throwable cause) { + _logger.info("AS: HELLO"); if (_failoverState == FailoverState.NOT_STARTED) { // if (!(cause instanceof AMQUndeliveredException) && (!(cause instanceof AMQAuthenticationException))) if ((cause instanceof AMQConnectionClosedException) || cause instanceof IOException) { _logger.info("Exception caught therefore going to attempt failover: " + cause, cause); - // this will attemp failover - - sessionClosed(session); + // this will attempt failover + _networkDriver.close(); + closed(); } else { @@ -437,6 +368,7 @@ public class AMQProtocolHandler extends IoHandlerAdapter public void propagateExceptionToAllWaiters(Exception e) { getStateManager().error(e); + propagateExceptionToFrameListeners(e); } @@ -490,48 +422,84 @@ public class AMQProtocolHandler extends IoHandlerAdapter private static int _messageReceivedCount; - public void messageReceived(IoSession session, Object message) throws Exception - { - if (PROTOCOL_DEBUG) - { - _protocolLogger.info(String.format("RECV: [%s] %s", this, message)); - } - if(message instanceof AMQFrame) + @Override + public void received(ByteBuffer msg) + { + try { - final boolean debug = _logger.isDebugEnabled(); - final long msgNumber = ++_messageReceivedCount; + _readBytes += msg.remaining(); + final ArrayList dataBlocks = _codecFactory.getDecoder().decodeBuffer(msg); - if (debug && ((msgNumber % 1000) == 0)) + Job.fireAsynchEvent(_poolReference.getPool(), _readJob, new Runnable() { - _logger.debug("Received " + _messageReceivedCount + " protocol messages"); - } - - AMQFrame frame = (AMQFrame) message; - - final AMQBody bodyFrame = frame.getBodyFrame(); - - HeartbeatDiagnostics.received(bodyFrame instanceof HeartbeatBody); + @Override + public void run() + { + // Decode buffer - bodyFrame.handle(frame.getChannel(), _protocolSession); + for (AMQDataBlock message : dataBlocks) + { - _connection.bytesReceived(_protocolSession.getIoSession().getReadBytes()); + try + { + if (PROTOCOL_DEBUG) + { + _protocolLogger.info(String.format("RECV: [%s] %s", this, message)); + } + + if(message instanceof AMQFrame) + { + final boolean debug = _logger.isDebugEnabled(); + final long msgNumber = ++_messageReceivedCount; + + if (debug && ((msgNumber % 1000) == 0)) + { + _logger.debug("Received " + _messageReceivedCount + " protocol messages"); + } + + AMQFrame frame = (AMQFrame) message; + + final AMQBody bodyFrame = frame.getBodyFrame(); + + HeartbeatDiagnostics.received(bodyFrame instanceof HeartbeatBody); + + bodyFrame.handle(frame.getChannel(), _protocolSession); + + _connection.bytesReceived(_readBytes); + } + else if (message instanceof ProtocolInitiation) + { + // We get here if the server sends a response to our initial protocol header + // suggesting an alternate ProtocolVersion; the server will then close the + // connection. + ProtocolInitiation protocolInit = (ProtocolInitiation) message; + ProtocolVersion pv = protocolInit.checkVersion(); + getConnection().setProtocolVersion(pv); + + // get round a bug in old versions of qpid whereby the connection is not closed + _stateManager.changeState(AMQState.CONNECTION_CLOSED); + } + } + catch (Exception e) + { + e.printStackTrace(); + _logger.error("Exception processing frame", e); + propagateExceptionToFrameListeners(e); + exception(e); + } + } + } + }); } - else if (message instanceof ProtocolInitiation) + catch (Exception e) { - // We get here if the server sends a response to our initial protocol header - // suggesting an alternate ProtocolVersion; the server will then close the - // connection. - ProtocolInitiation protocolInit = (ProtocolInitiation) message; - ProtocolVersion pv = protocolInit.checkVersion(); - getConnection().setProtocolVersion(pv); - - // get round a bug in old versions of qpid whereby the connection is not closed - _stateManager.changeState(AMQState.CONNECTION_CLOSED); + propagateExceptionToFrameListeners(e); + exception(e); } } - public void methodBodyReceived(final int channelId, final AMQBody bodyFrame, IoSession session)//, final IoSession session) + public void methodBodyReceived(final int channelId, final AMQBody bodyFrame) throws AMQException { @@ -571,32 +539,13 @@ public class AMQProtocolHandler extends IoHandlerAdapter { propagateExceptionToFrameListeners(e); - exceptionCaught(session, e); + exception(e); } } private static int _messagesOut; - public void messageSent(IoSession session, Object message) throws Exception - { - if (PROTOCOL_DEBUG) - { - _protocolLogger.debug(String.format("SEND: [%s] %s", this, message)); - } - - final long sentMessages = _messagesOut++; - - final boolean debug = _logger.isDebugEnabled(); - - if (debug && ((sentMessages % 1000) == 0)) - { - _logger.debug("Sent " + _messagesOut + " protocol messages"); - } - - _connection.bytesSent(session.getWrittenBytes()); - } - public StateWaiter createWaiter(Set states) throws AMQException { return getStateManager().createWaiter(states); @@ -610,12 +559,34 @@ public class AMQProtocolHandler extends IoHandlerAdapter */ public void writeFrame(AMQDataBlock frame) { - _protocolSession.writeFrame(frame); + writeFrame(frame, false); } public void writeFrame(AMQDataBlock frame, boolean wait) { - _protocolSession.writeFrame(frame, wait); + ByteBuffer buf = frame.toNioByteBuffer(); + _writtenBytes += buf.remaining(); + _networkDriver.send(buf); + if (PROTOCOL_DEBUG) + { + _protocolLogger.debug(String.format("SEND: [%s] %s", this, frame)); + } + + final long sentMessages = _messagesOut++; + + final boolean debug = _logger.isDebugEnabled(); + + if (debug && ((sentMessages % 1000) == 0)) + { + _logger.debug("Sent " + _messagesOut + " protocol messages"); + } + + _connection.bytesSent(_writtenBytes); + + if (wait) + { + _networkDriver.flush(); + } } /** @@ -673,7 +644,7 @@ public class AMQProtocolHandler extends IoHandlerAdapter //FIXME: At this point here we should check or before add we should check _stateManager is in an open // state so as we don't check we are likely just to time out here as I believe is being seen in QPID-1255 } - _protocolSession.writeFrame(frame); + writeFrame(frame); return listener.blockForFrame(timeout); // When control resumes before this line, a reply will have been received @@ -723,38 +694,36 @@ public class AMQProtocolHandler extends IoHandlerAdapter final AMQFrame frame = body.generateFrame(0); //If the connection is already closed then don't do a syncWrite - if (getStateManager().getCurrentState().equals(AMQState.CONNECTION_CLOSED)) - { - _protocolSession.closeProtocolSession(false); - } - else + if (!getStateManager().getCurrentState().equals(AMQState.CONNECTION_CLOSED)) { try { syncWrite(frame, ConnectionCloseOkBody.class, timeout); - _protocolSession.closeProtocolSession(); + _networkDriver.close(); + closed(); } catch (AMQTimeoutException e) { - _protocolSession.closeProtocolSession(false); + closed(); } catch (FailoverException e) { _logger.debug("FailoverException interrupted connection close, ignoring as connection close anyway."); } } + _poolReference.releaseExecutorService(); } /** @return the number of bytes read from this protocol session */ public long getReadBytes() { - return _protocolSession.getIoSession().getReadBytes(); + return _readBytes; } /** @return the number of bytes written to this protocol session */ public long getWrittenBytes() { - return _protocolSession.getIoSession().getWrittenBytes(); + return _writtenBytes; } public void failover(String host, int port) @@ -807,6 +776,7 @@ public class AMQProtocolHandler extends IoHandlerAdapter public void setStateManager(AMQStateManager stateManager) { _stateManager = stateManager; + _stateManager.setProtocolSession(_protocolSession); } public AMQProtocolSession getProtocolSession() @@ -843,4 +813,35 @@ public class AMQProtocolHandler extends IoHandlerAdapter { return _protocolSession.getProtocolVersion(); } + + public SocketAddress getRemoteAddress() + { + return _networkDriver.getRemoteAddress(); + } + + public SocketAddress getLocalAddress() + { + return _networkDriver.getLocalAddress(); + } + + public void setNetworkDriver(NetworkDriver driver) + { + _networkDriver = driver; + } + + /** @param delay delay in seconds (not ms) */ + void initHeartbeats(int delay) + { + if (delay > 0) + { + getNetworkDriver().setMaxWriteIdle(delay); + getNetworkDriver().setMaxReadIdle(HeartbeatConfig.CONFIG.getTimeout(delay)); + HeartbeatDiagnostics.init(delay, HeartbeatConfig.CONFIG.getTimeout(delay)); + } + } + + public NetworkDriver getNetworkDriver() + { + return _networkDriver; + } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java index 0e872170aa..cd049c24a1 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java @@ -20,11 +20,6 @@ */ package org.apache.qpid.client.protocol; -import org.apache.commons.lang.StringUtils; -import org.apache.mina.common.CloseFuture; -import org.apache.mina.common.IdleStatus; -import org.apache.mina.common.IoSession; -import org.apache.mina.common.WriteFuture; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -33,6 +28,7 @@ import javax.security.sasl.SaslClient; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import org.apache.commons.lang.StringUtils; import org.apache.qpid.AMQException; import org.apache.qpid.client.AMQConnection; import org.apache.qpid.client.AMQSession; @@ -65,10 +61,6 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession protected static final String SASL_CLIENT = "SASLClient"; - protected final IoSession _minaProtocolSession; - - protected WriteFuture _lastWriteFuture; - /** * The handler from which this session was created and which is used to handle protocol events. We send failover * events to the handler. @@ -102,28 +94,15 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession protected final AMQConnection _connection; - private static final int FAST_CHANNEL_ACCESS_MASK = 0xFFFFFFF0; + private ConnectionTuneParameters _connectionTuneParameters; - public AMQProtocolSession(AMQProtocolHandler protocolHandler, IoSession protocolSession, AMQConnection connection) - { - _protocolHandler = protocolHandler; - _minaProtocolSession = protocolSession; - _minaProtocolSession.setAttachment(this); - // properties of the connection are made available to the event handlers - _minaProtocolSession.setAttribute(AMQ_CONNECTION, connection); - // fixme - real value needed - _minaProtocolSession.setWriteTimeout(LAST_WRITE_FUTURE_JOIN_TIMEOUT); - _protocolVersion = connection.getProtocolVersion(); - _methodDispatcher = ClientMethodDispatcherImpl.newMethodDispatcher(ProtocolVersion.getLatestSupportedVersion(), - this); - _connection = connection; + private SaslClient _saslClient; - } + private static final int FAST_CHANNEL_ACCESS_MASK = 0xFFFFFFF0; public AMQProtocolSession(AMQProtocolHandler protocolHandler, AMQConnection connection) { - _protocolHandler = protocolHandler; - _minaProtocolSession = null; + _protocolHandler = protocolHandler; _protocolVersion = connection.getProtocolVersion(); _methodDispatcher = ClientMethodDispatcherImpl.newMethodDispatcher(ProtocolVersion.getLatestSupportedVersion(), this); @@ -134,7 +113,7 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession { // start the process of setting up the connection. This is the first place that // data is written to the server. - _minaProtocolSession.write(new ProtocolInitiation(_connection.getProtocolVersion())); + _protocolHandler.writeFrame(new ProtocolInitiation(_connection.getProtocolVersion())); } public String getClientID() @@ -175,14 +154,9 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession return getAMQConnection().getPassword(); } - public IoSession getIoSession() - { - return _minaProtocolSession; - } - public SaslClient getSaslClient() { - return (SaslClient) _minaProtocolSession.getAttribute(SASL_CLIENT); + return _saslClient; } /** @@ -192,28 +166,21 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession */ public void setSaslClient(SaslClient client) { - if (client == null) - { - _minaProtocolSession.removeAttribute(SASL_CLIENT); - } - else - { - _minaProtocolSession.setAttribute(SASL_CLIENT, client); - } + _saslClient = client; } public ConnectionTuneParameters getConnectionTuneParameters() { - return (ConnectionTuneParameters) _minaProtocolSession.getAttribute(CONNECTION_TUNE_PARAMETERS); + return _connectionTuneParameters; } public void setConnectionTuneParameters(ConnectionTuneParameters params) { - _minaProtocolSession.setAttribute(CONNECTION_TUNE_PARAMETERS, params); + _connectionTuneParameters = params; AMQConnection con = getAMQConnection(); con.setMaximumChannelCount(params.getChannelMax()); con.setMaximumFrameSize(params.getFrameMax()); - initHeartbeats((int) params.getHeartbeat()); + _protocolHandler.initHeartbeats((int) params.getHeartbeat()); } /** @@ -335,21 +302,12 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession */ public void writeFrame(AMQDataBlock frame) { - writeFrame(frame, false); + _protocolHandler.writeFrame(frame); } public void writeFrame(AMQDataBlock frame, boolean wait) { - WriteFuture f = _minaProtocolSession.write(frame); - if (wait) - { - // fixme -- time out? - f.join(); - } - else - { - _lastWriteFuture = f; - } + _protocolHandler.writeFrame(frame, wait); } /** @@ -407,33 +365,12 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession public AMQConnection getAMQConnection() { - return (AMQConnection) _minaProtocolSession.getAttribute(AMQ_CONNECTION); + return _connection; } - public void closeProtocolSession() + public void closeProtocolSession() throws AMQException { - closeProtocolSession(true); - } - - public void closeProtocolSession(boolean waitLast) - { - _logger.debug("Waiting for last write to join."); - if (waitLast && (_lastWriteFuture != null)) - { - _lastWriteFuture.join(LAST_WRITE_FUTURE_JOIN_TIMEOUT); - } - - _logger.debug("Closing protocol session"); - - final CloseFuture future = _minaProtocolSession.close(); - - // There is no recovery we can do if the join on the close failes so simply mark the connection CLOSED - // then wait for the connection to close. - // ritchiem: Could this release BlockingWaiters to early? The close has been done as much as possible so any - // error now shouldn't matter. - - _protocolHandler.getStateManager().changeState(AMQState.CONNECTION_CLOSED); - future.join(LAST_WRITE_FUTURE_JOIN_TIMEOUT); + _protocolHandler.closeConnection(0); } public void failover(String host, int port) @@ -449,22 +386,11 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession id = _queueId++; } // get rid of / and : and ; from address for spec conformance - String localAddress = StringUtils.replaceChars(_minaProtocolSession.getLocalAddress().toString(), "/;:", ""); + String localAddress = StringUtils.replaceChars(_protocolHandler.getLocalAddress().toString(), "/;:", ""); return new AMQShortString("tmp_" + localAddress + "_" + id); } - /** @param delay delay in seconds (not ms) */ - void initHeartbeats(int delay) - { - if (delay > 0) - { - _minaProtocolSession.setIdleTime(IdleStatus.WRITER_IDLE, delay); - _minaProtocolSession.setIdleTime(IdleStatus.READER_IDLE, HeartbeatConfig.CONFIG.getTimeout(delay)); - HeartbeatDiagnostics.init(delay, HeartbeatConfig.CONFIG.getTimeout(delay)); - } - } - public void confirmConsumerCancelled(int channelId, AMQShortString consumerTag) { final AMQSession session = getSession(channelId); @@ -530,7 +456,7 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession public void methodFrameReceived(final int channel, final AMQMethodBody amqMethodBody) throws AMQException { - _protocolHandler.methodBodyReceived(channel, amqMethodBody, _minaProtocolSession); + _protocolHandler.methodBodyReceived(channel, amqMethodBody); } public void notifyError(Exception error) diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/QpidThreadExecutor.java b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/QpidThreadExecutor.java deleted file mode 100644 index 3de410fb69..0000000000 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/QpidThreadExecutor.java +++ /dev/null @@ -1,22 +0,0 @@ -package org.apache.qpid.client.transport; - -import org.apache.qpid.thread.Threading; - -import edu.emory.mathcs.backport.java.util.concurrent.Executor; - -public class QpidThreadExecutor implements Executor -{ - @Override - public void execute(Runnable command) - { - try - { - Threading.getThreadFactory().createThread(command).start(); - } - catch(Exception e) - { - throw new RuntimeException("Error creating a thread using Qpid thread factory",e); - } - } - -} diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java index b2f7ae8395..1ac8f62e32 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java @@ -20,27 +20,20 @@ */ package org.apache.qpid.client.transport; +import java.io.IOException; +import java.net.InetSocketAddress; + import org.apache.mina.common.ByteBuffer; -import org.apache.mina.common.ConnectFuture; import org.apache.mina.common.IoConnector; import org.apache.mina.common.SimpleByteBufferAllocator; -import org.apache.mina.transport.socket.nio.ExistingSocketConnector; -import org.apache.mina.transport.socket.nio.SocketConnectorConfig; -import org.apache.mina.transport.socket.nio.SocketSessionConfig; - +import org.apache.qpid.client.SSLConfiguration; import org.apache.qpid.client.protocol.AMQProtocolHandler; import org.apache.qpid.jms.BrokerDetails; -import org.apache.qpid.pool.ReadWriteThreadModel; - +import org.apache.qpid.ssl.SSLContextFactory; +import org.apache.qpid.transport.network.mina.MINANetworkDriver; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.net.InetSocketAddress; -import java.net.Socket; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - public class SocketTransportConnection implements ITransportConnection { private static final Logger _logger = LoggerFactory.getLogger(SocketTransportConnection.class); @@ -71,61 +64,27 @@ public class SocketTransportConnection implements ITransportConnection } final IoConnector ioConnector = _socketConnectorFactory.newSocketConnector(); - SocketConnectorConfig cfg = (SocketConnectorConfig) ioConnector.getDefaultConfig(); - - // if we do not use our own thread model we get the MINA default which is to use - // its own leader-follower model - boolean readWriteThreading = Boolean.getBoolean("amqj.shared_read_write_pool"); - if (readWriteThreading) - { - cfg.setThreadModel(ReadWriteThreadModel.getInstance()); - } - - SocketSessionConfig scfg = (SocketSessionConfig) cfg.getSessionConfig(); - scfg.setTcpNoDelay("true".equalsIgnoreCase(System.getProperty("amqj.tcpNoDelay", "true"))); - scfg.setSendBufferSize(Integer.getInteger("amqj.sendBufferSize", DEFAULT_BUFFER_SIZE)); - _logger.info("send-buffer-size = " + scfg.getSendBufferSize()); - scfg.setReceiveBufferSize(Integer.getInteger("amqj.receiveBufferSize", DEFAULT_BUFFER_SIZE)); - _logger.info("recv-buffer-size = " + scfg.getReceiveBufferSize()); - final InetSocketAddress address; if (brokerDetail.getTransport().equals(BrokerDetails.SOCKET)) { address = null; - - Socket socket = TransportConnection.removeOpenSocket(brokerDetail.getHost()); - - if (socket != null) - { - _logger.info("Using existing Socket:" + socket); - - ((ExistingSocketConnector) ioConnector).setOpenSocket(socket); - } - else - { - throw new IllegalArgumentException("Active Socket must be provided for broker " + - "with 'socket://' transport:" + brokerDetail); - } } else { address = new InetSocketAddress(brokerDetail.getHost(), brokerDetail.getPort()); _logger.info("Attempting connection to " + address); } - - - ConnectFuture future = ioConnector.connect(address, protocolHandler); - - // wait for connection to complete - if (future.join(brokerDetail.getTimeout())) - { - // we call getSession which throws an IOException if there has been an error connecting - future.getSession(); - } - else + + SSLConfiguration sslConfig = protocolHandler.getConnection().getSSLConfiguration(); + SSLContextFactory sslFactory = null; + if (sslConfig != null) { - throw new IOException("Timeout waiting for connection."); + sslFactory = new SSLContextFactory(sslConfig.getKeystorePath(), sslConfig.getKeystorePassword(), sslConfig.getCertType()); } + + MINANetworkDriver driver = new MINANetworkDriver(ioConnector); + driver.open(brokerDetail.getPort(), address.getAddress(), protocolHandler, null, sslFactory); + protocolHandler.setNetworkDriver(driver); } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java index 0bacda04ff..a4f8bb0166 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java @@ -20,6 +20,12 @@ */ package org.apache.qpid.client.transport; +import java.io.IOException; +import java.net.Socket; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + import org.apache.mina.common.IoConnector; import org.apache.mina.common.IoHandlerAdapter; import org.apache.mina.common.IoServiceConfig; @@ -30,16 +36,12 @@ import org.apache.mina.transport.vmpipe.VmPipeAcceptor; import org.apache.mina.transport.vmpipe.VmPipeAddress; import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException; import org.apache.qpid.jms.BrokerDetails; -import org.apache.qpid.pool.ReadWriteThreadModel; +import org.apache.qpid.protocol.ProtocolEngineFactory; +import org.apache.qpid.thread.QpidThreadExecutor; +import org.apache.qpid.transport.network.mina.MINANetworkDriver; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.net.Socket; - /** * The TransportConnection is a helper class responsible for connecting to an AMQ server. It sets up the underlying * connector, which currently always uses TCP/IP sockets. It creates the "protocol handler" which deals with MINA @@ -61,7 +63,7 @@ public class TransportConnection private static Logger _logger = LoggerFactory.getLogger(TransportConnection.class); - private static final String DEFAULT_QPID_SERVER = "org.apache.qpid.server.protocol.AMQPFastProtocolHandler"; + private static final String DEFAULT_QPID_SERVER = "org.apache.qpid.server.protocol.AMQProtocolEngineFactory"; private static Map _openSocketRegister = new ConcurrentHashMap(); @@ -75,7 +77,7 @@ public class TransportConnection return _openSocketRegister.remove(socketID); } - public static synchronized ITransportConnection getInstance(BrokerDetails details) throws AMQTransportConnectionException + public static synchronized ITransportConnection getInstance(final BrokerDetails details) throws AMQTransportConnectionException { int transport = getTransport(details.getTransport()); @@ -91,7 +93,22 @@ public class TransportConnection { public IoConnector newSocketConnector() { - return new ExistingSocketConnector(1,new QpidThreadExecutor()); + ExistingSocketConnector connector = new ExistingSocketConnector(1,new QpidThreadExecutor()); + + Socket socket = TransportConnection.removeOpenSocket(details.getHost()); + + if (socket != null) + { + _logger.info("Using existing Socket:" + socket); + + ((ExistingSocketConnector) connector).setOpenSocket(socket); + } + else + { + throw new IllegalArgumentException("Active Socket must be provided for broker " + + "with 'socket://' transport:" + details); + } + return connector; } }); case TCP: @@ -189,8 +206,6 @@ public class TransportConnection _acceptor = new VmPipeAcceptor(); IoServiceConfig config = _acceptor.getDefaultConfig(); - - config.setThreadModel(ReadWriteThreadModel.getInstance()); } synchronized (_inVmPipeAddress) { @@ -275,7 +290,10 @@ public class TransportConnection { Class[] cnstr = {Integer.class}; Object[] params = {port}; - provider = (IoHandlerAdapter) Class.forName(protocolProviderClass).getConstructor(cnstr).newInstance(params); + + provider = new MINANetworkDriver(); + ProtocolEngineFactory engineFactory = (ProtocolEngineFactory) Class.forName(protocolProviderClass).getConstructor(cnstr).newInstance(params); + ((MINANetworkDriver) provider).setProtocolEngineFactory(engineFactory, true); // Give the broker a second to create _logger.info("Created VMBroker Instance:" + port); } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java index dca6efba67..504d475740 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java @@ -20,25 +20,26 @@ */ package org.apache.qpid.client.transport; +import java.io.IOException; + import org.apache.mina.common.ConnectFuture; -import org.apache.mina.common.IoServiceConfig; import org.apache.mina.transport.vmpipe.QpidVmPipeConnector; import org.apache.mina.transport.vmpipe.VmPipeAddress; import org.apache.mina.transport.vmpipe.VmPipeConnector; import org.apache.qpid.client.protocol.AMQProtocolHandler; import org.apache.qpid.jms.BrokerDetails; -import org.apache.qpid.pool.ReadWriteThreadModel; +import org.apache.qpid.transport.network.mina.MINANetworkDriver; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; - public class VmPipeTransportConnection implements ITransportConnection { private static final Logger _logger = LoggerFactory.getLogger(VmPipeTransportConnection.class); private static int _port; + private MINANetworkDriver _networkDriver; + public VmPipeTransportConnection(int port) { _port = port; @@ -47,16 +48,16 @@ public class VmPipeTransportConnection implements ITransportConnection public void connect(AMQProtocolHandler protocolHandler, BrokerDetails brokerDetail) throws IOException { final VmPipeConnector ioConnector = new QpidVmPipeConnector(); - final IoServiceConfig cfg = ioConnector.getDefaultConfig(); - - cfg.setThreadModel(ReadWriteThreadModel.getInstance()); final VmPipeAddress address = new VmPipeAddress(_port); _logger.info("Attempting connection to " + address); - ConnectFuture future = ioConnector.connect(address, protocolHandler); + _networkDriver = new MINANetworkDriver(ioConnector, protocolHandler); + protocolHandler.setNetworkDriver(_networkDriver); + ConnectFuture future = ioConnector.connect(address, _networkDriver); // wait for connection to complete future.join(); // we call getSession which throws an IOException if there has been an error connecting future.getSession(); + _networkDriver.setProtocolEngine(protocolHandler); } } diff --git a/qpid/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java b/qpid/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java index fc7f8148f0..f520a21ba0 100644 --- a/qpid/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java +++ b/qpid/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java @@ -27,6 +27,7 @@ import org.apache.qpid.framing.AMQMethodBody; import org.apache.qpid.framing.amqp_8_0.BasicRecoverOkBodyImpl; import org.apache.qpid.AMQException; import org.apache.qpid.protocol.AMQConstant; +import org.apache.qpid.transport.TestNetworkDriver; import org.apache.qpid.client.MockAMQConnection; import org.apache.qpid.client.AMQAuthenticationException; import org.apache.qpid.client.state.AMQState; @@ -72,9 +73,7 @@ public class AMQProtocolHandlerTest extends TestCase { //Create a new ProtocolHandler with a fake connection. _handler = new AMQProtocolHandler(new MockAMQConnection("amqp://guest:guest@client/test?brokerlist='vm://:1'")); - - _handler.sessionCreated(new MockIoSession()); - + _handler.setNetworkDriver(new TestNetworkDriver()); AMQBody body = BasicRecoverOkBodyImpl.getFactory().newInstance(null, 1); _blockFrame = new AMQFrame(0, body); diff --git a/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java b/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java index fa890d0ebb..591dbd085b 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java @@ -23,6 +23,7 @@ package org.apache.qpid.codec; import org.apache.mina.filter.codec.ProtocolCodecFactory; import org.apache.mina.filter.codec.ProtocolDecoder; import org.apache.mina.filter.codec.ProtocolEncoder; +import org.apache.qpid.protocol.AMQVersionAwareProtocolSession; /** * AMQCodecFactory is a Mina codec factory. It supplies the encoders and decoders need to read and write the bytes to @@ -50,9 +51,9 @@ public class AMQCodecFactory implements ProtocolCodecFactory * @param expectProtocolInitiation true if the first frame received is going to be a protocol initiation * frame, false if it is going to be a standard AMQ data block. */ - public AMQCodecFactory(boolean expectProtocolInitiation) + public AMQCodecFactory(boolean expectProtocolInitiation, AMQVersionAwareProtocolSession session) { - _frameDecoder = new AMQDecoder(expectProtocolInitiation); + _frameDecoder = new AMQDecoder(expectProtocolInitiation, session); } /** @@ -70,7 +71,7 @@ public class AMQCodecFactory implements ProtocolCodecFactory * * @return The AMQP decoder. */ - public ProtocolDecoder getDecoder() + public AMQDecoder getDecoder() { return _frameDecoder; } diff --git a/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java b/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java index 7eef73f337..281c0761d9 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java @@ -20,14 +20,21 @@ */ package org.apache.qpid.codec; +import java.util.ArrayList; + import org.apache.mina.common.ByteBuffer; import org.apache.mina.common.IoSession; import org.apache.mina.common.SimpleByteBufferAllocator; import org.apache.mina.filter.codec.CumulativeProtocolDecoder; import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.qpid.framing.AMQDataBlock; import org.apache.qpid.framing.AMQDataBlockDecoder; +import org.apache.qpid.framing.AMQFrameDecodingException; +import org.apache.qpid.framing.AMQMethodBodyFactory; +import org.apache.qpid.framing.AMQProtocolVersionException; import org.apache.qpid.framing.ProtocolInitiation; +import org.apache.qpid.protocol.AMQVersionAwareProtocolSession; /** * AMQDecoder delegates the decoding of AMQP either to a data block decoder, or in the case of new connections, to a @@ -62,14 +69,19 @@ public class AMQDecoder extends CumulativeProtocolDecoder private boolean _expectProtocolInitiation; private boolean firstDecode = true; + private AMQMethodBodyFactory _bodyFactory; + + private ByteBuffer _remainingBuf; + /** * Creates a new AMQP decoder. * * @param expectProtocolInitiation true if this decoder needs to handle protocol initiation. */ - public AMQDecoder(boolean expectProtocolInitiation) + public AMQDecoder(boolean expectProtocolInitiation, AMQVersionAwareProtocolSession session) { _expectProtocolInitiation = expectProtocolInitiation; + _bodyFactory = new AMQMethodBodyFactory(session); } /** @@ -120,7 +132,7 @@ public class AMQDecoder extends CumulativeProtocolDecoder protected boolean doDecodeDataBlock(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception { int pos = in.position(); - boolean enoughData = _dataBlockDecoder.decodable(session, in); + boolean enoughData = _dataBlockDecoder.decodable(in.buf()); in.position(pos); if (!enoughData) { @@ -149,7 +161,7 @@ public class AMQDecoder extends CumulativeProtocolDecoder */ private boolean doDecodePI(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception { - boolean enoughData = _piDecoder.decodable(session, in); + boolean enoughData = _piDecoder.decodable(in.buf()); if (!enoughData) { // returning false means it will leave the contents in the buffer and @@ -158,7 +170,8 @@ public class AMQDecoder extends CumulativeProtocolDecoder } else { - _piDecoder.decode(session, in, out); + ProtocolInitiation pi = new ProtocolInitiation(in.buf()); + out.write(pi); return true; } @@ -177,7 +190,7 @@ public class AMQDecoder extends CumulativeProtocolDecoder } - /** + /** * Cumulates content of in into internal buffer and forwards * decoding request to {@link #doDecode(IoSession, ByteBuffer, ProtocolDecoderOutput)}. * doDecode() is invoked repeatedly until it returns false @@ -268,4 +281,60 @@ public class AMQDecoder extends CumulativeProtocolDecoder session.setAttribute( BUFFER, remainingBuf ); } + public ArrayList decodeBuffer(java.nio.ByteBuffer buf) throws AMQFrameDecodingException, AMQProtocolVersionException + { + + // get prior remaining data from accumulator + ArrayList dataBlocks = new ArrayList(); + ByteBuffer msg; + // if we have a session buffer, append data to that otherwise + // use the buffer read from the network directly + if( _remainingBuf != null ) + { + _remainingBuf.put(buf); + _remainingBuf.flip(); + msg = _remainingBuf; + } + else + { + msg = ByteBuffer.wrap(buf); + } + + if (_expectProtocolInitiation + || (firstDecode + && (msg.remaining() > 0) + && (msg.get(msg.position()) == (byte)'A'))) + { + if (_piDecoder.decodable(msg.buf())) + { + dataBlocks.add(new ProtocolInitiation(msg.buf())); + } + } + else + { + boolean enoughData = true; + while (enoughData) + { + int pos = msg.position(); + + enoughData = _dataBlockDecoder.decodable(msg); + msg.position(pos); + if (enoughData) + { + dataBlocks.add(_dataBlockDecoder.createAndPopulateFrame(_bodyFactory, msg)); + } + else + { + _remainingBuf = SIMPLE_BYTE_BUFFER_ALLOCATOR.allocate(msg.remaining(), false); + _remainingBuf.setAutoExpand(true); + _remainingBuf.put(msg); + } + } + } + if(firstDecode && dataBlocks.size() > 0) + { + firstDecode = false; + } + return dataBlocks; + } } diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java index 82ffc60802..228867b2b0 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java @@ -47,7 +47,7 @@ public class AMQDataBlockDecoder public AMQDataBlockDecoder() { } - public boolean decodable(IoSession session, ByteBuffer in) throws AMQFrameDecodingException + public boolean decodable(java.nio.ByteBuffer in) throws AMQFrameDecodingException { final int remainingAfterAttributes = in.remaining() - (1 + 2 + 4 + 1); // type, channel, body length and end byte @@ -56,14 +56,15 @@ public class AMQDataBlockDecoder return false; } - in.skip(1 + 2); - final long bodySize = in.getUnsignedInt(); + in.position(in.position() + 1 + 2); + // Get an unsigned int, lifted from MINA ByteBuffer getUnsignedInt() + final long bodySize = in.getInt() & 0xffffffffL; return (remainingAfterAttributes >= bodySize); } - protected Object createAndPopulateFrame(IoSession session, ByteBuffer in) + public AMQFrame createAndPopulateFrame(AMQMethodBodyFactory methodBodyFactory, ByteBuffer in) throws AMQFrameDecodingException, AMQProtocolVersionException { final byte type = in.get(); @@ -71,15 +72,7 @@ public class AMQDataBlockDecoder BodyFactory bodyFactory; if (type == AMQMethodBody.TYPE) { - bodyFactory = (BodyFactory) session.getAttribute(SESSION_METHOD_BODY_FACTORY); - if (bodyFactory == null) - { - AMQVersionAwareProtocolSession protocolSession = (AMQVersionAwareProtocolSession) session.getAttachment(); - bodyFactory = new AMQMethodBodyFactory(protocolSession); - session.setAttribute(SESSION_METHOD_BODY_FACTORY, bodyFactory); - - } - + bodyFactory = methodBodyFactory; } else { @@ -115,6 +108,24 @@ public class AMQDataBlockDecoder public void decode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception { - out.write(createAndPopulateFrame(session, in)); + AMQMethodBodyFactory bodyFactory = (AMQMethodBodyFactory) session.getAttribute(SESSION_METHOD_BODY_FACTORY); + if (bodyFactory == null) + { + AMQVersionAwareProtocolSession protocolSession = (AMQVersionAwareProtocolSession) session.getAttachment(); + bodyFactory = new AMQMethodBodyFactory(protocolSession); + session.setAttribute(SESSION_METHOD_BODY_FACTORY, bodyFactory); + } + + out.write(createAndPopulateFrame(bodyFactory, in)); + } + + public boolean decodable(ByteBuffer msg) throws AMQFrameDecodingException + { + return decodable(msg.buf()); + } + + public AMQDataBlock createAndPopulateFrame(AMQMethodBodyFactory factory, java.nio.ByteBuffer msg) throws AMQProtocolVersionException, AMQFrameDecodingException + { + return createAndPopulateFrame(factory, ByteBuffer.wrap(msg)); } } diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java index 05fd2bb480..374644b4f2 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java @@ -50,7 +50,7 @@ public final class AMQDataBlockEncoder implements MessageEncoder { _logger.debug("Encoded frame byte-buffer is '" + EncodingUtils.convertToHexString(buffer) + "'"); } - + out.write(buffer); } diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java index 3ac17e9204..cf8a866e47 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java @@ -20,12 +20,10 @@ */ package org.apache.qpid.framing; -import org.apache.mina.common.ByteBuffer; -import org.apache.mina.common.IoSession; -import org.apache.mina.filter.codec.ProtocolDecoderOutput; import org.apache.qpid.AMQException; import java.io.UnsupportedEncodingException; +import java.nio.ByteBuffer; public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQDataBlock { @@ -53,13 +51,12 @@ public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQData _protocolMajor = protocolMajor; _protocolMinor = protocolMinor; } - + public ProtocolInitiation(ProtocolVersion pv) { this(AMQP_HEADER, CURRENT_PROTOCOL_CLASS, TCP_PROTOCOL_INSTANCE, pv.getMajorVersion(), pv.getMinorVersion()); } - public ProtocolInitiation(ByteBuffer in) { _protocolHeader = new byte[4]; @@ -71,6 +68,11 @@ public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQData _protocolMinor = in.get(); } + public void writePayload(org.apache.mina.common.ByteBuffer buffer) + { + writePayload(buffer.buf()); + } + public long getSize() { return 4 + 1 + 1 + 1 + 1; @@ -127,16 +129,11 @@ public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQData * @return true if we have enough data to decode the PI frame fully, false if more * data is required */ - public boolean decodable(IoSession session, ByteBuffer in) + public boolean decodable(ByteBuffer in) { return (in.remaining() >= 8); } - public void decode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) - { - ProtocolInitiation pi = new ProtocolInitiation(in); - out.write(pi); - } } public ProtocolVersion checkVersion() throws AMQException @@ -192,4 +189,5 @@ public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQData buffer.append(Integer.toHexString(_protocolMinor)); return buffer.toString(); } + } diff --git a/qpid/java/common/src/main/java/org/apache/qpid/pool/Event.java b/qpid/java/common/src/main/java/org/apache/qpid/pool/Event.java deleted file mode 100644 index 5996cbf89c..0000000000 --- a/qpid/java/common/src/main/java/org/apache/qpid/pool/Event.java +++ /dev/null @@ -1,155 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ -package org.apache.qpid.pool; - -import org.apache.mina.common.IoFilter; -import org.apache.mina.common.IoSession; - -/** - * An Event is a continuation, which is used to break a Mina filter chain and save the current point in the chain - * for later processing. It is an abstract class, with different implementations for continuations of different kinds - * of Mina events. - * - *

These continuations are typically batched by {@link Job} for processing by a worker thread pool. - * - *

- *
CRC Card
Responsibilities Collaborations - *
Process a continuation in the context of a Mina session. - *
- * - * @todo Pull up _nextFilter and getNextFilter into Event, as all events use it. Inner classes need to be non-static - * to use instance variables in the parent. Consequently they need to be non-inner to be instantiable outside of - * the context of the outer Event class. The inner class construction used here is preventing common code re-use - * (though not by a huge amount), but makes for an inelegent way of handling inheritance and doesn't seem like - * a justifiable use of inner classes. Move the inner classes out into their own files. - * - * @todo Could make Event implement Runnable, FutureTask, or a custom Continuation interface, to clarify its status as - * a continuation. Job is also a continuation, as is the job completion handler. Or, as Event is totally abstract, - * it is really an interface, so could just drop it and use the continuation interface instead. - */ -public abstract class Event -{ - /** - * Creates a continuation. - */ - public Event() - { } - - /** - * Processes the continuation in the context of a Mina session. - * - * @param session The Mina session. - */ - public abstract void process(IoSession session); - - /** - * A continuation ({@link Event}) that takes a Mina messageReceived event, and passes it to a NextFilter. - * - *

- *
CRC Card
Responsibilities Collaborations - *
Pass a Mina messageReceived event to a NextFilter. {@link IoFilter.NextFilter}, {@link IoSession} - *
- */ - public static final class ReceivedEvent extends Event - { - private final Object _data; - - private final IoFilter.NextFilter _nextFilter; - - public ReceivedEvent(final IoFilter.NextFilter nextFilter, final Object data) - { - super(); - _nextFilter = nextFilter; - _data = data; - } - - public void process(IoSession session) - { - _nextFilter.messageReceived(session, _data); - } - - public IoFilter.NextFilter getNextFilter() - { - return _nextFilter; - } - } - - /** - * A continuation ({@link Event}) that takes a Mina filterWrite event, and passes it to a NextFilter. - * - *

- *
CRC Card
Responsibilities Collaborations - *
Pass a Mina filterWrite event to a NextFilter. - * {@link IoFilter.NextFilter}, {@link IoFilter.WriteRequest}, {@link IoSession} - *
- */ - public static final class WriteEvent extends Event - { - private final IoFilter.WriteRequest _data; - private final IoFilter.NextFilter _nextFilter; - - public WriteEvent(final IoFilter.NextFilter nextFilter, final IoFilter.WriteRequest data) - { - super(); - _nextFilter = nextFilter; - _data = data; - } - - public void process(IoSession session) - { - _nextFilter.filterWrite(session, _data); - } - - public IoFilter.NextFilter getNextFilter() - { - return _nextFilter; - } - } - - /** - * A continuation ({@link Event}) that takes a Mina sessionClosed event, and passes it to a NextFilter. - * - *

- *
CRC Card
Responsibilities Collaborations - *
Pass a Mina sessionClosed event to a NextFilter. {@link IoFilter.NextFilter}, {@link IoSession} - *
- */ - public static final class CloseEvent extends Event - { - private final IoFilter.NextFilter _nextFilter; - - public CloseEvent(final IoFilter.NextFilter nextFilter) - { - super(); - _nextFilter = nextFilter; - } - - public void process(IoSession session) - { - _nextFilter.sessionClosed(session); - } - - public IoFilter.NextFilter getNextFilter() - { - return _nextFilter; - } - } -} diff --git a/qpid/java/common/src/main/java/org/apache/qpid/pool/Job.java b/qpid/java/common/src/main/java/org/apache/qpid/pool/Job.java index 00da005515..82b600de88 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/pool/Job.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/pool/Job.java @@ -21,9 +21,12 @@ package org.apache.qpid.pool; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.atomic.AtomicBoolean; -import org.apache.mina.common.IoSession; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * A Job is a continuation that batches together other continuations, specifically {@link Event}s, into one continuation. @@ -52,35 +55,28 @@ import org.apache.mina.common.IoSession; */ public class Job implements ReadWriteRunnable { + + /** Defines the maximum number of events that will be batched into a single job. */ + public static final int MAX_JOB_EVENTS = Integer.getInteger("amqj.server.read_write_pool.max_events", 10); + /** The maximum number of events to process per run of the job. More events than this may be queued in the job. */ private final int _maxEvents; - /** The Mina session. */ - private final IoSession _session; - /** Holds the queue of events that make up the job. */ - private final java.util.Queue _eventQueue = new ConcurrentLinkedQueue(); + private final java.util.Queue _eventQueue = new ConcurrentLinkedQueue(); /** Holds a status flag, that indicates when the job is actively running. */ private final AtomicBoolean _active = new AtomicBoolean(); - /** Holds the completion continuation, called upon completion of a run of the job. */ - private final JobCompletionHandler _completionHandler; - private final boolean _readJob; - /** - * Creates a new job that aggregates many continuations together. - * - * @param session The Mina session. - * @param completionHandler The per job run, terminal continuation. - * @param maxEvents The maximum number of aggregated continuations to process per run of the job. - * @param readJob - */ - Job(IoSession session, JobCompletionHandler completionHandler, int maxEvents, final boolean readJob) + private ReferenceCountingExecutorService _poolReference; + + private final static Logger _logger = LoggerFactory.getLogger(Job.class); + + public Job(ReferenceCountingExecutorService poolReference, int maxEvents, boolean readJob) { - _session = session; - _completionHandler = completionHandler; + _poolReference = poolReference; _maxEvents = maxEvents; _readJob = readJob; } @@ -90,7 +86,7 @@ public class Job implements ReadWriteRunnable * * @param evt The continuation to enqueue. */ - void add(Event evt) + public void add(Runnable evt) { _eventQueue.add(evt); } @@ -104,14 +100,14 @@ public class Job implements ReadWriteRunnable int i = _maxEvents; while( --i != 0 ) { - Event e = _eventQueue.poll(); + Runnable e = _eventQueue.poll(); if (e == null) { return true; } else { - e.process(_session); + e.run(); } } return false; @@ -153,40 +149,105 @@ public class Job implements ReadWriteRunnable if(processAll()) { deactivate(); - _completionHandler.completed(_session, this); + completed(); } else { - _completionHandler.notCompleted(_session, this); + notCompleted(); } } - public boolean isReadJob() - { - return _readJob; - } - public boolean isRead() { return _readJob; } - - public boolean isWrite() + + /** + * Adds an {@link Event} to a {@link Job}, triggering the execution of the job if it is not already running. + * + * @param job The job. + * @param event The event to hand off asynchronously. + */ + public static void fireAsynchEvent(ExecutorService pool, Job job, Runnable event) { - return !_readJob; - } + job.add(event); + + + if(pool == null) + { + return; + } + + // rather than perform additional checks on pool to check that it hasn't shutdown. + // catch the RejectedExecutionException that will result from executing on a shutdown pool + if (job.activate()) + { + try + { + pool.execute(job); + } + catch(RejectedExecutionException e) + { + _logger.warn("Thread pool shutdown while tasks still outstanding"); + } + } + } + /** - * Another interface for a continuation. + * Implements a terminal continuation for the {@link Job} for this filter. Whenever the Job completes its processing + * of a batch of events this is called. This method simply re-activates the job, if it has more events to process. * - * @todo Get rid of this interface as there are other interfaces that could be used instead, such as FutureTask, - * Runnable or a custom Continuation interface. + * @param session The Mina session to work in. + * @param job The job that completed. */ - static interface JobCompletionHandler + public void completed() + { + if (!isComplete()) + { + final ExecutorService pool = _poolReference.getPool(); + + if(pool == null) + { + return; + } + + + // ritchiem : 2006-12-13 Do we need to perform the additional checks here? + // Can the pool be shutdown at this point? + if (activate()) + { + try + { + pool.execute(this); + } + catch(RejectedExecutionException e) + { + _logger.warn("Thread pool shutdown while tasks still outstanding"); + } + + } + } + } + + public void notCompleted() { - public void completed(IoSession session, Job job); + final ExecutorService pool = _poolReference.getPool(); + + if(pool == null) + { + return; + } - public void notCompleted(final IoSession session, final Job job); + try + { + pool.execute(this); + } + catch(RejectedExecutionException e) + { + _logger.warn("Thread pool shutdown while tasks still outstanding"); + } } + } diff --git a/qpid/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java b/qpid/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java deleted file mode 100644 index a080cc7e04..0000000000 --- a/qpid/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java +++ /dev/null @@ -1,491 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ -package org.apache.qpid.pool; - -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import org.apache.mina.common.IdleStatus; -import org.apache.mina.common.IoFilterAdapter; -import org.apache.mina.common.IoSession; -import org.apache.qpid.pool.Event.CloseEvent; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.ExecutorService; - -/** - * PoolingFilter, is a no-op pass through filter that hands all events down the Mina filter chain by default. As it - * adds no behaviour by default to the filter chain, it is abstract. - * - *

PoolingFilter provides a capability, available to sub-classes, to handle events in the chain asynchronously, by - * adding them to a job. If a job is not active, adding an event to it activates it. If it is active, the event is - * added to the job, which will run to completion and eventually process the event. The queue on the job itself acts as - * a buffer between stages of the pipeline. - * - *

There are two convenience methods, {@link #createAynschReadPoolingFilter} and - * {@link #createAynschWritePoolingFilter}, for obtaining pooling filters that handle 'messageReceived' and - * 'filterWrite' events, making it possible to process these event streams seperately. - * - *

Pooling filters have a name, in order to distinguish different filter types. They set up a {@link Job} on the - * Mina session they are working with, and store it in the session against their identifying name. This allows different - * filters with different names to be set up on the same filter chain, on the same Mina session, that batch their - * workloads in different jobs. - * - *

- *
CRC Card
Responsibilities Collaborations - *
Implement default, pass through filter. - *
Create pooling filters and a specific thread pool. {@link ReferenceCountingExecutorService} - *
Provide the ability to batch Mina events for asynchronous processing. {@link Job}, {@link Event} - *
Provide a terminal continuation to keep jobs running till empty. - * {@link Job}, {@link Job.JobCompletionHandler} - *
- * - * @todo The static helper methods are pointless. Could just call new. - */ -public abstract class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionHandler -{ - /** Used for debugging purposes. */ - private static final Logger _logger = LoggerFactory.getLogger(PoolingFilter.class); - - /** Holds the managed reference to obtain the executor for the batched jobs. */ - private final ReferenceCountingExecutorService _poolReference; - - /** Used to hold a name for identifying differeny pooling filter types. */ - private final String _name; - - /** Defines the maximum number of events that will be batched into a single job. */ - static final int MAX_JOB_EVENTS = Integer.getInteger("amqj.server.read_write_pool.max_events", 10); - - private final int _maxEvents; - - private final boolean _readFilter; - - /** - * Creates a named pooling filter, on the specified shared thread pool. - * - * @param refCountingPool The thread pool reference. - * @param name The identifying name of the filter type. - */ - public PoolingFilter(ReferenceCountingExecutorService refCountingPool, String name, int maxEvents, boolean readFilter) - { - _poolReference = refCountingPool; - _name = name; - _maxEvents = maxEvents; - _readFilter = readFilter; - } - - /** - * Helper method to get an instance of a pooling filter that handles read events asynchronously. - * - * @param refCountingPool A managed reference to the thread pool. - * @param name The filter types identifying name. - * - * @return A pooling filter for asynchronous read events. - */ - public static PoolingFilter createAynschReadPoolingFilter(ReferenceCountingExecutorService refCountingPool, String name) - { - return new AsynchReadPoolingFilter(refCountingPool, name); - } - - /** - * Helper method to get an instance of a pooling filter that handles write events asynchronously. - * - * @param refCountingPool A managed reference to the thread pool. - * @param name The filter types identifying name. - * - * @return A pooling filter for asynchronous write events. - */ - public static PoolingFilter createAynschWritePoolingFilter(ReferenceCountingExecutorService refCountingPool, String name) - { - return new AsynchWritePoolingFilter(refCountingPool, name); - } - - /** - * Called by Mina to initialize this filter. Takes a reference to the thread pool. - */ - public void init() - { - _logger.debug("Init called on PoolingFilter " + toString()); - - // Called when the filter is initialised in the chain. If the reference count is - // zero this acquire will initialise the pool. - _poolReference.acquireExecutorService(); - } - - /** - * Called by Mina to clean up this filter. Releases the reference to the thread pool. - */ - public void destroy() - { - _logger.debug("Destroy called on PoolingFilter " + toString()); - - // When the reference count gets to zero we release the executor service. - _poolReference.releaseExecutorService(); - } - - /** - * Adds an {@link Event} to a {@link Job}, triggering the execution of the job if it is not already running. - * - * @param job The job. - * @param event The event to hand off asynchronously. - */ - void fireAsynchEvent(Job job, Event event) - { - - job.add(event); - - final ExecutorService pool = _poolReference.getPool(); - - if(pool == null) - { - return; - } - - // rather than perform additional checks on pool to check that it hasn't shutdown. - // catch the RejectedExecutionException that will result from executing on a shutdown pool - if (job.activate()) - { - try - { - pool.execute(job); - } - catch(RejectedExecutionException e) - { - _logger.warn("Thread pool shutdown while tasks still outstanding"); - } - } - - } - - /** - * Creates a Job on the Mina session, identified by this filters name, in which this filter places asynchronously - * handled events. - * - * @param session The Mina session. - */ - public void createNewJobForSession(IoSession session) - { - Job job = new Job(session, this, MAX_JOB_EVENTS,_readFilter); - session.setAttribute(_name, job); - } - - /** - * Retrieves this filters Job, by this filters name, from the Mina session. - * - * @param session The Mina session. - * - * @return The Job for this filter to place asynchronous events into. - */ - public Job getJobForSession(IoSession session) - { - return (Job) session.getAttribute(_name); - } - - /** - * Implements a terminal continuation for the {@link Job} for this filter. Whenever the Job completes its processing - * of a batch of events this is called. This method simply re-activates the job, if it has more events to process. - * - * @param session The Mina session to work in. - * @param job The job that completed. - */ - public void completed(IoSession session, Job job) - { - - - if (!job.isComplete()) - { - final ExecutorService pool = _poolReference.getPool(); - - if(pool == null) - { - return; - } - - - // ritchiem : 2006-12-13 Do we need to perform the additional checks here? - // Can the pool be shutdown at this point? - if (job.activate()) - { - try - { - pool.execute(job); - } - catch(RejectedExecutionException e) - { - _logger.warn("Thread pool shutdown while tasks still outstanding"); - } - - } - } - } - - public void notCompleted(IoSession session, Job job) - { - final ExecutorService pool = _poolReference.getPool(); - - if(pool == null) - { - return; - } - - try - { - pool.execute(job); - } - catch(RejectedExecutionException e) - { - _logger.warn("Thread pool shutdown while tasks still outstanding"); - } - - } - - - - /** - * No-op pass through filter to the next filter in the chain. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - * - * @throws Exception This method does not throw any exceptions, but has Exception in its signature to allow - * overriding sub-classes the ability to. - */ - public void sessionOpened(final NextFilter nextFilter, final IoSession session) throws Exception - { - nextFilter.sessionOpened(session); - } - - /** - * No-op pass through filter to the next filter in the chain. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - * - * @throws Exception This method does not throw any exceptions, but has Exception in its signature to allow - * overriding sub-classes the ability to. - */ - public void sessionClosed(final NextFilter nextFilter, final IoSession session) throws Exception - { - nextFilter.sessionClosed(session); - } - - /** - * No-op pass through filter to the next filter in the chain. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - * @param status The session idle status. - * - * @throws Exception This method does not throw any exceptions, but has Exception in its signature to allow - * overriding sub-classes the ability to. - */ - public void sessionIdle(final NextFilter nextFilter, final IoSession session, final IdleStatus status) throws Exception - { - nextFilter.sessionIdle(session, status); - } - - /** - * No-op pass through filter to the next filter in the chain. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - * @param cause The underlying exception. - * - * @throws Exception This method does not throw any exceptions, but has Exception in its signature to allow - * overriding sub-classes the ability to. - */ - public void exceptionCaught(final NextFilter nextFilter, final IoSession session, final Throwable cause) throws Exception - { - nextFilter.exceptionCaught(session, cause); - } - - /** - * No-op pass through filter to the next filter in the chain. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - * @param message The message received. - * - * @throws Exception This method does not throw any exceptions, but has Exception in its signature to allow - * overriding sub-classes the ability to. - */ - public void messageReceived(final NextFilter nextFilter, final IoSession session, final Object message) throws Exception - { - nextFilter.messageReceived(session, message); - } - - /** - * No-op pass through filter to the next filter in the chain. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - * @param message The message sent. - * - * @throws Exception This method does not throw any exceptions, but has Exception in its signature to allow - * overriding sub-classes the ability to. - */ - public void messageSent(final NextFilter nextFilter, final IoSession session, final Object message) throws Exception - { - nextFilter.messageSent(session, message); - } - - /** - * No-op pass through filter to the next filter in the chain. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - * @param writeRequest The write request event. - * - * @throws Exception This method does not throw any exceptions, but has Exception in its signature to allow - * overriding sub-classes the ability to. - */ - public void filterWrite(final NextFilter nextFilter, final IoSession session, final WriteRequest writeRequest) - throws Exception - { - nextFilter.filterWrite(session, writeRequest); - } - - /** - * No-op pass through filter to the next filter in the chain. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - * - * @throws Exception This method does not throw any exceptions, but has Exception in its signature to allow - * overriding sub-classes the ability to. - */ - public void filterClose(NextFilter nextFilter, IoSession session) throws Exception - { - nextFilter.filterClose(session); - } - - /** - * No-op pass through filter to the next filter in the chain. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - * - * @throws Exception This method does not throw any exceptions, but has Exception in its signature to allow - * overriding sub-classes the ability to. - */ - public void sessionCreated(NextFilter nextFilter, IoSession session) throws Exception - { - nextFilter.sessionCreated(session); - } - - /** - * Prints the filter types identifying name to a string, mainly for debugging purposes. - * - * @return The filter types identifying name. - */ - public String toString() - { - return _name; - } - - /** - * AsynchReadPoolingFilter is a pooling filter that handles 'messageReceived' and 'sessionClosed' events - * asynchronously. - */ - public static class AsynchReadPoolingFilter extends PoolingFilter - { - /** - * Creates a pooling filter that handles read events asynchronously. - * - * @param refCountingPool A managed reference to the thread pool. - * @param name The filter types identifying name. - */ - public AsynchReadPoolingFilter(ReferenceCountingExecutorService refCountingPool, String name) - { - super(refCountingPool, name, Integer.getInteger("amqj.server.read_write_pool.max_read_events", MAX_JOB_EVENTS),true); - } - - /** - * Hands off this event for asynchronous execution. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - * @param message The message received. - */ - public void messageReceived(NextFilter nextFilter, final IoSession session, Object message) - { - Job job = getJobForSession(session); - fireAsynchEvent(job, new Event.ReceivedEvent(nextFilter, message)); - } - - /** - * Hands off this event for asynchronous execution. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - */ - public void sessionClosed(final NextFilter nextFilter, final IoSession session) - { - Job job = getJobForSession(session); - fireAsynchEvent(job, new CloseEvent(nextFilter)); - } - } - - /** - * AsynchWritePoolingFilter is a pooling filter that handles 'filterWrite' and 'sessionClosed' events - * asynchronously. - */ - public static class AsynchWritePoolingFilter extends PoolingFilter - { - /** - * Creates a pooling filter that handles write events asynchronously. - * - * @param refCountingPool A managed reference to the thread pool. - * @param name The filter types identifying name. - */ - public AsynchWritePoolingFilter(ReferenceCountingExecutorService refCountingPool, String name) - { - super(refCountingPool, name, Integer.getInteger("amqj.server.read_write_pool.max_write_events", MAX_JOB_EVENTS),false); - } - - /** - * Hands off this event for asynchronous execution. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - * @param writeRequest The write request event. - */ - public void filterWrite(final NextFilter nextFilter, final IoSession session, final WriteRequest writeRequest) - { - Job job = getJobForSession(session); - fireAsynchEvent(job, new Event.WriteEvent(nextFilter, writeRequest)); - } - - /** - * Hands off this event for asynchronous execution. - * - * @param nextFilter The next filter in the chain. - * @param session The Mina session. - */ - public void sessionClosed(final NextFilter nextFilter, final IoSession session) - { - Job job = getJobForSession(session); - fireAsynchEvent(job, new CloseEvent(nextFilter)); - } - } -} diff --git a/qpid/java/common/src/main/java/org/apache/qpid/pool/ReadWriteRunnable.java b/qpid/java/common/src/main/java/org/apache/qpid/pool/ReadWriteRunnable.java index ad04a923e1..140c93ca8d 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/pool/ReadWriteRunnable.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/pool/ReadWriteRunnable.java @@ -23,5 +23,4 @@ package org.apache.qpid.pool; public interface ReadWriteRunnable extends Runnable { boolean isRead(); - boolean isWrite(); } diff --git a/qpid/java/common/src/main/java/org/apache/qpid/pool/ReadWriteThreadModel.java b/qpid/java/common/src/main/java/org/apache/qpid/pool/ReadWriteThreadModel.java deleted file mode 100644 index 8cea70e597..0000000000 --- a/qpid/java/common/src/main/java/org/apache/qpid/pool/ReadWriteThreadModel.java +++ /dev/null @@ -1,102 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ -package org.apache.qpid.pool; - -import org.apache.mina.common.IoFilterChain; -import org.apache.mina.common.ThreadModel; -import org.apache.mina.filter.ReferenceCountingIoFilter; - -/** - * ReadWriteThreadModel is a Mina i/o filter chain factory, which creates a filter chain with seperate filters to - * handle read and write events. The seperate filters are {@link PoolingFilter}s, which have thread pools to handle - * these events. The effect of this is that reading and writing may happen concurrently. - * - *

Socket i/o will only happen with concurrent reads and writes if Mina has seperate selector threads for each. - * - *

- *
CRC Card
Responsibilities Collaborations - *
Create a filter chain with seperate read and write thread pools for read/write Mina events. - * {@link PoolingFilter} - *
- */ -public class ReadWriteThreadModel implements ThreadModel -{ - /** Holds the singleton instance of this factory. */ - private static final ReadWriteThreadModel _instance = new ReadWriteThreadModel(); - - /** Holds the thread pooling filter for reads. */ - private final PoolingFilter _asynchronousReadFilter; - - /** Holds the thread pooloing filter for writes. */ - private final PoolingFilter _asynchronousWriteFilter; - - /** - * Creates a new factory for concurrent i/o, thread pooling filter chain construction. This is private, so that - * only a singleton instance of the factory is ever created. - */ - private ReadWriteThreadModel() - { - final ReferenceCountingExecutorService executor = ReferenceCountingExecutorService.getInstance(); - _asynchronousReadFilter = PoolingFilter.createAynschReadPoolingFilter(executor, "AsynchronousReadFilter"); - _asynchronousWriteFilter = PoolingFilter.createAynschWritePoolingFilter(executor, "AsynchronousWriteFilter"); - } - - /** - * Gets the singleton instance of this filter chain factory. - * - * @return The singleton instance of this filter chain factory. - */ - public static ReadWriteThreadModel getInstance() - { - return _instance; - } - - /** - * Gets the read filter. - * - * @return The read filter. - */ - public PoolingFilter getAsynchronousReadFilter() - { - return _asynchronousReadFilter; - } - - /** - * Gets the write filter. - * - * @return The write filter. - */ - public PoolingFilter getAsynchronousWriteFilter() - { - return _asynchronousWriteFilter; - } - - /** - * Adds the concurrent read and write filters to a filter chain. - * - * @param chain The Mina filter chain to add to. - */ - public void buildFilterChain(IoFilterChain chain) - { - chain.addFirst("AsynchronousReadFilter", new ReferenceCountingIoFilter(_asynchronousReadFilter)); - chain.addLast("AsynchronousWriteFilter", new ReferenceCountingIoFilter(_asynchronousWriteFilter)); - } -} diff --git a/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java b/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java new file mode 100644 index 0000000000..5bfc189b02 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java @@ -0,0 +1,65 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.protocol; + +import java.net.SocketAddress; + +import org.apache.qpid.framing.AMQDataBlock; +import org.apache.qpid.transport.NetworkDriver; +import org.apache.qpid.transport.Receiver; + +/** + * A ProtocolEngine is a Receiver for java.nio.ByteBuffers. It takes the data passed to it in the received + * decodes it and then process the result. + */ +public interface ProtocolEngine extends Receiver +{ + // Sets the network driver providing data for this ProtocolEngine + void setNetworkDriver (NetworkDriver driver); + + // Returns the remote address of the NetworkDriver + SocketAddress getRemoteAddress(); + + // Returns the local address of the NetworkDriver + SocketAddress getLocalAddress(); + + // Returns number of bytes written + long getWrittenBytes(); + + // Returns number of bytes read + long getReadBytes(); + + // Called by the NetworkDriver when the socket has been closed for reading + void closed(); + + // Called when the NetworkEngine has not written data for the specified period of time (will trigger a + // heartbeat) + void writerIdle(); + + // Called when the NetworkEngine has not read data for the specified period of time (will close the connection) + void readerIdle(); + + /** + * Accepts an AMQFrame for writing to the network. The ProtocolEngine encodes the frame into bytes and + * passes the data onto the NetworkDriver for sending + */ + void writeFrame(AMQDataBlock frame); +} \ No newline at end of file diff --git a/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java b/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java new file mode 100644 index 0000000000..9df84eef90 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java @@ -0,0 +1,31 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.protocol; + +import org.apache.qpid.transport.NetworkDriver; + +public interface ProtocolEngineFactory +{ + + // Returns a new instance of a ProtocolEngine + ProtocolEngine newProtocolEngine(NetworkDriver networkDriver); + +} \ No newline at end of file diff --git a/qpid/java/common/src/main/java/org/apache/qpid/thread/QpidThreadExecutor.java b/qpid/java/common/src/main/java/org/apache/qpid/thread/QpidThreadExecutor.java new file mode 100644 index 0000000000..376658bb99 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/qpid/thread/QpidThreadExecutor.java @@ -0,0 +1,43 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +package org.apache.qpid.thread; + +import org.apache.qpid.thread.Threading; + +import edu.emory.mathcs.backport.java.util.concurrent.Executor; + +public class QpidThreadExecutor implements Executor +{ + @Override + public void execute(Runnable command) + { + try + { + Threading.getThreadFactory().createThread(command).start(); + } + catch(Exception e) + { + throw new RuntimeException("Error creating a thread using Qpid thread factory",e); + } + } + +} diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkDriver.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkDriver.java new file mode 100644 index 0000000000..86af97bf7e --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkDriver.java @@ -0,0 +1,63 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.transport; + +import java.net.BindException; +import java.net.InetAddress; +import java.net.SocketAddress; + +import org.apache.qpid.protocol.ProtocolEngine; +import org.apache.qpid.protocol.ProtocolEngineFactory; +import org.apache.qpid.ssl.SSLContextFactory; + +public interface NetworkDriver extends Sender +{ + // Creates a NetworkDriver which attempts to connect to destination on port and attaches the ProtocolEngine to + // it using the SSLContextFactory if provided + void open(int port, InetAddress destination, ProtocolEngine engine, + NetworkDriverConfiguration config, SSLContextFactory sslFactory) + throws OpenException; + + // listens for incoming connections on the specified ports and address and creates a new NetworkDriver which + // processes incoming connections with ProtocolEngines and SSLEngines created from the factories + // (in the case of an SSLContextFactory, if provided) + void bind (int port, InetAddress[] addresses, ProtocolEngineFactory protocolFactory, + NetworkDriverConfiguration config, SSLContextFactory sslFactory) throws BindException; + + // Returns the remote address of the underlying socket + SocketAddress getRemoteAddress(); + + // Returns the local address of the underlying socket + SocketAddress getLocalAddress(); + + /** + * The length of time after which the ProtocolEngines readIdle() method should be called if no data has been + * read in seconds + */ + void setMaxReadIdle(int idleTime); + + /** + * The length of time after which the ProtocolEngines writeIdle() method should be called if no data has been + * written in seconds + */ + void setMaxWriteIdle(int idleTime); + +} \ No newline at end of file diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkDriverConfiguration.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkDriverConfiguration.java new file mode 100644 index 0000000000..c38afe5dd5 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkDriverConfiguration.java @@ -0,0 +1,44 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.transport; + +/** + * This interface provides a means for NetworkDrivers to configure TCP options such as incoming and outgoing + * buffer sizes and set particular options on the socket. NetworkDrivers should honour the values returned + * from here if the underlying implementation supports them. + */ +public interface NetworkDriverConfiguration +{ + // Taken from Socket + Boolean getKeepAlive(); + Boolean getOOBInline(); + Boolean getReuseAddress(); + Integer getSoLinger(); // null means off + Integer getSoTimeout(); + Boolean getTcpNoDelay(); + Integer getTrafficClass(); + + // The amount of memory in bytes to allocate to the incoming buffer + Integer getReceiveBufferSize(); + + // The amount of memory in bytes to allocate to the outgoing buffer + Integer getSendBufferSize(); +} diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/OpenException.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/OpenException.java new file mode 100644 index 0000000000..68fbb5e8ec --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/OpenException.java @@ -0,0 +1,34 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +package org.apache.qpid.transport; + +import java.io.IOException; + +public class OpenException extends IOException +{ + + public OpenException(String string, Throwable lastException) + { + super(string, lastException); + } + +} diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/mina/MINANetworkDriver.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/mina/MINANetworkDriver.java new file mode 100644 index 0000000000..38ea9307b7 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/mina/MINANetworkDriver.java @@ -0,0 +1,418 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +package org.apache.qpid.transport.network.mina; + +import java.io.IOException; +import java.net.BindException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.nio.ByteBuffer; + +import org.apache.mina.common.ConnectFuture; +import org.apache.mina.common.IdleStatus; +import org.apache.mina.common.IoAcceptor; +import org.apache.mina.common.IoConnector; +import org.apache.mina.common.IoFilterChain; +import org.apache.mina.common.IoHandlerAdapter; +import org.apache.mina.common.IoSession; +import org.apache.mina.common.SimpleByteBufferAllocator; +import org.apache.mina.common.WriteFuture; +import org.apache.mina.filter.ReadThrottleFilterBuilder; +import org.apache.mina.filter.SSLFilter; +import org.apache.mina.filter.WriteBufferLimitFilterBuilder; +import org.apache.mina.filter.executor.ExecutorFilter; +import org.apache.mina.transport.socket.nio.MultiThreadSocketConnector; +import org.apache.mina.transport.socket.nio.SocketAcceptorConfig; +import org.apache.mina.transport.socket.nio.SocketConnector; +import org.apache.mina.transport.socket.nio.SocketConnectorConfig; +import org.apache.mina.transport.socket.nio.SocketSessionConfig; +import org.apache.mina.util.NewThreadExecutor; +import org.apache.mina.util.SessionUtil; +import org.apache.qpid.protocol.ProtocolEngine; +import org.apache.qpid.protocol.ProtocolEngineFactory; +import org.apache.qpid.ssl.SSLContextFactory; +import org.apache.qpid.thread.QpidThreadExecutor; +import org.apache.qpid.transport.NetworkDriver; +import org.apache.qpid.transport.NetworkDriverConfiguration; +import org.apache.qpid.transport.OpenException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class MINANetworkDriver extends IoHandlerAdapter implements NetworkDriver +{ + + private static final int DEFAULT_BUFFER_SIZE = 32 * 1024; + + ProtocolEngine _protocolEngine; + private boolean _useNIO = false; + private int _processors = 4; + private boolean _executorPool = false; + private SSLContextFactory _sslFactory = null; + private IoConnector _socketConnector; + private IoAcceptor _acceptor; + private IoSession _ioSession; + private ProtocolEngineFactory _factory; + private boolean _protectIO; + private NetworkDriverConfiguration _config; + private Throwable _lastException; + private boolean _acceptingConnections = false; + + private WriteFuture _lastWriteFuture; + + private static final Logger _logger = LoggerFactory.getLogger(MINANetworkDriver.class); + + public MINANetworkDriver(boolean useNIO, int processors, boolean executorPool, boolean protectIO) + { + _useNIO = useNIO; + _processors = processors; + _executorPool = executorPool; + _protectIO = protectIO; + } + + public MINANetworkDriver(boolean useNIO, int processors, boolean executorPool, boolean protectIO, + ProtocolEngine protocolEngine, IoSession session) + { + _useNIO = useNIO; + _processors = processors; + _executorPool = executorPool; + _protectIO = protectIO; + _protocolEngine = protocolEngine; + _ioSession = session; + _ioSession.setAttachment(_protocolEngine); + } + + public MINANetworkDriver() + { + + } + + public MINANetworkDriver(IoConnector ioConnector) + { + _socketConnector = ioConnector; + } + + public MINANetworkDriver(IoConnector ioConnector, ProtocolEngine engine) + { + _socketConnector = ioConnector; + _protocolEngine = engine; + } + + public void bind(int port, InetAddress[] addresses, ProtocolEngineFactory factory, + NetworkDriverConfiguration config, SSLContextFactory sslFactory) throws BindException + { + + _factory = factory; + _config = config; + + if (_useNIO) + { + _acceptor = new org.apache.mina.transport.socket.nio.MultiThreadSocketAcceptor(_processors, + new NewThreadExecutor()); + } + else + { + _acceptor = new org.apache.mina.transport.socket.nio.SocketAcceptor(_processors, new NewThreadExecutor()); + } + + SocketAcceptorConfig sconfig = (SocketAcceptorConfig) _acceptor.getDefaultConfig(); + SocketSessionConfig sc = (SocketSessionConfig) sconfig.getSessionConfig(); + + if (config != null) + { + sc.setReceiveBufferSize(config.getReceiveBufferSize()); + sc.setSendBufferSize(config.getSendBufferSize()); + sc.setTcpNoDelay(config.getTcpNoDelay()); + } + + if (sslFactory != null) + { + _sslFactory = sslFactory; + } + + if (addresses != null && addresses.length > 0) + { + for (InetAddress addr : addresses) + { + try + { + _acceptor.bind(new InetSocketAddress(addr, port), this, sconfig); + } + catch (IOException e) + { + throw new BindException(String.format("Could not bind to {0}:{2}", addr, port)); + } + } + } + else + { + try + { + _acceptor.bind(new InetSocketAddress(port), this, sconfig); + } + catch (IOException e) + { + throw new BindException(String.format("Could not bind to *:{1}", port)); + } + } + _acceptingConnections = true; + } + + public SocketAddress getRemoteAddress() + { + return _ioSession.getRemoteAddress(); + } + + public SocketAddress getLocalAddress() + { + return _ioSession.getLocalAddress(); + } + + + public void open(int port, InetAddress destination, ProtocolEngine engine, NetworkDriverConfiguration config, + SSLContextFactory sslFactory) throws OpenException + { + if (sslFactory != null) + { + _sslFactory = sslFactory; + } + + if (_useNIO) + { + _socketConnector = new MultiThreadSocketConnector(1, new QpidThreadExecutor()); + } + else + { + _socketConnector = new SocketConnector(1, new QpidThreadExecutor()); // non-blocking + // connector + } + + org.apache.mina.common.ByteBuffer.setUseDirectBuffers(Boolean.getBoolean("amqj.enableDirectBuffers")); + // the MINA default is currently to use the pooled allocator although this may change in future + // once more testing of the performance of the simple allocator has been done + if (!Boolean.getBoolean("amqj.enablePooledAllocator")) + { + org.apache.mina.common.ByteBuffer.setAllocator(new SimpleByteBufferAllocator()); + } + + SocketConnectorConfig cfg = (SocketConnectorConfig) _socketConnector.getDefaultConfig(); + + SocketSessionConfig scfg = (SocketSessionConfig) cfg.getSessionConfig(); + scfg.setTcpNoDelay((config != null) ? config.getTcpNoDelay() : true); + scfg.setSendBufferSize((config != null) ? config.getSendBufferSize() : DEFAULT_BUFFER_SIZE); + scfg.setReceiveBufferSize((config != null) ? config.getReceiveBufferSize() : DEFAULT_BUFFER_SIZE); + + // Don't have the connector's worker thread wait around for other + // connections (we only use + // one SocketConnector per connection at the moment anyway). This allows + // short-running + // clients (like unit tests) to complete quickly. + if (_socketConnector instanceof SocketConnector) + { + ((SocketConnector) _socketConnector).setWorkerTimeout(0); + } + + ConnectFuture future = _socketConnector.connect(new InetSocketAddress(destination, port), this, cfg); + future.join(); + if (!future.isConnected()) + { + throw new OpenException("Could not open connection", _lastException); + } + _ioSession = future.getSession(); + _ioSession.setAttachment(engine); + engine.setNetworkDriver(this); + _protocolEngine = engine; + } + + public void setMaxReadIdle(int idleTime) + { + _ioSession.setIdleTime(IdleStatus.READER_IDLE, idleTime); + } + + public void setMaxWriteIdle(int idleTime) + { + _ioSession.setIdleTime(IdleStatus.WRITER_IDLE, idleTime); + } + + public void close() + { + if (_lastWriteFuture != null) + { + _lastWriteFuture.join(); + } + if (_acceptor != null) + { + _acceptor.unbindAll(); + } + if (_ioSession != null) + { + _ioSession.close(); + } + } + + public void flush() + { + if (_lastWriteFuture != null) + { + _lastWriteFuture.join(); + } + } + + public void send(ByteBuffer msg) + { + _lastWriteFuture = _ioSession.write(org.apache.mina.common.ByteBuffer.wrap(msg)); + } + + public void setIdleTimeout(long l) + { + // MINA doesn't support setting SO_TIMEOUT + } + + public void exceptionCaught(IoSession protocolSession, Throwable throwable) throws Exception + { + if (_protocolEngine != null) + { + _protocolEngine.exception(throwable); + } + else + { + _logger.error("Exception thrown and no ProtocolEngine to handle it", throwable); + } + _lastException = throwable; + } + + /** + * Invoked when a message is received on a particular protocol session. Note + * that a protocol session is directly tied to a particular physical + * connection. + * + * @param protocolSession + * the protocol session that received the message + * @param message + * the message itself (i.e. a decoded frame) + * + * @throws Exception + * if the message cannot be processed + */ + public void messageReceived(IoSession protocolSession, Object message) throws Exception + { + if (message instanceof org.apache.mina.common.ByteBuffer) + { + ((ProtocolEngine) protocolSession.getAttachment()).received(((org.apache.mina.common.ByteBuffer) message).buf()); + } + else + { + throw new IllegalStateException("Handed unhandled message. message.class = " + message.getClass() + " message = " + message); + } + } + + public void sessionClosed(IoSession protocolSession) throws Exception + { + ((ProtocolEngine) protocolSession.getAttachment()).closed(); + } + + public void sessionCreated(IoSession protocolSession) throws Exception + { + // Configure the session with SSL if necessary + SessionUtil.initialize(protocolSession); + if (_executorPool) + { + if (_sslFactory != null) + { + protocolSession.getFilterChain().addAfter("AsynchronousReadFilter", "sslFilter", + new SSLFilter(_sslFactory.buildServerContext())); + } + } + else + { + if (_sslFactory != null) + { + protocolSession.getFilterChain().addBefore("protocolFilter", "sslFilter", + new SSLFilter(_sslFactory.buildServerContext())); + } + } + // Do we want to have read/write buffer limits? + if (_protectIO) + { + //Add IO Protection Filters + IoFilterChain chain = protocolSession.getFilterChain(); + + protocolSession.getFilterChain().addLast("tempExecutorFilterForFilterBuilder", new ExecutorFilter()); + + ReadThrottleFilterBuilder readfilter = new ReadThrottleFilterBuilder(); + readfilter.setMaximumConnectionBufferSize(_config.getReceiveBufferSize()); + readfilter.attach(chain); + + WriteBufferLimitFilterBuilder writefilter = new WriteBufferLimitFilterBuilder(); + writefilter.setMaximumConnectionBufferSize(_config.getSendBufferSize()); + writefilter.attach(chain); + + protocolSession.getFilterChain().remove("tempExecutorFilterForFilterBuilder"); + } + + if (_ioSession == null) + { + _ioSession = protocolSession; + } + + if (_acceptingConnections) + { + // Set up the protocol engine + ProtocolEngine protocolEngine = _factory.newProtocolEngine(this); + MINANetworkDriver newDriver = new MINANetworkDriver(_useNIO, _processors, _executorPool, _protectIO, protocolEngine, protocolSession); + protocolEngine.setNetworkDriver(newDriver); + } + } + + public void sessionIdle(IoSession session, IdleStatus status) throws Exception + { + if (IdleStatus.WRITER_IDLE.equals(status)) + { + ((ProtocolEngine) session.getAttachment()).writerIdle(); + } + else if (IdleStatus.READER_IDLE.equals(status)) + { + ((ProtocolEngine) session.getAttachment()).readerIdle(); + } + } + + private ProtocolEngine getProtocolEngine() + { + return _protocolEngine; + } + + public void setProtocolEngineFactory(ProtocolEngineFactory engineFactory, boolean acceptingConnections) + { + _factory = engineFactory; + _acceptingConnections = acceptingConnections; + } + + public void setProtocolEngine(ProtocolEngine protocolEngine) + { + _protocolEngine = protocolEngine; + if (_ioSession != null) + { + _ioSession.setAttachment(protocolEngine); + } + } + +} diff --git a/qpid/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java b/qpid/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java new file mode 100644 index 0000000000..46c812e265 --- /dev/null +++ b/qpid/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java @@ -0,0 +1,130 @@ +package org.apache.qpid.codec; + +import java.nio.ByteBuffer; +import java.util.ArrayList; + +import junit.framework.TestCase; + +import org.apache.qpid.framing.AMQDataBlock; +import org.apache.qpid.framing.AMQFrame; +import org.apache.qpid.framing.AMQFrameDecodingException; +import org.apache.qpid.framing.AMQProtocolVersionException; +import org.apache.qpid.framing.HeartbeatBody; + +public class AMQDecoderTest extends TestCase +{ + + private AMQCodecFactory _factory; + private AMQDecoder _decoder; + + + public void setUp() + { + _factory = new AMQCodecFactory(false, null); + _decoder = _factory.getDecoder(); + } + + + public void testSingleFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException + { + ByteBuffer msg = HeartbeatBody.FRAME.toNioByteBuffer(); + ArrayList frames = _decoder.decodeBuffer(msg); + if (frames.get(0) instanceof AMQFrame) + { + assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frames.get(0)).getBodyFrame().getFrameType()); + } + else + { + fail("decode was not a frame"); + } + } + + public void testPartialFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException + { + ByteBuffer msg = HeartbeatBody.FRAME.toNioByteBuffer(); + ByteBuffer msgA = msg.slice(); + int msgbPos = msg.remaining() / 2; + int msgaLimit = msg.remaining() - msgbPos; + msgA.limit(msgaLimit); + msg.position(msgbPos); + ByteBuffer msgB = msg.slice(); + ArrayList frames = _decoder.decodeBuffer(msgA); + assertEquals(0, frames.size()); + frames = _decoder.decodeBuffer(msgB); + assertEquals(1, frames.size()); + if (frames.get(0) instanceof AMQFrame) + { + assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frames.get(0)).getBodyFrame().getFrameType()); + } + else + { + fail("decode was not a frame"); + } + } + + public void testMultipleFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException + { + ByteBuffer msgA = HeartbeatBody.FRAME.toNioByteBuffer(); + ByteBuffer msgB = HeartbeatBody.FRAME.toNioByteBuffer(); + ByteBuffer msg = ByteBuffer.allocate(msgA.remaining() + msgB.remaining()); + msg.put(msgA); + msg.put(msgB); + msg.flip(); + ArrayList frames = _decoder.decodeBuffer(msg); + assertEquals(2, frames.size()); + for (AMQDataBlock frame : frames) + { + if (frame instanceof AMQFrame) + { + assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frame).getBodyFrame().getFrameType()); + } + else + { + fail("decode was not a frame"); + } + } + } + + public void testMultiplePartialFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException + { + ByteBuffer msgA = HeartbeatBody.FRAME.toNioByteBuffer(); + ByteBuffer msgB = HeartbeatBody.FRAME.toNioByteBuffer(); + ByteBuffer msgC = HeartbeatBody.FRAME.toNioByteBuffer(); + + ByteBuffer sliceA = ByteBuffer.allocate(msgA.remaining() + msgB.remaining() / 2); + sliceA.put(msgA); + int limit = msgB.limit(); + int pos = msgB.remaining() / 2; + msgB.limit(pos); + sliceA.put(msgB); + sliceA.flip(); + msgB.limit(limit); + msgB.position(pos); + + ByteBuffer sliceB = ByteBuffer.allocate(msgB.remaining() + pos); + sliceB.put(msgB); + msgC.limit(pos); + sliceB.put(msgC); + sliceB.flip(); + msgC.limit(limit); + + ArrayList frames = _decoder.decodeBuffer(sliceA); + assertEquals(1, frames.size()); + frames = _decoder.decodeBuffer(sliceB); + assertEquals(1, frames.size()); + frames = _decoder.decodeBuffer(msgC); + assertEquals(1, frames.size()); + for (AMQDataBlock frame : frames) + { + if (frame instanceof AMQFrame) + { + assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frame).getBodyFrame().getFrameType()); + } + else + { + fail("decode was not a frame"); + } + } + } + +} diff --git a/qpid/java/common/src/test/java/org/apache/qpid/codec/MockAMQVersionAwareProtocolSession.java b/qpid/java/common/src/test/java/org/apache/qpid/codec/MockAMQVersionAwareProtocolSession.java new file mode 100644 index 0000000000..bd7fb68d93 --- /dev/null +++ b/qpid/java/common/src/test/java/org/apache/qpid/codec/MockAMQVersionAwareProtocolSession.java @@ -0,0 +1,95 @@ +package org.apache.qpid.codec; + +import java.nio.ByteBuffer; + +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.AMQDataBlock; +import org.apache.qpid.framing.AMQMethodBody; +import org.apache.qpid.framing.ContentBody; +import org.apache.qpid.framing.ContentHeaderBody; +import org.apache.qpid.framing.HeartbeatBody; +import org.apache.qpid.framing.MethodRegistry; +import org.apache.qpid.framing.ProtocolVersion; +import org.apache.qpid.protocol.AMQVersionAwareProtocolSession; +import org.apache.qpid.transport.Sender; + +public class MockAMQVersionAwareProtocolSession implements AMQVersionAwareProtocolSession +{ + + @Override + public void contentBodyReceived(int channelId, ContentBody body) throws AMQException + { + // TODO Auto-generated method stub + + } + + @Override + public void contentHeaderReceived(int channelId, ContentHeaderBody body) throws AMQException + { + // TODO Auto-generated method stub + + } + + @Override + public MethodRegistry getMethodRegistry() + { + return MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9); + } + + @Override + public void heartbeatBodyReceived(int channelId, HeartbeatBody body) throws AMQException + { + // TODO Auto-generated method stub + + } + + @Override + public void init() + { + // TODO Auto-generated method stub + + } + + @Override + public void methodFrameReceived(int channelId, AMQMethodBody body) throws AMQException + { + // TODO Auto-generated method stub + + } + + @Override + public void setSender(Sender sender) + { + // TODO Auto-generated method stub + + } + + @Override + public void writeFrame(AMQDataBlock frame) + { + // TODO Auto-generated method stub + + } + + @Override + public byte getProtocolMajorVersion() + { + // TODO Auto-generated method stub + return 0; + } + + @Override + public byte getProtocolMinorVersion() + { + // TODO Auto-generated method stub + return 0; + } + + @Override + public ProtocolVersion getProtocolVersion() + { + // TODO Auto-generated method stub + return null; + } + +} diff --git a/qpid/java/common/src/test/java/org/apache/qpid/pool/PoolingFilterTest.java b/qpid/java/common/src/test/java/org/apache/qpid/pool/PoolingFilterTest.java deleted file mode 100644 index 6383d52298..0000000000 --- a/qpid/java/common/src/test/java/org/apache/qpid/pool/PoolingFilterTest.java +++ /dev/null @@ -1,111 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - * - */ -package org.apache.qpid.pool; - -import junit.framework.TestCase; -import junit.framework.Assert; -import org.apache.qpid.session.TestSession; -import org.apache.mina.common.IoFilter; -import org.apache.mina.common.IoSession; -import org.apache.mina.common.IdleStatus; - -import java.util.concurrent.RejectedExecutionException; - -public class PoolingFilterTest extends TestCase -{ - private PoolingFilter _pool; - ReferenceCountingExecutorService _executorService; - - public void setUp() - { - - //Create Pool - _executorService = ReferenceCountingExecutorService.getInstance(); - _executorService.acquireExecutorService(); - _pool = PoolingFilter.createAynschWritePoolingFilter(_executorService, - "AsynchronousWriteFilter"); - - } - - public void testRejectedExecution() throws Exception - { - - TestSession testSession = new TestSession(); - _pool.createNewJobForSession(testSession); - _pool.filterWrite(new NoOpFilter(), testSession, new IoFilter.WriteRequest("Message")); - - //Shutdown the pool - _executorService.getPool().shutdownNow(); - - try - { - - testSession = new TestSession(); - _pool.createNewJobForSession(testSession); - //prior to fix for QPID-172 this would throw RejectedExecutionException - _pool.filterWrite(null, testSession, null); - } - catch (RejectedExecutionException rje) - { - Assert.fail("RejectedExecutionException should not occur after pool has shutdown:" + rje); - } - } - - private static class NoOpFilter implements IoFilter.NextFilter - { - - public void sessionOpened(IoSession session) - { - } - - public void sessionClosed(IoSession session) - { - } - - public void sessionIdle(IoSession session, IdleStatus status) - { - } - - public void exceptionCaught(IoSession session, Throwable cause) - { - } - - public void messageReceived(IoSession session, Object message) - { - } - - public void messageSent(IoSession session, Object message) - { - } - - public void filterWrite(IoSession session, IoFilter.WriteRequest writeRequest) - { - } - - public void filterClose(IoSession session) - { - } - - public void sessionCreated(IoSession session) - { - } - } -} diff --git a/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkDriver.java b/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkDriver.java new file mode 100644 index 0000000000..a4c4b59cdd --- /dev/null +++ b/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkDriver.java @@ -0,0 +1,122 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.transport; + +import java.net.BindException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.nio.ByteBuffer; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +import org.apache.qpid.protocol.ProtocolEngine; +import org.apache.qpid.protocol.ProtocolEngineFactory; +import org.apache.qpid.ssl.SSLContextFactory; + +/** + * Test implementation of IoSession, which is required for some tests. Methods not being used are not implemented, + * so if this class is being used and some methods are to be used, then please update those. + */ +public class TestNetworkDriver implements NetworkDriver +{ + private final ConcurrentMap attributes = new ConcurrentHashMap(); + private String _remoteAddress = "127.0.0.1"; + private String _localAddress = "127.0.0.1"; + private int _port = 1; + + public TestNetworkDriver() + { + } + + public void setRemoteAddress(String string) + { + this._remoteAddress = string; + } + + public void setPort(int _port) + { + this._port = _port; + } + + public int getPort() + { + return _port; + } + + public void bind(int port, InetAddress[] addresses, ProtocolEngineFactory protocolFactory, + NetworkDriverConfiguration config, SSLContextFactory sslFactory) throws BindException + { + + } + + public SocketAddress getLocalAddress() + { + return new InetSocketAddress(_localAddress, _port); + } + + public SocketAddress getRemoteAddress() + { + return new InetSocketAddress(_remoteAddress, _port); + } + + public void open(int port, InetAddress destination, ProtocolEngine engine, NetworkDriverConfiguration config, + SSLContextFactory sslFactory) throws OpenException + { + + } + + public void setMaxReadIdle(int idleTime) + { + + } + + public void setMaxWriteIdle(int idleTime) + { + + } + + public void close() + { + + } + + public void flush() + { + + } + + public void send(ByteBuffer msg) + { + + } + + public void setIdleTimeout(long l) + { + + } + + public void setLocalAddress(String localAddress) + { + _localAddress = localAddress; + } + +} diff --git a/qpid/java/common/src/test/java/org/apache/qpid/transport/network/mina/MINANetworkDriverTest.java b/qpid/java/common/src/test/java/org/apache/qpid/transport/network/mina/MINANetworkDriverTest.java new file mode 100644 index 0000000000..5500ff9d4b --- /dev/null +++ b/qpid/java/common/src/test/java/org/apache/qpid/transport/network/mina/MINANetworkDriverTest.java @@ -0,0 +1,487 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +package org.apache.qpid.transport.network.mina; + +import java.net.BindException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.net.UnknownHostException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import junit.framework.TestCase; + +import org.apache.qpid.framing.AMQDataBlock; +import org.apache.qpid.protocol.ProtocolEngine; +import org.apache.qpid.protocol.ProtocolEngineFactory; +import org.apache.qpid.transport.NetworkDriver; +import org.apache.qpid.transport.OpenException; + +public class MINANetworkDriverTest extends TestCase +{ + + private static final String TEST_DATA = "YHALOTHAR"; + private static final int TEST_PORT = 2323; + private NetworkDriver _server; + private NetworkDriver _client; + private CountingProtocolEngine _countingEngine; // Keeps a count of how many bytes it's read + private Exception _thrownEx; + + @Override + public void setUp() + { + _server = new MINANetworkDriver(); + _client = new MINANetworkDriver(); + _thrownEx = null; + _countingEngine = new CountingProtocolEngine(); + } + + @Override + public void tearDown() + { + if (_server != null) + { + _server.close(); + } + + if (_client != null) + { + _client.close(); + } + } + + /** + * Tests that a socket can't be opened if a driver hasn't been bound + * to the port and can be opened if a driver has been bound. + * @throws BindException + * @throws UnknownHostException + * @throws OpenException + */ + public void testBindOpen() throws BindException, UnknownHostException, OpenException + { + try + { + _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null); + } + catch (OpenException e) + { + _thrownEx = e; + } + + assertNotNull("Open should have failed since no engine bound", _thrownEx); + + _server.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null); + + _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null); + } + + /** + * Tests that a socket can't be opened after a bound NetworkDriver has been closed + * @throws BindException + * @throws UnknownHostException + * @throws OpenException + */ + public void testBindOpenCloseOpen() throws BindException, UnknownHostException, OpenException + { + _server.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null); + _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null); + _client.close(); + _server.close(); + + try + { + _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null); + } + catch (OpenException e) + { + _thrownEx = e; + } + assertNotNull("Open should have failed", _thrownEx); + } + + /** + * Checks that the right exception is thrown when binding a NetworkDriver to an already + * existing socket. + */ + public void testBindPortInUse() + { + try + { + _server.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null); + } + catch (BindException e) + { + fail("First bind should not fail"); + } + + try + { + _client.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null); + } + catch (BindException e) + { + _thrownEx = e; + } + assertNotNull("Second bind should throw BindException", _thrownEx); + } + + /** + * tests that bytes sent on a network driver are received at the other end + * + * @throws UnknownHostException + * @throws OpenException + * @throws InterruptedException + * @throws BindException + */ + public void testSend() throws UnknownHostException, OpenException, InterruptedException, BindException + { + // Open a connection from a counting engine to an echo engine + _server.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null); + _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null); + + // Tell the counting engine how much data we're sending + _countingEngine.setNewLatch(TEST_DATA.getBytes().length); + + // Send the data and wait for up to 2 seconds to get it back + _client.send(ByteBuffer.wrap(TEST_DATA.getBytes())); + _countingEngine.getLatch().await(2, TimeUnit.SECONDS); + + // Check what we got + assertEquals("Wrong amount of data recieved", TEST_DATA.getBytes().length, _countingEngine.getReadBytes()); + } + + /** + * Opens a connection with a low read idle and check that it gets triggered + * @throws BindException + * @throws OpenException + * @throws UnknownHostException + * + */ + public void testSetReadIdle() throws BindException, UnknownHostException, OpenException + { + // Open a connection from a counting engine to an echo engine + _server.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null); + _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null); + assertFalse("Reader should not have been idle", _countingEngine.getReaderHasBeenIdle()); + _client.setMaxReadIdle(1); + try + { + Thread.sleep(1000); + } + catch (InterruptedException e) + { + // Eat it + } + assertTrue("Reader should have been idle", _countingEngine.getReaderHasBeenIdle()); + } + + /** + * Opens a connection with a low write idle and check that it gets triggered + * @throws BindException + * @throws OpenException + * @throws UnknownHostException + * + */ + public void testSetWriteIdle() throws BindException, UnknownHostException, OpenException + { + // Open a connection from a counting engine to an echo engine + _server.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null); + _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null); + assertFalse("Reader should not have been idle", _countingEngine.getWriterHasBeenIdle()); + _client.setMaxWriteIdle(1); + try + { + Thread.sleep(1000); + } + catch (InterruptedException e) + { + // Eat it + } + assertTrue("Reader should have been idle", _countingEngine.getWriterHasBeenIdle()); + } + + + /** + * Creates and then closes a connection from client to server and checks that the server + * has its closed() method called. Then creates a new client and closes the server to check + * that the client has its closed() method called. + * @throws BindException + * @throws UnknownHostException + * @throws OpenException + */ + public void testClosed() throws BindException, UnknownHostException, OpenException + { + // Open a connection from a counting engine to an echo engine + EchoProtocolEngineSingletonFactory factory = new EchoProtocolEngineSingletonFactory(); + _server.bind(TEST_PORT, null, factory, null, null); + _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null); + EchoProtocolEngine serverEngine = null; + while (serverEngine == null) + { + serverEngine = factory.getEngine(); + if (serverEngine == null) + { + try + { + Thread.sleep(10); + } + catch (InterruptedException e) + { + } + } + } + assertFalse("Server should not have been closed", serverEngine.getClosed()); + serverEngine.setNewLatch(1); + _client.close(); + try + { + serverEngine.getLatch().await(2, TimeUnit.SECONDS); + } + catch (InterruptedException e) + { + } + assertTrue("Server should have been closed", serverEngine.getClosed()); + + _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null); + _countingEngine.setClosed(false); + assertFalse("Client should not have been closed", _countingEngine.getClosed()); + _countingEngine.setNewLatch(1); + _server.close(); + try + { + _countingEngine.getLatch().await(2, TimeUnit.SECONDS); + } + catch (InterruptedException e) + { + } + assertTrue("Client should have been closed", _countingEngine.getClosed()); + } + + /** + * Create a connection and instruct the client to throw an exception when it gets some data + * and that the latch gets counted down. + * @throws BindException + * @throws UnknownHostException + * @throws OpenException + * @throws InterruptedException + */ + public void testExceptionCaught() throws BindException, UnknownHostException, OpenException, InterruptedException + { + _server.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null); + _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null); + + + assertEquals("Exception should not have been thrown", 1, + _countingEngine.getExceptionLatch().getCount()); + _countingEngine.setErrorOnNextRead(true); + _countingEngine.setNewLatch(TEST_DATA.getBytes().length); + _client.send(ByteBuffer.wrap(TEST_DATA.getBytes())); + _countingEngine.getExceptionLatch().await(2, TimeUnit.SECONDS); + assertEquals("Exception should have been thrown", 0, + _countingEngine.getExceptionLatch().getCount()); + } + + /** + * Opens a connection and checks that the remote address is the one that was asked for + * @throws BindException + * @throws UnknownHostException + * @throws OpenException + */ + public void testGetRemoteAddress() throws BindException, UnknownHostException, OpenException + { + _server.bind(TEST_PORT, null, new EchoProtocolEngineSingletonFactory(), null, null); + _client.open(TEST_PORT, InetAddress.getLocalHost(), _countingEngine, null, null); + assertEquals(new InetSocketAddress(InetAddress.getLocalHost(), TEST_PORT), + _client.getRemoteAddress()); + } + + private class EchoProtocolEngineSingletonFactory implements ProtocolEngineFactory + { + EchoProtocolEngine _engine = null; + + public ProtocolEngine newProtocolEngine(NetworkDriver driver) + { + if (_engine == null) + { + _engine = new EchoProtocolEngine(); + _engine.setNetworkDriver(driver); + } + return getEngine(); + } + + public EchoProtocolEngine getEngine() + { + return _engine; + } + } + + public class CountingProtocolEngine implements ProtocolEngine + { + + protected NetworkDriver _driver; + public ArrayList _receivedBytes = new ArrayList(); + private int _readBytes; + private CountDownLatch _latch = new CountDownLatch(0); + private boolean _readerHasBeenIdle; + private boolean _writerHasBeenIdle; + private boolean _closed = false; + private boolean _nextReadErrors = false; + private CountDownLatch _exceptionLatch = new CountDownLatch(1); + + public void closed() + { + setClosed(true); + _latch.countDown(); + } + + public void setErrorOnNextRead(boolean b) + { + _nextReadErrors = b; + } + + public void setNewLatch(int length) + { + _latch = new CountDownLatch(length); + } + + public long getReadBytes() + { + return _readBytes; + } + + public SocketAddress getRemoteAddress() + { + if (_driver != null) + { + return _driver.getRemoteAddress(); + } + else + { + return null; + } + } + + public SocketAddress getLocalAddress() + { + if (_driver != null) + { + return _driver.getLocalAddress(); + } + else + { + return null; + } + } + + public long getWrittenBytes() + { + return 0; + } + + public void readerIdle() + { + _readerHasBeenIdle = true; + } + + public void setNetworkDriver(NetworkDriver driver) + { + _driver = driver; + } + + public void writeFrame(AMQDataBlock frame) + { + + } + + public void writerIdle() + { + _writerHasBeenIdle = true; + } + + public void exception(Throwable t) + { + _exceptionLatch.countDown(); + } + + public CountDownLatch getExceptionLatch() + { + return _exceptionLatch; + } + + public void received(ByteBuffer msg) + { + // increment read bytes and count down the latch for that many + int bytes = msg.remaining(); + _readBytes += bytes; + for (int i = 0; i < bytes; i++) + { + _latch.countDown(); + } + + // Throw an error if we've been asked too, but we can still count + if (_nextReadErrors) + { + throw new RuntimeException("Was asked to error"); + } + } + + public CountDownLatch getLatch() + { + return _latch; + } + + public boolean getWriterHasBeenIdle() + { + return _writerHasBeenIdle; + } + + public boolean getReaderHasBeenIdle() + { + return _readerHasBeenIdle; + } + + public void setClosed(boolean _closed) + { + this._closed = _closed; + } + + public boolean getClosed() + { + return _closed; + } + + } + + private class EchoProtocolEngine extends CountingProtocolEngine + { + + public void received(ByteBuffer msg) + { + super.received(msg); + msg.rewind(); + _driver.send(msg); + } + } +} \ No newline at end of file diff --git a/qpid/java/systests/build.xml b/qpid/java/systests/build.xml index ac3c77e4a3..34a360ecad 100644 --- a/qpid/java/systests/build.xml +++ b/qpid/java/systests/build.xml @@ -20,7 +20,7 @@ nn - or more contributor license agreements. See the NOTICE file --> - + diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/SimpleACLTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/SimpleACLTest.java index a755bbfaa7..bb7b5efc75 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/SimpleACLTest.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/SimpleACLTest.java @@ -21,31 +21,37 @@ package org.apache.qpid.server.security.acl; -import org.apache.qpid.AMQException; -import org.apache.qpid.AMQConnectionFailureException; -import org.apache.qpid.client.AMQAuthenticationException; -import org.apache.qpid.client.AMQConnection; -import org.apache.qpid.client.AMQSession; + +import junit.framework.TestCase; + +import org.apache.log4j.BasicConfigurator; +import org.apache.log4j.Logger; +import org.apache.qpid.client.transport.TransportConnection; +import org.apache.qpid.client.*; import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.jms.ConnectionListener; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry; +import org.apache.qpid.AMQConnectionFailureException; +import org.apache.qpid.AMQException; import org.apache.qpid.test.utils.QpidTestCase; +import org.apache.qpid.jms.ConnectionListener; import org.apache.qpid.url.URLSyntaxException; -import javax.jms.Connection; -import javax.jms.DeliveryMode; +import javax.jms.*; import javax.jms.IllegalStateException; -import javax.jms.JMSException; -import javax.jms.Message; -import javax.jms.MessageConsumer; -import javax.jms.MessageProducer; -import javax.jms.Queue; -import javax.jms.Session; -import javax.jms.TextMessage; import javax.naming.NamingException; + import java.io.File; +import java.util.ArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; -public class SimpleACLTest extends QpidTestCase implements ConnectionListener +public class SimpleACLTest extends QpidTestCase implements ConnectionListener, ExceptionListener { + private String BROKER = "vm://:"+ApplicationRegistry.DEFAULT_INSTANCE;//"tcp://localhost:5672"; + private ArrayList _thrownExceptions = new ArrayList(); + private CountDownLatch _awaitError = new CountDownLatch(51); + public void setUp() throws Exception { final String QPID_HOME = System.getProperty("QPID_HOME"); @@ -134,7 +140,7 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener } } - public void testClientConsumeFromNamedQueueInvalid() throws NamingException + public void testClientConsumeFromNamedQueueInvalid() throws NamingException, Exception { try { @@ -183,7 +189,7 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener } } - public void testClientCreateNamedQueue() throws NamingException, JMSException, AMQException + public void testClientCreateNamedQueue() throws NamingException, JMSException, AMQException, Exception { try { @@ -263,7 +269,7 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener } } - public void testClientPublishInvalidQueueSuccess() throws AMQException, URLSyntaxException, JMSException, NamingException + public void testClientPublishInvalidQueueSuccess() throws AMQException, URLSyntaxException, JMSException, NamingException, Exception { try { @@ -271,41 +277,38 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener ((AMQConnection) conn).setConnectionListener(this); - Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + Session session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE); conn.start(); - + conn.setExceptionListener(this); MessageProducer sender = ((AMQSession) session).createProducer(null); - Queue queue = session.createQueue("Invalid"); - + Queue queue = session.createQueue("NewQueueThatIDoNotHaveRightsToPublishToo"); + // Send a message that we will wait to be sent, this should give the broker time to close the connection // before we finish this test. Message is set !immed !mand as the queue is invalid so want to test ACLs not // queue existence. ((org.apache.qpid.jms.MessageProducer) sender).send(queue, session.createTextMessage("test"), DeliveryMode.NON_PERSISTENT, 0, 0L, false, false, true); - // Test the connection with a valid consumer - // This may fail as the session may be closed before the queue or the consumer created. - Queue temp = session.createTemporaryQueue(); - - session.createConsumer(temp).close(); - - //Connection should now be closed and will throw the exception caused by the above send - conn.close(); - - fail("Close is not expected to succeed."); + _awaitError.await(1, TimeUnit.SECONDS); } catch (JMSException e) { - Throwable cause = e.getLinkedException(); - if (!(cause instanceof AMQAuthenticationException)) + fail("Unknown exception thrown:" + e.getMessage()); + } + boolean foundCorrectException = false; + for (Exception cause : _thrownExceptions) + { + if (cause instanceof JMSException) { - e.printStackTrace(); + if (((JMSException) cause).getLinkedException() instanceof AMQAuthenticationException) + { + foundCorrectException = true; + } } - assertEquals("Incorrect exception", AMQAuthenticationException.class, cause.getClass()); - assertEquals("Incorrect error code thrown", 403, ((AMQAuthenticationException) cause).getErrorCode().getCode()); } + assertTrue("Did not get AMQAuthenticationException thrown", foundCorrectException); } public void testServerConsumeFromNamedQueueValid() throws AMQException, URLSyntaxException @@ -328,7 +331,7 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener } } - public void testServerConsumeFromNamedQueueInvalid() throws AMQException, URLSyntaxException, NamingException + public void testServerConsumeFromNamedQueueInvalid() throws AMQException, URLSyntaxException, NamingException, Exception { try { @@ -353,7 +356,7 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener } } - public void testServerConsumeFromTemporaryQueue() throws AMQException, URLSyntaxException, NamingException + public void testServerConsumeFromTemporaryQueue() throws AMQException, URLSyntaxException, NamingException, Exception { try { @@ -409,7 +412,7 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener } } - public void testServerCreateNamedQueueInvalid() throws JMSException, URLSyntaxException, AMQException, NamingException + public void testServerCreateNamedQueueInvalid() throws JMSException, URLSyntaxException, AMQException, NamingException, Exception { try { @@ -431,7 +434,7 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener } } - public void testServerCreateTemporaryQueueInvalid() throws NamingException + public void testServerCreateTemporaryQueueInvalid() throws NamingException, Exception { try { @@ -456,7 +459,7 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener } } - public void testServerCreateAutoDeleteQueueInvalid() throws NamingException, JMSException, AMQException + public void testServerCreateAutoDeleteQueueInvalid() throws NamingException, JMSException, AMQException, Exception { Connection connection = null; try @@ -487,7 +490,7 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener * @throws URLSyntaxException * @throws JMSException */ - public void testServerPublishUsingTransactionSuccess() throws AMQException, URLSyntaxException, JMSException, NamingException + public void testServerPublishUsingTransactionSuccess() throws AMQException, URLSyntaxException, JMSException, NamingException, Exception { //Set up the Server Connection serverConnection = getConnection("server", "guest"); @@ -567,7 +570,7 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener } } - public void testServerPublishInvalidQueueSuccess() throws AMQException, URLSyntaxException, JMSException, NamingException + public void testServerPublishInvalidQueueSuccess() throws AMQException, URLSyntaxException, JMSException, NamingException, Exception { try { @@ -642,4 +645,10 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener public void failoverComplete() { } + + public void onException(JMSException arg0) + { + _thrownExceptions.add(arg0); + _awaitError.countDown(); + } } diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java index 91cb37e455..7a27925a36 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java @@ -20,27 +20,27 @@ */ package org.apache.qpid.test.unit.client.protocol; -import org.apache.mina.common.IoSession; import org.apache.qpid.client.AMQConnection; import org.apache.qpid.client.protocol.AMQProtocolHandler; import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.test.utils.QpidTestCase; -import org.apache.qpid.test.utils.protocol.TestIoSession; +import org.apache.qpid.transport.TestNetworkDriver; +import org.apache.qpid.transport.NetworkDriver; public class AMQProtocolSessionTest extends QpidTestCase { private static class AMQProtSession extends AMQProtocolSession { - public AMQProtSession(AMQProtocolHandler protocolHandler, IoSession protocolSession, AMQConnection connection) + public AMQProtSession(AMQProtocolHandler protocolHandler, AMQConnection connection) { - super(protocolHandler,protocolSession,connection); + super(protocolHandler,connection); } - public TestIoSession getMinaProtocolSession() + public TestNetworkDriver getNetworkDriver() { - return (TestIoSession) _minaProtocolSession; + return (TestNetworkDriver) _protocolHandler.getNetworkDriver(); } public AMQShortString genQueueName() @@ -63,8 +63,11 @@ public class AMQProtocolSessionTest extends QpidTestCase { super.setUp(); + AMQConnection con = (AMQConnection) getConnection("guest", "guest"); + AMQProtocolHandler protocolHandler = new AMQProtocolHandler(con); + protocolHandler.setNetworkDriver(new TestNetworkDriver()); //don't care about the values set here apart from the dummy IoSession - _testSession = new AMQProtSession(null,new TestIoSession(), (AMQConnection) getConnection("guest", "guest")); + _testSession = new AMQProtSession(protocolHandler , con); //initialise addresses for test and expected results _port = 123; @@ -81,20 +84,20 @@ public class AMQProtocolSessionTest extends QpidTestCase AMQShortString testAddress; //test address with / and ; chars which generateQueueName should removeKey - _testSession.getMinaProtocolSession().setStringLocalAddress(_brokenAddress); - _testSession.getMinaProtocolSession().setLocalPort(_port); + _testSession.getNetworkDriver().setLocalAddress(_brokenAddress); + _testSession.getNetworkDriver().setPort(_port); testAddress = _testSession.genQueueName(); assertEquals("Failure when generating a queue exchange from an address with special chars",_generatedAddress,testAddress.toString()); //test empty address - _testSession.getMinaProtocolSession().setStringLocalAddress(_emptyAddress); + _testSession.getNetworkDriver().setLocalAddress(_emptyAddress); testAddress = _testSession.genQueueName(); assertEquals("Failure when generating a queue exchange from an empty address",_generatedAddress_2,testAddress.toString()); //test address with no special chars - _testSession.getMinaProtocolSession().setStringLocalAddress(_validAddress); + _testSession.getNetworkDriver().setLocalAddress(_validAddress); testAddress = _testSession.genQueueName(); assertEquals("Failure when generating a queue exchange from an address with no special chars",_generatedAddress_3,testAddress.toString()); diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/FailoverBaseCase.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/FailoverBaseCase.java index 440bc31fe9..deb021fd96 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/FailoverBaseCase.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/FailoverBaseCase.java @@ -21,6 +21,8 @@ package org.apache.qpid.test.utils; import javax.jms.Connection; +import javax.jms.JMSException; +import javax.naming.NamingException; public class FailoverBaseCase extends QpidTestCase { @@ -62,7 +64,7 @@ public class FailoverBaseCase extends QpidTestCase * @return a connection * @throws Exception */ - public Connection getConnection() throws Exception + public Connection getConnection() throws JMSException, NamingException { Connection conn = (Boolean.getBoolean("profile.use_ssl"))? diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java index 666c97c9de..2f40d3a62a 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java @@ -958,7 +958,7 @@ public class QpidTestCase extends TestCase return (AMQConnectionFactory) getInitialContext().lookup(factoryName); } - public Connection getConnection() throws Exception + public Connection getConnection() throws JMSException, NamingException { return getConnection("guest", "guest"); } -- cgit v1.2.1