diff options
Diffstat (limited to 'java/cluster/src')
3 files changed, 17 insertions, 13 deletions
diff --git a/java/cluster/src/main/java/org/apache/qpid/server/cluster/Main.java b/java/cluster/src/main/java/org/apache/qpid/server/cluster/Main.java index 57779a0550..3eeddd7b4e 100644 --- a/java/cluster/src/main/java/org/apache/qpid/server/cluster/Main.java +++ b/java/cluster/src/main/java/org/apache/qpid/server/cluster/Main.java @@ -31,6 +31,7 @@ import org.apache.commons.cli.PosixParser; import org.apache.log4j.Logger; import org.apache.mina.common.IoAcceptor; import org.apache.mina.transport.socket.nio.SocketAcceptor; +import org.apache.mina.transport.socket.nio.SocketAcceptorConfig; import org.apache.mina.transport.socket.nio.SocketSessionConfig; import org.apache.qpid.pool.ReadWriteThreadModel; import org.apache.qpid.server.registry.ApplicationRegistry; @@ -71,7 +72,8 @@ public class Main extends org.apache.qpid.server.Main try { IoAcceptor acceptor = new SocketAcceptor(); - SocketSessionConfig sc = (SocketSessionConfig) acceptor.getSessionConfig(); + SocketAcceptorConfig sconfig = (SocketAcceptorConfig) acceptor.getDefaultConfig(); + SocketSessionConfig sc = (SocketSessionConfig) sconfig.getSessionConfig(); sc.setReceiveBufferSize(connectorConfig.socketReceiveBufferSize); sc.setSendBufferSize(connectorConfig.socketWriteBuferSize); @@ -81,16 +83,14 @@ public class Main extends org.apache.qpid.server.Main // implementation provided by MINA if (connectorConfig.enableExecutorPool) { - acceptor.setThreadModel(new ReadWriteThreadModel()); + sconfig.setThreadModel(new ReadWriteThreadModel()); } String host = InetAddress.getLocalHost().getHostName(); ClusteredProtocolHandler handler = new ClusteredProtocolHandler(new InetSocketAddress(host, port)); if (connectorConfig.enableNonSSL) { - acceptor.setLocalAddress(new InetSocketAddress(port)); - acceptor.setHandler(handler); - acceptor.bind(); + acceptor.bind(new InetSocketAddress(port), handler, sconfig); _logger.info("Qpid.AMQP listening on non-SSL port " + port); handler.connect(commandLine.getOptionValue("j")); } @@ -99,9 +99,7 @@ public class Main extends org.apache.qpid.server.Main { ClusteredProtocolHandler sslHandler = new ClusteredProtocolHandler(handler); sslHandler.setUseSSL(true); - acceptor.setLocalAddress(new InetSocketAddress(connectorConfig.sslPort)); - acceptor.setHandler(handler); - acceptor.bind(); + acceptor.bind(new InetSocketAddress(connectorConfig.sslPort), handler, sconfig); _logger.info("Qpid.AMQP listening on SSL port " + connectorConfig.sslPort); } } diff --git a/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java b/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java index ee3b3ceb8a..275ed39b5f 100644 --- a/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java +++ b/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java @@ -27,6 +27,7 @@ import org.apache.mina.common.IoSession; import org.apache.mina.common.RuntimeIOException; import org.apache.mina.filter.codec.ProtocolCodecFilter; 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.qpid.AMQException; import org.apache.qpid.server.cluster.util.LogMessage; @@ -62,7 +63,7 @@ public class MinaBrokerProxy extends Broker implements MethodHandler { super(host, port); _local = local; - _legacyHandler = new ClientHandlerRegistry(local); + _legacyHandler = new ClientHandlerRegistry(local); } private void init(IoSession session) @@ -75,14 +76,14 @@ public class MinaBrokerProxy extends Broker implements MethodHandler { _logger.info("Connecting to cluster peer: " + getDetails()); SocketConnector ioConnector = new SocketConnector(); + SocketConnectorConfig cfg = (SocketConnectorConfig) ioConnector.getDefaultConfig(); - SocketSessionConfig scfg = (SocketSessionConfig) ioConnector.getSessionConfig(); + SocketSessionConfig scfg = (SocketSessionConfig) cfg.getSessionConfig(); scfg.setTcpNoDelay(true); scfg.setSendBufferSize(32768); scfg.setReceiveBufferSize(32768); InetSocketAddress address = new InetSocketAddress(getHost(), getPort()); - ioConnector.setHandler(_binding); - return ioConnector.connect(address); + return ioConnector.connect(address, _binding); } //extablish connection without handling redirect diff --git a/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestSession.java b/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestSession.java index da7c17c181..86ec808924 100644 --- a/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestSession.java +++ b/java/cluster/src/test/java/org/apache/qpid/server/cluster/TestSession.java @@ -32,6 +32,11 @@ class TestSession implements IoSession return null; //TODO } + public IoServiceConfig getServiceConfig() + { + return null; //TODO + } + public IoHandler getHandler() { return null; //TODO @@ -217,7 +222,7 @@ class TestSession implements IoSession return 0; //TODO } - public int getScheduledWriteMessages() + public int getScheduledWriteRequests() { return 0; //TODO } |