You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mina.apache.org by jv...@apache.org on 2013/05/26 17:15:31 UTC
[2/5] adding BIO based UDP server (WIP)
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java b/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java
new file mode 100644
index 0000000..7842770
--- /dev/null
+++ b/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java
@@ -0,0 +1,345 @@
+/*
+ * 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.mina.transport.nio;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+
+import org.apache.mina.api.IdleStatus;
+import org.apache.mina.service.executor.IoHandlerExecutor;
+import org.apache.mina.service.executor.OrderedHandlerExecutor;
+import org.apache.mina.service.idlechecker.IdleChecker;
+import org.apache.mina.service.idlechecker.IndexedIdleChecker;
+import org.apache.mina.transport.tcp.AbstractTcpServer;
+import org.apache.mina.transport.tcp.TcpSessionConfig;
+import org.apache.mina.util.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class implements a TCP NIO based server.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ */
+public class NioTcpServer extends AbstractTcpServer implements SelectorListener {
+ /** A logger for this class */
+ static final Logger LOG = LoggerFactory.getLogger(NioTcpServer.class);
+
+ /** the bound local address */
+ private SocketAddress address = null;
+
+ private final SelectorLoop acceptSelectorLoop;
+
+ private final SelectorLoopPool readWriteSelectorPool;
+
+ // the key used for selecting accept event
+ private SelectionKey acceptKey = null;
+
+ // the server socket for accepting clients
+ private ServerSocketChannel serverChannel = null;
+
+ private IdleChecker idleChecker;
+
+ /**
+ * Create a TCP server with new selector pool of default size and a {@link IoHandlerExecutor} of default type (
+ * {@link OrderedHandlerExecutor})
+ */
+ public NioTcpServer() {
+ this(new NioSelectorLoop("accept", 0), new FixedSelectorLoopPool("Server", Runtime.getRuntime()
+ .availableProcessors() + 1), null);
+ }
+
+ /**
+ * Create a TCP server with new selector pool of default size and a {@link IoHandlerExecutor} of default type (
+ * {@link OrderedHandlerExecutor})
+ *
+ * @param config The specific configuration to use
+ */
+ public NioTcpServer(TcpSessionConfig config) {
+ this(config, new NioSelectorLoop("accept", 0), new FixedSelectorLoopPool("Server", Runtime.getRuntime()
+ .availableProcessors() + 1), null);
+ }
+
+ /**
+ * Create a TCP server with provided selector loops pool. We will use one SelectorLoop get from the pool to manage
+ * the OP_ACCEPT events. If the pool contains only one SelectorLoop, then all the events will be managed by the same
+ * Selector.
+ *
+ * @param selectorLoopPool the selector loop pool for handling all I/O events (accept, read, write)
+ * @param ioHandlerExecutor used for executing IoHandler event in another pool of thread (not in the low level I/O
+ * one). Use <code>null</code> if you don't want one. Be careful, the IoHandler processing will block the I/O
+ * operations.
+ */
+ public NioTcpServer(SelectorLoopPool selectorLoopPool, IoHandlerExecutor handlerExecutor) {
+ this(selectorLoopPool.getSelectorLoop(), selectorLoopPool, handlerExecutor);
+ }
+
+ /**
+ * Create a TCP server with provided selector loops pool. We will use one SelectorLoop get from the pool to manage
+ * the OP_ACCEPT events. If the pool contains only one SelectorLoop, then all the events will be managed by the same
+ * Selector.
+ *
+ * @param config The specific configuration to use
+ * @param selectorLoopPool the selector loop pool for handling all I/O events (accept, read, write)
+ * @param ioHandlerExecutor used for executing IoHandler event in another pool of thread (not in the low level I/O
+ * one). Use <code>null</code> if you don't want one. Be careful, the IoHandler processing will block the I/O
+ * operations.
+ */
+ public NioTcpServer(TcpSessionConfig config, SelectorLoopPool selectorLoopPool, IoHandlerExecutor handlerExecutor) {
+ this(config, selectorLoopPool.getSelectorLoop(), selectorLoopPool, handlerExecutor);
+ }
+
+ /**
+ * Create a TCP server with provided selector loops pool
+ *
+ * @param acceptSelectorLoop the selector loop for handling accept events (connection of new session)
+ * @param readWriteSelectorLoop the pool of selector loop for handling read/write events of connected sessions
+ * @param ioHandlerExecutor used for executing IoHandler event in another pool of thread (not in the low level I/O
+ * one). Use <code>null</code> if you don't want one. Be careful, the IoHandler processing will block the I/O
+ * operations.
+ */
+ public NioTcpServer(SelectorLoop acceptSelectorLoop, SelectorLoopPool readWriteSelectorLoop,
+ IoHandlerExecutor handlerExecutor) {
+ super(handlerExecutor);
+ this.acceptSelectorLoop = acceptSelectorLoop;
+ this.readWriteSelectorPool = readWriteSelectorLoop;
+ }
+
+ /**
+ * Create a TCP server with provided selector loops pool
+ *
+ * @param config The specific configuration to use
+ * @param acceptSelectorLoop the selector loop for handling accept events (connection of new session)
+ * @param readWriteSelectorLoop the pool of selector loop for handling read/write events of connected sessions
+ * @param ioHandlerExecutor used for executing IoHandler event in another pool of thread (not in the low level I/O
+ * one). Use <code>null</code> if you don't want one. Be careful, the IoHandler processing will block the I/O
+ * operations.
+ */
+ public NioTcpServer(TcpSessionConfig config, SelectorLoop acceptSelectorLoop,
+ SelectorLoopPool readWriteSelectorLoop, IoHandlerExecutor handlerExecutor) {
+ super(config, handlerExecutor);
+ this.acceptSelectorLoop = acceptSelectorLoop;
+ this.readWriteSelectorPool = readWriteSelectorLoop;
+ }
+
+ /**
+ * Get the inner Server socket for accepting new client connections
+ *
+ * @return
+ */
+ public synchronized ServerSocketChannel getServerSocketChannel() {
+ return this.serverChannel;
+ }
+
+ public synchronized void setServerSocketChannel(final ServerSocketChannel serverChannel) {
+ this.serverChannel = serverChannel;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void bind(final int port) throws IOException {
+ bind(new InetSocketAddress(port));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public synchronized void bind(final SocketAddress localAddress) throws IOException {
+ Assert.assertNotNull(localAddress, "localAddress");
+
+ // check if the address is already bound
+ if (this.address != null) {
+ throw new IOException("address " + address + " already bound");
+ }
+
+ LOG.info("binding address {}", localAddress);
+ this.address = localAddress;
+
+ serverChannel = ServerSocketChannel.open();
+ serverChannel.socket().setReuseAddress(isReuseAddress());
+ serverChannel.socket().bind(address);
+ serverChannel.configureBlocking(false);
+
+ acceptSelectorLoop.register(true, false, false, false, this, serverChannel, null);
+
+ idleChecker = new IndexedIdleChecker();
+ idleChecker.start();
+
+ // it's the first address bound, let's fire the event
+ this.fireServiceActivated();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public SocketAddress getBoundAddress() {
+ return address;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public synchronized void unbind() throws IOException {
+ LOG.info("unbinding {}", address);
+ if (this.address == null) {
+ throw new IllegalStateException("server not bound");
+ }
+ serverChannel.socket().close();
+ serverChannel.close();
+ acceptSelectorLoop.unregister(this, serverChannel);
+
+ this.address = null;
+ this.fireServiceInactivated();
+
+ // will stop the acceptor processor if we are the last service
+ idleChecker.destroy();
+ }
+
+ /**
+ * @return the acceptKey
+ */
+ public SelectionKey getAcceptKey() {
+ return acceptKey;
+ }
+
+ /**
+ * @param acceptKey the acceptKey to set
+ */
+ public void setAcceptKey(final SelectionKey acceptKey) {
+ this.acceptKey = acceptKey;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void ready(final boolean accept, boolean connect, final boolean read, final ByteBuffer readBuffer,
+ final boolean write) {
+ if (accept) {
+ LOG.debug("acceptable new client");
+
+ // accepted connection
+ try {
+ LOG.debug("new client accepted");
+ createSession(getServerSocketChannel().accept());
+
+ } catch (final IOException e) {
+ LOG.error("error while accepting new client", e);
+ }
+ }
+
+ if (read || write) {
+ throw new IllegalStateException("should not receive read or write events");
+ }
+ }
+
+ private void createSession(SocketChannel clientSocket) throws IOException {
+ LOG.debug("create session");
+ SocketChannel socketChannel = clientSocket;
+ TcpSessionConfig config = getSessionConfig();
+ SelectorLoop readWriteSelectorLoop = readWriteSelectorPool.getSelectorLoop();
+ final NioTcpSession session = new NioTcpSession(this, socketChannel, readWriteSelectorLoop, idleChecker);
+
+ socketChannel.configureBlocking(false);
+
+ // apply idle configuration
+ session.getConfig().setIdleTimeInMillis(IdleStatus.READ_IDLE, config.getIdleTimeInMillis(IdleStatus.READ_IDLE));
+ session.getConfig().setIdleTimeInMillis(IdleStatus.WRITE_IDLE,
+ config.getIdleTimeInMillis(IdleStatus.WRITE_IDLE));
+
+ // apply the default service socket configuration
+ Boolean keepAlive = config.isKeepAlive();
+
+ if (keepAlive != null) {
+ session.getConfig().setKeepAlive(keepAlive);
+ }
+
+ Boolean oobInline = config.isOobInline();
+
+ if (oobInline != null) {
+ session.getConfig().setOobInline(oobInline);
+ }
+
+ Boolean reuseAddress = config.isReuseAddress();
+
+ if (reuseAddress != null) {
+ session.getConfig().setReuseAddress(reuseAddress);
+ }
+
+ Boolean tcpNoDelay = config.isTcpNoDelay();
+
+ if (tcpNoDelay != null) {
+ session.getConfig().setTcpNoDelay(tcpNoDelay);
+ }
+
+ Integer receiveBufferSize = config.getReadBufferSize();
+
+ if (receiveBufferSize != null) {
+ session.getConfig().setReadBufferSize(receiveBufferSize);
+ }
+
+ Integer sendBufferSize = config.getSendBufferSize();
+
+ if (sendBufferSize != null) {
+ session.getConfig().setSendBufferSize(sendBufferSize);
+ }
+
+ Integer trafficClass = config.getTrafficClass();
+
+ if (trafficClass != null) {
+ session.getConfig().setTrafficClass(trafficClass);
+ }
+
+ Integer soLinger = config.getSoLinger();
+
+ if (soLinger != null) {
+ session.getConfig().setSoLinger(soLinger);
+ }
+
+ // Set the secured flag if the service is to be used over SSL/TLS
+ if (config.isSecured()) {
+ session.initSecure(config.getSslContext());
+ }
+
+ // add the session to the queue for being added to the selector
+ readWriteSelectorLoop.register(false, false, true, false, session, socketChannel, new RegistrationCallback() {
+
+ @Override
+ public void done(SelectionKey selectionKey) {
+ session.setSelectionKey(selectionKey);
+ session.setConnected();
+ }
+ });
+
+ idleChecker.sessionRead(session, System.currentTimeMillis());
+ idleChecker.sessionWritten(session, System.currentTimeMillis());
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/transport/nio/NioTcpSession.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/nio/NioTcpSession.java b/core/src/main/java/org/apache/mina/transport/nio/NioTcpSession.java
new file mode 100644
index 0000000..c9f395f
--- /dev/null
+++ b/core/src/main/java/org/apache/mina/transport/nio/NioTcpSession.java
@@ -0,0 +1,386 @@
+/*
+ * 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.mina.transport.nio;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.SocketChannel;
+
+import org.apache.mina.api.IoService;
+import org.apache.mina.service.idlechecker.IdleChecker;
+import org.apache.mina.session.WriteRequest;
+import org.apache.mina.transport.ConnectFuture;
+import org.apache.mina.transport.tcp.ProxyTcpSessionConfig;
+import org.apache.mina.transport.tcp.TcpSessionConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A NIO based TCP session, should be used by {@link NioTcpServer} and {@link NioTcpClient}. A TCP session is a
+ * connection between a our server/client and the remote end-point.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ *
+ */
+public class NioTcpSession extends AbstractNioSession implements SelectorListener {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NioTcpSession.class);
+
+ /** the selector loop in charge of generating read/write events for this session */
+ private final SelectorLoop selectorLoop;
+
+ /** the socket configuration */
+ private final TcpSessionConfig configuration;
+
+ /** the future representing this session connection operation (client only) */
+ private ConnectFuture connectFuture;
+
+ /** The associated selectionKey */
+ private SelectionKey selectionKey;
+
+ /** The Direct Buffer used to send data */
+ private ByteBuffer sendBuffer;
+
+ /** The size of the buffer configured in the socket to send data */
+ private int sendBufferSize;
+
+ /* No qualifier */NioTcpSession(final IoService service, final SocketChannel channel,
+ final SelectorLoop selectorLoop, final IdleChecker idleChecker) {
+ super(service, channel, idleChecker);
+ this.selectorLoop = selectorLoop;
+ this.configuration = new ProxyTcpSessionConfig(channel.socket());
+ sendBufferSize = configuration.getSendBufferSize();
+ sendBuffer = ByteBuffer.allocateDirect(sendBufferSize);
+ }
+
+ void setConnectFuture(ConnectFuture connectFuture) {
+ this.connectFuture = connectFuture;
+ }
+
+ /**
+ * Get the underlying {@link SocketChannel} of this session
+ *
+ * @return the socket channel used by this session
+ */
+ SocketChannel getSocketChannel() {
+ return (SocketChannel) channel;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public InetSocketAddress getRemoteAddress() {
+ if (channel == null) {
+ return null;
+ }
+ final Socket socket = ((SocketChannel) channel).socket();
+
+ if (socket == null) {
+ return null;
+ }
+
+ return (InetSocketAddress) socket.getRemoteSocketAddress();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public InetSocketAddress getLocalAddress() {
+ if (channel == null) {
+ return null;
+ }
+
+ final Socket socket = ((SocketChannel) channel).socket();
+
+ if (socket == null) {
+ return null;
+ }
+
+ return (InetSocketAddress) socket.getLocalSocketAddress();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void suspendRead() {
+ // TODO
+ throw new RuntimeException("Not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void suspendWrite() {
+ // TODO
+ throw new RuntimeException("Not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected int writeDirect(Object message) {
+ try {
+ // Check that we can write into the channel
+ if (!isRegisteredForWrite()) {
+ // We don't have pending writes
+ return ((SocketChannel) channel).write((ByteBuffer) message);
+ } else {
+ return -1;
+ }
+ } catch (final IOException e) {
+ LOG.error("Exception while reading : ", e);
+ processException(e);
+
+ return -1;
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected ByteBuffer convertToDirectBuffer(WriteRequest writeRequest, boolean createNew) {
+ ByteBuffer message = (ByteBuffer) writeRequest.getMessage();
+
+ if (!message.isDirect()) {
+ int remaining = message.remaining();
+
+ if ((remaining > sendBufferSize) || createNew) {
+ ByteBuffer directBuffer = ByteBuffer.allocateDirect(remaining);
+ directBuffer.put(message);
+ directBuffer.flip();
+ writeRequest.setMessage(directBuffer);
+
+ return directBuffer;
+ } else {
+ sendBuffer.clear();
+ sendBuffer.put(message);
+ sendBuffer.flip();
+ writeRequest.setMessage(sendBuffer);
+
+ return sendBuffer;
+ }
+ }
+
+ return message;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void resumeRead() {
+ // TODO
+ throw new RuntimeException("Not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void resumeWrite() {
+ // TODO
+ throw new RuntimeException("Not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isReadSuspended() {
+ // TODO
+ return false;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isWriteSuspended() {
+ // TODO
+ throw new RuntimeException("Not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public TcpSessionConfig getConfig() {
+ return configuration;
+ }
+
+ /**
+ * Set this session status as connected. To be called by the processor selecting/polling this session.
+ */
+ void setConnected() {
+ if (!isCreated()) {
+ throw new RuntimeException("Trying to open a non created session");
+ }
+
+ state = SessionState.CONNECTED;
+
+ if (connectFuture != null) {
+ connectFuture.complete(this);
+ // free some memory
+ connectFuture = null;
+ }
+
+ processSessionOpen();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected void channelClose() {
+ try {
+ selectorLoop.unregister(this, channel);
+ channel.close();
+ } catch (final IOException e) {
+ LOG.error("Exception while closing the channel : ", e);
+ processException(e);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void flushWriteQueue() {
+ // register for write
+ selectorLoop.modifyRegistration(false, !isReadSuspended(), true, this, channel, true);
+ }
+
+ /**
+ * Process a read operation : read the data from the channel and push them to the chain.
+ *
+ * @param readBuffer The buffer that will contain the read data
+ */
+ private void processRead(final ByteBuffer readBuffer) {
+ try {
+ LOG.debug("readable session : {}", this);
+
+ // Read everything we can up to the buffer size
+ final int readCount = ((SocketChannel) channel).read(readBuffer);
+
+ LOG.debug("read {} bytes", readCount);
+
+ if (readCount < 0) {
+ // session closed by the remote peer
+ LOG.debug("session closed by the remote peer");
+ close(true);
+ } else if (readCount > 0) {
+ // we have read some data
+ // limit at the current position & rewind buffer back to start &
+ // push to the chain
+ readBuffer.flip();
+
+ if (isSecured()) {
+ // We are reading data over a SSL/TLS encrypted connection.
+ // Redirect the processing to the SslHelper class.
+ final SslHelper sslHelper = getAttribute(SSL_HELPER, null);
+
+ if (sslHelper == null) {
+ throw new IllegalStateException();
+ }
+
+ sslHelper.processRead(this, readBuffer);
+
+ // We don't clear the buffer. It has been done by the sslHelper
+ } else {
+ // Plain message, not encrypted : go directly to the chain
+ processMessageReceived(readBuffer);
+
+ // And now, clear the buffer
+ readBuffer.clear();
+ }
+
+ // Update the session idle status
+ idleChecker.sessionRead(this, System.currentTimeMillis());
+ }
+ } catch (final IOException e) {
+ LOG.error("Exception while reading : ", e);
+ processException(e);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void ready(final boolean accept, boolean connect, final boolean read, final ByteBuffer readBuffer,
+ final boolean write) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("session {} ready for accept={}, connect={}, read={}, write={}", new Object[] { this, accept,
+ connect, read, write });
+ }
+ if (connect) {
+ try {
+
+ boolean isConnected = ((SocketChannel) channel).finishConnect();
+
+ if (!isConnected) {
+ LOG.error("unable to connect session {}", this);
+ } else {
+ // cancel current registration for connection
+ selectionKey.cancel();
+ selectionKey = null;
+
+ // Register for reading
+ selectorLoop.register(false, false, true, false, this, channel, new RegistrationCallback() {
+
+ @Override
+ public void done(SelectionKey selectionKey) {
+ setConnected();
+ }
+ });
+ }
+ } catch (IOException e) {
+ LOG.debug("Connection error, we cancel the future", e);
+ if (connectFuture != null) {
+ connectFuture.error(e);
+ }
+ }
+ }
+
+ if (read) {
+ processRead(readBuffer);
+ }
+
+ if (write) {
+ processWrite(selectorLoop);
+ }
+ if (accept) {
+ throw new IllegalStateException("accept event should never occur on NioTcpSession");
+ }
+ }
+
+ void setSelectionKey(SelectionKey key) {
+ this.selectionKey = key;
+ }
+}
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/transport/nio/NioUdpClient.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/nio/NioUdpClient.java b/core/src/main/java/org/apache/mina/transport/nio/NioUdpClient.java
new file mode 100644
index 0000000..1d51ccf
--- /dev/null
+++ b/core/src/main/java/org/apache/mina/transport/nio/NioUdpClient.java
@@ -0,0 +1,139 @@
+/*
+ * 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.mina.transport.nio;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.nio.channels.DatagramChannel;
+
+import org.apache.mina.api.IdleStatus;
+import org.apache.mina.api.IoFuture;
+import org.apache.mina.api.IoSession;
+import org.apache.mina.service.executor.IoHandlerExecutor;
+import org.apache.mina.service.idlechecker.IndexedIdleChecker;
+import org.apache.mina.transport.ConnectFuture;
+import org.apache.mina.transport.udp.AbstractUdpClient;
+import org.apache.mina.transport.udp.UdpSessionConfig;
+import org.apache.mina.util.Assert;
+
+/**
+ * This class implements a UDP NIO based client.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ */
+public class NioUdpClient extends AbstractUdpClient {
+
+ /** the SelectorLoop for handling read/write session events */
+ // This is final, so that we know if it's not initialized
+ private final SelectorLoopPool readWriteSelectorPool;
+
+ private final IndexedIdleChecker idleChecker = new IndexedIdleChecker();
+
+ /**
+ * Create a new instance of NioUdpClient
+ */
+ public NioUdpClient() {
+ this(null);
+ }
+
+ /**
+ * Create a new instance of NioUdpClient
+ */
+ public NioUdpClient(IoHandlerExecutor ioHandlerExecutor) {
+ super(ioHandlerExecutor);
+ readWriteSelectorPool = new FixedSelectorLoopPool("Client", 2);
+ idleChecker.start();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public IoFuture<IoSession> connect(SocketAddress remoteAddress) throws IOException {
+ Assert.assertNotNull(remoteAddress, "remoteAddress");
+
+ DatagramChannel ch = DatagramChannel.open();
+ ch.configureBlocking(false);
+
+ UdpSessionConfig config = getSessionConfig();
+
+ NioSelectorLoop loop = (NioSelectorLoop) readWriteSelectorPool.getSelectorLoop();
+
+ NioUdpSession session = new NioUdpSession(this, idleChecker, ch, null, remoteAddress, loop);
+
+ session.setConnected();
+
+ // apply idle configuration
+ session.getConfig().setIdleTimeInMillis(IdleStatus.READ_IDLE, config.getIdleTimeInMillis(IdleStatus.READ_IDLE));
+ session.getConfig().setIdleTimeInMillis(IdleStatus.WRITE_IDLE,
+ config.getIdleTimeInMillis(IdleStatus.WRITE_IDLE));
+
+ // Manage the Idle status
+ idleChecker.sessionRead(session, System.currentTimeMillis());
+ idleChecker.sessionWritten(session, System.currentTimeMillis());
+
+ // apply the default service socket configuration
+
+ Boolean reuseAddress = config.isReuseAddress();
+
+ if (reuseAddress != null) {
+ session.getConfig().setReuseAddress(reuseAddress);
+ }
+
+ Integer readBufferSize = config.getReadBufferSize();
+
+ if (readBufferSize != null) {
+ session.getConfig().setReadBufferSize(readBufferSize);
+ } else {
+ int rcvBufferSize = ch.socket().getReceiveBufferSize();
+ session.getConfig().setReadBufferSize(rcvBufferSize);
+ }
+
+ Integer sendBufferSize = config.getSendBufferSize();
+
+ if (sendBufferSize != null) {
+ session.getConfig().setSendBufferSize(sendBufferSize);
+ } else {
+ int sndBufferSize = ch.socket().getSendBufferSize();
+ session.getConfig().setSendBufferSize(sndBufferSize);
+ }
+
+ Integer trafficClass = config.getTrafficClass();
+
+ if (trafficClass != null) {
+ session.getConfig().setTrafficClass(trafficClass);
+ }
+
+ loop.register(false, false, true, false, session, ch, null);
+
+ ConnectFuture cf = new ConnectFuture();
+ cf.complete(session);
+
+ return cf;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public IoFuture<IoSession> connect(SocketAddress remoteAddress, SocketAddress localAddress) {
+ throw new IllegalStateException("not supported for UDP");
+ }
+}
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/transport/nio/NioUdpServer.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/nio/NioUdpServer.java b/core/src/main/java/org/apache/mina/transport/nio/NioUdpServer.java
new file mode 100644
index 0000000..f7f1d40
--- /dev/null
+++ b/core/src/main/java/org/apache/mina/transport/nio/NioUdpServer.java
@@ -0,0 +1,293 @@
+/*
+ * 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.mina.transport.nio;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
+import java.nio.channels.SelectionKey;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.mina.api.IdleStatus;
+import org.apache.mina.service.executor.IoHandlerExecutor;
+import org.apache.mina.service.executor.OrderedHandlerExecutor;
+import org.apache.mina.service.idlechecker.IdleChecker;
+import org.apache.mina.service.idlechecker.IndexedIdleChecker;
+import org.apache.mina.transport.udp.AbstractUdpServer;
+import org.apache.mina.transport.udp.UdpSessionConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class implements a UDP NIO based server.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ */
+public class NioUdpServer extends AbstractUdpServer implements SelectorListener {
+
+ static final Logger LOG = LoggerFactory.getLogger(NioUdpServer.class);
+
+ // the bound local address
+ private SocketAddress address = null;
+
+ // used for detecting idle sessions
+ private final IdleChecker idleChecker = new IndexedIdleChecker();
+
+ // the inner channel for read/write UDP datagrams
+ private DatagramChannel datagramChannel = null;
+
+ // the key used for selecting read event
+ private SelectionKey readKey = null;
+
+ // list of all the sessions by remote socket address
+ private final Map<SocketAddress /* remote socket address */, NioUdpSession> sessions = new ConcurrentHashMap<SocketAddress, NioUdpSession>();
+
+ /** The selector loop used to incoming data */
+ private final SelectorLoop readSelectorLoop;
+
+ /**
+ * Create an UDP server with a new selector pool of default size and a {@link IoHandlerExecutor} of default type (
+ * {@link OrderedHandlerExecutor})
+ */
+ public NioUdpServer() {
+ this(new NioSelectorLoop("accept", 0), null);
+ }
+
+ /**
+ * Create an UDP server with a new selector pool of default size and a {@link IoHandlerExecutor} of default type (
+ * {@link OrderedHandlerExecutor})
+ *
+ * @param sessionConfig The configuration to use for this server
+ */
+ public NioUdpServer(UdpSessionConfig config) {
+ this(config, new NioSelectorLoop("accept", 0), null);
+ }
+
+ /**
+ * Create an UDP server with provided selector loops pool
+ *
+ * @param acceptSelectorLoop the selector loop for handling accept events (connection of new session)
+ * @param readWriteSelectorLoop the pool of selector loop for handling read/write events of connected sessions
+ * @param ioHandlerExecutor used for executing IoHandler event in another pool of thread (not in the low level I/O
+ * one). Use <code>null</code> if you don't want one. Be careful, the IoHandler processing will block the I/O
+ * operations.
+ */
+ public NioUdpServer(SelectorLoop readSelectorLoop, IoHandlerExecutor handlerExecutor) {
+ super(handlerExecutor);
+ this.readSelectorLoop = readSelectorLoop;
+ }
+
+ /**
+ * Create an UDP server with provided selector loops pool
+ *
+ * @param sessionConfig The configuration to use for this server
+ * @param acceptSelectorLoop the selector loop for handling accept events (connection of new session)
+ * @param ioHandlerExecutor used for executing IoHandler event in another pool of thread (not in the low level I/O
+ * one). Use <code>null</code> if you don't want one. Be careful, the IoHandler processing will block the I/O
+ * operations.
+ */
+ public NioUdpServer(UdpSessionConfig config, SelectorLoop readSelectorLoop, IoHandlerExecutor handlerExecutor) {
+ super(config, handlerExecutor);
+ this.readSelectorLoop = readSelectorLoop;
+ }
+
+ /**
+ * Get the inner datagram channel for read and write operations. To be called by the {@link NioSelectorProcessor}
+ *
+ * @return the datagram channel bound to this {@link NioUdpServer}.
+ */
+ public DatagramChannel getDatagramChannel() {
+ return datagramChannel;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public SocketAddress getBoundAddress() {
+ return address;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void bind(final int port) throws IOException {
+ bind(new InetSocketAddress(port));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void bind(final SocketAddress localAddress) throws IOException {
+ if (localAddress == null) {
+ // We should at least have one address to bind on
+ throw new IllegalArgumentException("LocalAdress cannot be null");
+ }
+
+ // check if the address is already bound
+ if (this.address != null) {
+ throw new IOException("address " + address + " already bound");
+ }
+ address = localAddress;
+
+ LOG.info("binding address {}", localAddress);
+
+ datagramChannel = DatagramChannel.open();
+
+ datagramChannel.socket().setReuseAddress(isReuseAddress());
+ datagramChannel.socket().bind(address);
+ datagramChannel.configureBlocking(false);
+
+ readSelectorLoop.register(false, false, true, false, this, datagramChannel, null);
+
+ // it's the first address bound, let's fire the event
+ this.fireServiceActivated();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void unbind() throws IOException {
+ LOG.info("unbinding {}", address);
+ if (this.address == null) {
+ throw new IllegalStateException("server not bound");
+ }
+
+ readSelectorLoop.unregister(this, datagramChannel);
+ datagramChannel.socket().close();
+ datagramChannel.close();
+
+ this.address = null;
+ this.fireServiceInactivated();
+ }
+
+ /**
+ * @return the readKey
+ */
+ public SelectionKey getReadKey() {
+ return readKey;
+ }
+
+ /**
+ * @param readKey the readKey to set
+ */
+ public void setReadKey(final SelectionKey readKey) {
+ this.readKey = readKey;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void ready(final boolean accept, boolean connect, final boolean read, final ByteBuffer readBuffer,
+ final boolean write) {
+ // Process the reads first
+ try {
+ System.err.println("remaining : " + readBuffer.remaining());
+ final SocketAddress source = datagramChannel.receive(readBuffer);
+ NioUdpSession session = null;
+
+ // let's find the corresponding session
+ if (source != null) {
+ session = sessions.get(source);
+
+ if (session == null) {
+ session = createSession(source, datagramChannel);
+ }
+ if (read) {
+ LOG.debug("readable datagram for UDP service : {}", this);
+ readBuffer.flip();
+
+ LOG.debug("read {} bytes form {}", readBuffer.remaining(), source);
+
+ session.receivedDatagram(readBuffer);
+
+ }
+
+ // Now, process the writes
+ if (write) {
+ session.processWrite(readSelectorLoop);
+ }
+ } else {
+ LOG.debug("Do data to read");
+ }
+
+ } catch (final IOException ex) {
+ LOG.error("IOException while reading the socket", ex);
+ }
+ }
+
+ private NioUdpSession createSession(SocketAddress remoteAddress, DatagramChannel datagramChannel)
+ throws IOException {
+ LOG.debug("create session");
+ UdpSessionConfig config = getSessionConfig();
+ SocketAddress localAddress = new InetSocketAddress(datagramChannel.socket().getLocalAddress(), datagramChannel
+ .socket().getLocalPort());
+ final NioUdpSession session = new NioUdpSession(this, idleChecker, datagramChannel, localAddress, remoteAddress);
+
+ // apply idle configuration
+ session.getConfig().setIdleTimeInMillis(IdleStatus.READ_IDLE, config.getIdleTimeInMillis(IdleStatus.READ_IDLE));
+ session.getConfig().setIdleTimeInMillis(IdleStatus.WRITE_IDLE,
+ config.getIdleTimeInMillis(IdleStatus.WRITE_IDLE));
+
+ // apply the default service socket configuration
+
+ Boolean reuseAddress = config.isReuseAddress();
+
+ if (reuseAddress != null) {
+ session.getConfig().setReuseAddress(reuseAddress);
+ }
+
+ Integer readBufferSize = config.getReadBufferSize();
+
+ if (readBufferSize != null) {
+ session.getConfig().setReadBufferSize(readBufferSize);
+ }
+
+ Integer sendBufferSize = config.getSendBufferSize();
+
+ if (sendBufferSize != null) {
+ session.getConfig().setSendBufferSize(sendBufferSize);
+ }
+
+ Integer trafficClass = config.getTrafficClass();
+
+ if (trafficClass != null) {
+ session.getConfig().setTrafficClass(trafficClass);
+ }
+
+ // Manage the Idle status
+ idleChecker.sessionRead(session, System.currentTimeMillis());
+ idleChecker.sessionWritten(session, System.currentTimeMillis());
+
+ sessions.put(remoteAddress, session);
+
+ // Inform the handler that the session has been created
+ session.setConnected();
+
+ return session;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/transport/nio/NioUdpSession.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/nio/NioUdpSession.java b/core/src/main/java/org/apache/mina/transport/nio/NioUdpSession.java
new file mode 100644
index 0000000..22ac9e7
--- /dev/null
+++ b/core/src/main/java/org/apache/mina/transport/nio/NioUdpSession.java
@@ -0,0 +1,311 @@
+/**
+ * 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.mina.transport.nio;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
+
+import org.apache.mina.api.IoService;
+import org.apache.mina.service.idlechecker.IdleChecker;
+import org.apache.mina.session.WriteRequest;
+import org.apache.mina.transport.udp.UdpSessionConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A UDP session based on NIO
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ */
+public class NioUdpSession extends AbstractNioSession implements SelectorListener {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NioUdpSession.class);
+
+ private final SocketAddress localAddress;
+
+ private final SocketAddress remoteAddress;
+
+ /**
+ * The selector loop in charge of generating read/write events for this session Used only for UDP client session.
+ */
+ private SelectorLoop selectorLoop = null;
+
+ /** the socket configuration */
+ private final UdpSessionConfig configuration;
+
+ /**
+ * For server handled UDP sessions
+ */
+ /* No qualifier */NioUdpSession(IoService service, IdleChecker idleChecker, DatagramChannel datagramChannel,
+ SocketAddress localAddress, SocketAddress remoteAddress) {
+ super(service, datagramChannel, idleChecker);
+ this.localAddress = localAddress;
+ this.remoteAddress = remoteAddress;
+ this.config = service.getSessionConfig();
+ this.configuration = (UdpSessionConfig) this.config;
+ }
+
+ /**
+ * For client handled UDP sessions
+ */
+ /* No qualifier */NioUdpSession(IoService service, IdleChecker idleChecker, DatagramChannel datagramChannel,
+ SocketAddress localAddress, SocketAddress remoteAddress, NioSelectorLoop selectorLoop) {
+ super(service, datagramChannel, idleChecker);
+ this.selectorLoop = selectorLoop;
+ this.localAddress = localAddress;
+ this.remoteAddress = remoteAddress;
+ this.config = service.getSessionConfig();
+ this.configuration = (UdpSessionConfig) this.config;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected void channelClose() {
+ LOG.debug("channelClose");
+ // No inner socket to close for UDP server, but some for UDP client
+ if (channel != null) {
+ try {
+ selectorLoop.unregister(this, channel);
+ channel.close();
+ } catch (final IOException e) {
+ LOG.error("Exception while closing the channel : ", e);
+ processException(e);
+ }
+ }
+ processSessionClosed();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void flushWriteQueue() {
+ // register for write
+ if (selectorLoop != null) {
+ selectorLoop.modifyRegistration(false, !isReadSuspended(), true, this, channel, true);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public SocketAddress getRemoteAddress() {
+ return remoteAddress;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public SocketAddress getLocalAddress() {
+ return localAddress;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ /*
+ * @Override public IoFuture<Void> close(boolean immediately) { switch (state) { case CREATED:
+ * LOG.error("Session {} not opened", this); throw new IllegalStateException("cannot close an not opened session");
+ * case CONNECTED: case CLOSING: if (immediately) { state = SessionState.CLOSED; } else { // we wait for the write
+ * queue to be depleted state = SessionState.CLOSING; } break; case CLOSED: LOG.warn("Already closed session {}",
+ * this); break; default: throw new IllegalStateException("not implemented session state : " + state); } return
+ * closeFuture; }
+ */
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void suspendRead() {
+ // TODO
+ throw new IllegalStateException("not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void suspendWrite() {
+ // TODO
+ throw new IllegalStateException("not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void resumeRead() {
+ // TODO
+ throw new IllegalStateException("not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void resumeWrite() {
+ // TODO
+ throw new IllegalStateException("not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isReadSuspended() {
+ // TODO
+ return false;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isWriteSuspended() {
+ // TODO
+ return false;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public UdpSessionConfig getConfig() {
+ return configuration;
+ }
+
+ /**
+ * Called when the session received a datagram.
+ *
+ * @param readBuffer the received datagram
+ */
+ void receivedDatagram(ByteBuffer readBuffer) {
+ processMessageReceived(readBuffer);
+ idleChecker.sessionRead(this, System.currentTimeMillis());
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected int writeDirect(Object message) {
+ try {
+ // Check that we can write into the channel
+ if (!isRegisteredForWrite()) {
+ // We don't have pending writes
+ // First, connect if we aren't already connected
+ if (!((DatagramChannel) channel).isConnected()) {
+ ((DatagramChannel) channel).connect(remoteAddress);
+ }
+
+ // And try to write the data. We will either write them all,
+ // or none
+ return ((DatagramChannel) channel).write((ByteBuffer) message);
+ } else {
+ LOG.debug("Cannot write");
+ return -1;
+ }
+ } catch (final IOException e) {
+ LOG.error("Exception while reading : ", e);
+ processException(e);
+
+ return -1;
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected ByteBuffer convertToDirectBuffer(WriteRequest writeRequest, boolean createNew) {
+ ByteBuffer message = (ByteBuffer) writeRequest.getMessage();
+
+ if (!message.isDirect()) {
+ int remaining = message.remaining();
+
+ ByteBuffer directBuffer = ByteBuffer.allocateDirect(remaining);
+ directBuffer.put(message);
+ directBuffer.flip();
+ writeRequest.setMessage(directBuffer);
+
+ return directBuffer;
+ }
+
+ return message;
+ }
+
+ /**
+ * Set this session status as connected. To be called by the processor selecting/polling this session.
+ */
+ void setConnected() {
+ if (!isCreated()) {
+ throw new IllegalStateException("Trying to open a non created session");
+ }
+
+ state = SessionState.CONNECTED;
+ processSessionOpen();
+ }
+
+ @Override
+ public void ready(boolean accept, boolean connect, boolean read, ByteBuffer readBuffer, boolean write) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("session {} ready for accept={}, connect={}, read={}, write={}", new Object[] { this, accept,
+ connect, read, write });
+ }
+
+ if (read) {
+ LOG.debug("readable datagram for UDP service : {}", this);
+
+ // Read everything we can up to the buffer size
+ try {
+ ((DatagramChannel) channel).receive(readBuffer);
+ readBuffer.flip();
+
+ int readbytes = readBuffer.remaining();
+ LOG.debug("read {} bytes", readbytes);
+ if (readbytes <= 0) {
+ // session closed by the remote peer
+ LOG.debug("session closed by the remote peer");
+ close(true);
+ } else {
+ receivedDatagram(readBuffer);
+ }
+ } catch (IOException e) {
+ processException(e);
+ }
+
+ }
+
+ if (write) {
+ processWrite(selectorLoop);
+ }
+ if (accept) {
+ throw new IllegalStateException("accept event should never occur on NioUdpSession");
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/transport/nio/RegistrationCallback.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/nio/RegistrationCallback.java b/core/src/main/java/org/apache/mina/transport/nio/RegistrationCallback.java
index e0d53dc..cb00b18 100644
--- a/core/src/main/java/org/apache/mina/transport/nio/RegistrationCallback.java
+++ b/core/src/main/java/org/apache/mina/transport/nio/RegistrationCallback.java
@@ -1,7 +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.mina.transport.nio;
import java.nio.channels.SelectionKey;
+/**
+ * To be used with {@link NioSelectorLoop} for being informed when the {@link SelectionKey} was created and registered.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ */
public interface RegistrationCallback {
void done(SelectionKey selectionKey);
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/transport/nio/SslHelper.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/nio/SslHelper.java b/core/src/main/java/org/apache/mina/transport/nio/SslHelper.java
new file mode 100644
index 0000000..d6e063f
--- /dev/null
+++ b/core/src/main/java/org/apache/mina/transport/nio/SslHelper.java
@@ -0,0 +1,421 @@
+/*
+ * 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.mina.transport.nio;
+
+import static org.apache.mina.session.AttributeKey.createKey;
+
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.util.Queue;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLEngineResult;
+import javax.net.ssl.SSLEngineResult.HandshakeStatus;
+import javax.net.ssl.SSLEngineResult.Status;
+import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLSession;
+
+import org.apache.mina.api.IoClient;
+import org.apache.mina.api.IoSession;
+import org.apache.mina.api.IoSession.SessionState;
+import org.apache.mina.session.AbstractIoSession;
+import org.apache.mina.session.AttributeKey;
+import org.apache.mina.session.DefaultWriteRequest;
+import org.apache.mina.session.WriteRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An helper class used to manage everything related to SSL/TLS establishment and management.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ */
+public class SslHelper {
+ /** A logger for this class */
+ private static final Logger LOGGER = LoggerFactory.getLogger(SslHelper.class);
+
+ /** The SSL engine instance */
+ private SSLEngine sslEngine;
+
+ /** The SSLContext instance */
+ private final SSLContext sslContext;
+
+ /** The current session */
+ private final IoSession session;
+
+ /**
+ * A session attribute key that should be set to an {@link InetSocketAddress}. Setting this attribute causes
+ * {@link SSLContext#createSSLEngine(String, int)} to be called passing the hostname and port of the
+ * {@link InetSocketAddress} to get an {@link SSLEngine} instance. If not set {@link SSLContext#createSSLEngine()}
+ * will be called.<br/>
+ * Using this feature {@link SSLSession} objects may be cached and reused when in client mode.
+ *
+ * @see SSLContext#createSSLEngine(String, int)
+ */
+ public static final AttributeKey<InetSocketAddress> PEER_ADDRESS = createKey(InetSocketAddress.class,
+ "internal_peerAddress");
+
+ public static final AttributeKey<Boolean> WANT_CLIENT_AUTH = createKey(Boolean.class, "internal_wantClientAuth");
+
+ public static final AttributeKey<Boolean> NEED_CLIENT_AUTH = createKey(Boolean.class, "internal_needClientAuth");
+
+ /** Incoming buffer accumulating bytes read from the channel */
+ /** An empty buffer used during the handshake phase */
+ private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0);
+
+ /** An empty buffer used during the handshake phase */
+ private static final ByteBuffer HANDSHAKE_BUFFER = ByteBuffer.allocate(1024);
+
+ /**
+ * Create a new SSL Handler.
+ *
+ * @param session The associated session
+ */
+ public SslHelper(IoSession session, SSLContext sslContext) {
+ this.session = session;
+ this.sslContext = sslContext;
+ }
+
+ /**
+ * @return The associated session
+ */
+ /* no qualifier */IoSession getSession() {
+ return session;
+ }
+
+ /**
+ * @return The associated SSLEngine
+ */
+ /* no qualifier */SSLEngine getEngine() {
+ return sslEngine;
+ }
+
+ /**
+ * Initialize the SSL handshake.
+ *
+ * @throws SSLException If the underlying SSLEngine handshake initialization failed
+ */
+ public void init() throws SSLException {
+ if (sslEngine != null) {
+ // We already have a SSL engine created, no need to create a new one
+ return;
+ }
+
+ LOGGER.debug("{} Initializing the SSL Helper", session);
+
+ InetSocketAddress peer = session.getAttribute(PEER_ADDRESS, null);
+
+ // Create the SSL engine here
+ if (peer == null) {
+ sslEngine = sslContext.createSSLEngine();
+ } else {
+ sslEngine = sslContext.createSSLEngine(peer.getHostName(), peer.getPort());
+ }
+
+ // Initialize the engine in client mode if necessary
+ sslEngine.setUseClientMode(session.getService() instanceof IoClient);
+
+ // Initialize the different SslEngine modes
+ if (!sslEngine.getUseClientMode()) {
+ // Those parameters are only valid when in server mode
+ boolean needClientAuth = session.getAttribute(NEED_CLIENT_AUTH, false);
+ boolean wantClientAuth = session.getAttribute(WANT_CLIENT_AUTH, false);
+
+ // The WantClientAuth supersede the NeedClientAuth, if set.
+ if (needClientAuth) {
+ sslEngine.setNeedClientAuth(true);
+ }
+
+ if (wantClientAuth) {
+ sslEngine.setWantClientAuth(true);
+ }
+ }
+
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("{} SSL Handler Initialization done.", session);
+ }
+ }
+
+ /**
+ * Process the NEED_TASK action.
+ *
+ * @param engine The SSLEngine instance
+ * @return The resulting HandshakeStatus
+ * @throws SSLException If we've got an error while processing the tasks
+ */
+ private HandshakeStatus processTasks(SSLEngine engine) throws SSLException {
+ Runnable runnable;
+
+ while ((runnable = engine.getDelegatedTask()) != null) {
+ // TODO : we may have to use a thread pool here to improve the
+ // performances
+ runnable.run();
+ }
+
+ HandshakeStatus hsStatus = engine.getHandshakeStatus();
+
+ return hsStatus;
+ }
+
+ /**
+ * Process the NEED_UNWRAP action. We have to read the incoming buffer, and to feed the application buffer.
+ */
+ private SSLEngineResult unwrap(ByteBuffer inBuffer, ByteBuffer appBuffer) throws SSLException {
+ // First work with either the new incoming buffer, or the accumulating buffer
+ ByteBuffer tempBuffer = inBuffer;
+
+ // Loop until we have processed the entire incoming buffer,
+ // or until we have to stop
+ while (true) {
+ // Do the unwrapping
+ SSLEngineResult result = sslEngine.unwrap(tempBuffer, appBuffer);
+
+ switch (result.getStatus()) {
+ case OK:
+ // Ok, we have unwrapped a message, return.
+ return result;
+
+ case BUFFER_UNDERFLOW:
+ // We need to read some more data from the channel.
+
+ inBuffer.clear();
+
+ return result;
+
+ case CLOSED:
+ // We have received a Close message, we can exit now
+ if (session.isConnectedSecured()) {
+ return result;
+ } else {
+ throw new IllegalStateException();
+ }
+
+ case BUFFER_OVERFLOW:
+ // We have to increase the appBuffer size. In any case
+ // we aren't processing an handshake here. Read again.
+ appBuffer = ByteBuffer.allocate(appBuffer.capacity() + 4096);
+ }
+ }
+ }
+
+ /**
+ * Process a read ByteBuffer over a secured connection, or during the SSL/TLS Handshake.
+ *
+ * @param session The session we are processing a read for
+ * @param readBuffer The data we get from the channel
+ * @throws SSLException If the unwrapping or handshaking failed
+ */
+ public void processRead(AbstractIoSession session, ByteBuffer readBuffer) throws SSLException {
+ if (session.isConnectedSecured()) {
+ // Unwrap the incoming data
+ while (readBuffer.hasRemaining()) {
+ processUnwrap(session, readBuffer);
+ }
+ } else {
+ // Process the SSL handshake now
+ processHandShake(session, readBuffer);
+ }
+ }
+
+ /**
+ * Unwrap a SSL/TLS message. The message might not be encrypted (if we are processing a Handshake message or an
+ * Alert message).
+ */
+ private void processUnwrap(AbstractIoSession session, ByteBuffer inBuffer) throws SSLException {
+ // Blind guess : once uncompressed, the resulting buffer will be 3 times bigger
+ ByteBuffer appBuffer = ByteBuffer.allocate(inBuffer.limit() * 3);
+ SSLEngineResult result = unwrap(inBuffer, appBuffer);
+
+ switch (result.getStatus()) {
+ case OK:
+ // Ok, go through the chain now
+ appBuffer.flip();
+ session.processMessageReceived(appBuffer);
+ break;
+
+ case CLOSED:
+ // This was a Alert Closure message. Process it
+ processClosed(result);
+
+ break;
+ }
+ }
+
+ /**
+ * Process the SSL/TLS Alert Closure message
+ */
+ private void processClosed(SSLEngineResult result) throws SSLException {
+ // We have received a Alert_CLosure message, we will have to do a wrap
+ HandshakeStatus hsStatus = result.getHandshakeStatus();
+
+ if (hsStatus == HandshakeStatus.NEED_WRAP) {
+ // We need to send back the Alert Closure message
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("{} processing the NEED_WRAP state", session);
+ }
+
+ int capacity = sslEngine.getSession().getPacketBufferSize();
+ ByteBuffer outBuffer = ByteBuffer.allocate(capacity);
+ session.changeState(SessionState.CONNECTED);
+
+ // Loop until the SSLEngine has nothing more to produce
+ while (!sslEngine.isOutboundDone()) {
+ sslEngine.wrap(EMPTY_BUFFER, outBuffer);
+ outBuffer.flip();
+
+ // Get out of the Connected state
+ WriteRequest writeRequest = new DefaultWriteRequest(outBuffer);
+ session.enqueueWriteRequest(writeRequest);
+ }
+ }
+ session.close(false);
+ }
+
+ /**
+ * Process the SLL/TLS Handshake. We may enter in this method more than once, as the handshake is a dialogue between
+ * the client and the server.
+ */
+ private void processHandShake(IoSession session, ByteBuffer inBuffer) throws SSLException {
+ // Start the Handshake if we aren't already processing a HandShake
+ // and switch to the SECURING state
+ HandshakeStatus hsStatus = sslEngine.getHandshakeStatus();
+
+ // Initilize the session status when we enter into the Handshake process.
+ // Not that we don't call the SSLEngine.beginHandshake() method :
+ // It's implicitely done internally by the unwrap() method.
+ if (hsStatus == HandshakeStatus.NOT_HANDSHAKING) {
+ session.changeState(SessionState.SECURING);
+ }
+
+ SSLEngineResult result = null;
+
+ // If the SSLEngine has not be started, then the status will be NOT_HANDSHAKING
+ // We loop until we reach the FINISHED state
+ while (hsStatus != HandshakeStatus.FINISHED) {
+ if (hsStatus == HandshakeStatus.NEED_TASK) {
+ hsStatus = processTasks(sslEngine);
+ } else if (hsStatus == HandshakeStatus.NEED_WRAP) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("{} processing the NEED_WRAP state", session);
+ }
+
+ // Create an insanely wide buffer, as the SSLEngine requires it
+ int capacity = sslEngine.getSession().getPacketBufferSize();
+ ByteBuffer outBuffer = ByteBuffer.allocate(capacity);
+
+ boolean completed = false;
+
+ // Loop until we are able to wrap the message (we may have
+ // to increase the buffer size more than once.
+ while (!completed) {
+ result = sslEngine.wrap(EMPTY_BUFFER, outBuffer);
+
+ switch (result.getStatus()) {
+ case OK:
+ case CLOSED:
+ completed = true;
+ break;
+
+ case BUFFER_OVERFLOW:
+ // Increase the target buffer size
+ outBuffer = ByteBuffer.allocate(outBuffer.capacity() + 4096);
+ break;
+ }
+ }
+
+ // Done. We can now push this buffer into the write queue.
+ outBuffer.flip();
+ WriteRequest writeRequest = new DefaultWriteRequest(inBuffer);
+ writeRequest.setMessage(outBuffer);
+ session.enqueueWriteRequest(writeRequest);
+ hsStatus = result.getHandshakeStatus();
+
+ // Nothing more to wrap : get out.
+ // Note to self : we can probably use only one ByteBuffer for the
+ // multiple wrapped messages. (see https://issues.apache.org/jira/browse/DIRMINA-878)
+ if (hsStatus != HandshakeStatus.NEED_WRAP) {
+ break;
+ }
+ } else if ((hsStatus == HandshakeStatus.NEED_UNWRAP) || (hsStatus == HandshakeStatus.NOT_HANDSHAKING)) {
+ // We cover the ongoing handshake (NEED_UNWRAP) and
+ // the initial call to the handshake (NOT_HANDSHAKING)
+ result = unwrap(inBuffer, HANDSHAKE_BUFFER);
+
+ if (result.getStatus() == Status.BUFFER_UNDERFLOW) {
+ // Read more data
+ break;
+ } else {
+ hsStatus = result.getHandshakeStatus();
+ }
+ }
+ }
+
+ if (hsStatus == HandshakeStatus.FINISHED) {
+ // The handshake has been completed. We can change the session's state.
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("{} processing the FINISHED state", session);
+ }
+
+ session.changeState(SessionState.SECURED);
+ }
+ }
+
+ /**
+ * Process the application data encryption for a session.
+ *
+ * @param session The session sending encrypted data to the peer.
+ * @param message The message to encrypt
+ * @param writeQueue The queue in which the encrypted buffer will be written
+ * @return The written WriteRequest
+ */
+ /** No qualifier */
+ WriteRequest processWrite(IoSession session, Object message, Queue<WriteRequest> writeQueue) {
+ ByteBuffer buf = (ByteBuffer) message;
+ ByteBuffer appBuffer = ByteBuffer.allocate(sslEngine.getSession().getPacketBufferSize());
+
+ try {
+ while (true) {
+ // Encrypt the message
+ SSLEngineResult result = sslEngine.wrap(buf, appBuffer);
+
+ switch (result.getStatus()) {
+ case BUFFER_OVERFLOW:
+ // Increase the buffer size as needed
+ appBuffer = ByteBuffer.allocate(appBuffer.capacity() + 4096);
+ break;
+
+ case BUFFER_UNDERFLOW:
+ case CLOSED:
+ break;
+
+ case OK:
+ // We are done. Flip the buffer and push it to the write queue.
+ appBuffer.flip();
+ WriteRequest request = new DefaultWriteRequest(appBuffer);
+
+ return request;
+ }
+ }
+ } catch (SSLException se) {
+ throw new IllegalStateException(se.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/transport/nio/tcp/NioTcpClient.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/nio/tcp/NioTcpClient.java b/core/src/main/java/org/apache/mina/transport/nio/tcp/NioTcpClient.java
deleted file mode 100644
index be35a62..0000000
--- a/core/src/main/java/org/apache/mina/transport/nio/tcp/NioTcpClient.java
+++ /dev/null
@@ -1,243 +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.mina.transport.nio.tcp;
-
-import java.io.IOException;
-import java.net.SocketAddress;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.SocketChannel;
-
-import org.apache.mina.api.IdleStatus;
-import org.apache.mina.api.IoFuture;
-import org.apache.mina.api.IoSession;
-import org.apache.mina.service.executor.IoHandlerExecutor;
-import org.apache.mina.service.executor.OrderedHandlerExecutor;
-import org.apache.mina.service.idlechecker.IdleChecker;
-import org.apache.mina.service.idlechecker.IndexedIdleChecker;
-import org.apache.mina.transport.nio.ConnectFuture;
-import org.apache.mina.transport.nio.FixedSelectorLoopPool;
-import org.apache.mina.transport.nio.NioSelectorLoop;
-import org.apache.mina.transport.nio.RegistrationCallback;
-import org.apache.mina.transport.nio.SelectorLoop;
-import org.apache.mina.transport.nio.SelectorLoopPool;
-import org.apache.mina.transport.tcp.AbstractTcpClient;
-import org.apache.mina.transport.tcp.TcpSessionConfig;
-import org.apache.mina.util.Assert;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * This class implements a TCP NIO based client.
- *
- * @author <a href="http://mina.apache.org">Apache MINA Project</a>
- */
-public class NioTcpClient extends AbstractTcpClient {
-
- /** A logger for this class */
- private static final Logger LOG = LoggerFactory.getLogger(NioTcpClient.class);
-
- /** the SelectorLoop for connecting the sessions */
- // This is final, so that we know if it's not initialized
- private final SelectorLoop connectSelectorLoop;
-
- /** the Selectorloop for handling read/write session events */
- // This is final, so that we know if it's not initialized
- private final SelectorLoopPool readWriteSelectorPool;
-
- /** for detecting idle session */
- private IdleChecker idleChecker;
-
- /**
- * Create a TCP client with new selector pool of default size and a {@link IoHandlerExecutor} of default type (
- * {@link OrderedHandlerExecutor})
- */
- public NioTcpClient() {
- // Default to 2 threads in the pool
- this(new NioSelectorLoop("connect", 0), new FixedSelectorLoopPool("Client", 2), null);
- }
-
- /**
- * Create a TCP client with provided selector loops pool. We will use one SelectorLoop get from the pool to manage
- * the OP_CONNECT events. If the pool contains only one SelectorLoop, then all the events will be managed by the
- * same Selector.
- *
- * @param readWriteSelectorLoop the pool of selector loop for handling read/write events of connected sessions
- * @param ioHandlerExecutor used for executing IoHandler event in another pool of thread (not in the low level I/O
- * one). Use <code>null</code> if you don't want one. Be careful, the IoHandler processing will block the I/O
- * operations.
- */
- public NioTcpClient(SelectorLoopPool selectorLoopPool, IoHandlerExecutor handlerExecutor) {
- super(handlerExecutor);
- connectSelectorLoop = selectorLoopPool.getSelectorLoop();
- readWriteSelectorPool = selectorLoopPool;
- }
-
- /**
- * Create a TCP client with provided selector loops pool
- *
- * @param connectSelectorLoop the selector loop for handling connection events (connection of new session)
- * @param readWriteSelectorLoop the pool of selector loop for handling read/write events of connected sessions
- * @param ioHandlerExecutor used for executing IoHandler event in another pool of thread (not in the low level I/O
- * one). Use <code>null</code> if you don't want one. Be careful, the IoHandler processing will block the I/O
- * operations.
- */
- public NioTcpClient(SelectorLoop connectSelectorLoop, SelectorLoopPool readWriteSelectorLoop,
- IoHandlerExecutor handlerExecutor) {
- super(handlerExecutor);
- this.connectSelectorLoop = connectSelectorLoop;
- this.readWriteSelectorPool = readWriteSelectorLoop;
- idleChecker = new IndexedIdleChecker();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public IoFuture<IoSession> connect(SocketAddress remoteAddress) throws IOException {
- Assert.assertNotNull(remoteAddress, "remoteAddress");
-
- SocketChannel clientSocket = SocketChannel.open();
-
- clientSocket.socket().setSoTimeout(getConnectTimeoutMillis());
-
- // non blocking
- clientSocket.configureBlocking(false);
-
- // apply idle configuration
- // Has to be final, as it's used in a inner class...
- final NioTcpSession session = new NioTcpSession(this, clientSocket, readWriteSelectorPool.getSelectorLoop(),
- idleChecker);
- TcpSessionConfig config = getSessionConfig();
-
- session.getConfig().setIdleTimeInMillis(IdleStatus.READ_IDLE, config.getIdleTimeInMillis(IdleStatus.READ_IDLE));
- session.getConfig().setIdleTimeInMillis(IdleStatus.WRITE_IDLE,
- config.getIdleTimeInMillis(IdleStatus.WRITE_IDLE));
-
- // apply the default service socket configuration
- Boolean keepAlive = config.isKeepAlive();
-
- if (keepAlive != null) {
- session.getConfig().setKeepAlive(keepAlive);
- }
-
- Boolean oobInline = config.isOobInline();
-
- if (oobInline != null) {
- session.getConfig().setOobInline(oobInline);
- }
-
- Boolean reuseAddress = config.isReuseAddress();
-
- if (reuseAddress != null) {
- session.getConfig().setReuseAddress(reuseAddress);
- }
-
- Boolean tcpNoDelay = config.isTcpNoDelay();
-
- if (tcpNoDelay != null) {
- session.getConfig().setTcpNoDelay(tcpNoDelay);
- }
-
- Integer receiveBufferSize = config.getReadBufferSize();
-
- if (receiveBufferSize != null) {
- session.getConfig().setReadBufferSize(receiveBufferSize);
- } else {
- int rcvBufferSize = clientSocket.socket().getReceiveBufferSize();
- session.getConfig().setReadBufferSize(rcvBufferSize);
- }
-
- Integer sendBufferSize = config.getSendBufferSize();
-
- if (sendBufferSize != null) {
- session.getConfig().setSendBufferSize(sendBufferSize);
- } else {
- int sndBufferSize = clientSocket.socket().getSendBufferSize();
- session.getConfig().setSendBufferSize(sndBufferSize);
- }
-
- Integer trafficClass = config.getTrafficClass();
-
- if (trafficClass != null) {
- session.getConfig().setTrafficClass(trafficClass);
- }
-
- Integer soLinger = config.getSoLinger();
-
- if (soLinger != null) {
- session.getConfig().setSoLinger(soLinger);
- }
-
- // Set the secured flag if the service is to be used over SSL/TLS
- if (config.isSecured()) {
- session.initSecure(config.getSslContext());
- }
-
- // connect to a running server. We get an immediate result if
- // the socket is blocking, and either true or false if it's non blocking
- boolean connected = clientSocket.connect(remoteAddress);
-
- ConnectFuture connectFuture = new ConnectFuture();
- session.setConnectFuture(connectFuture);
-
- if (!connected) {
- // async connection, let's the connection complete in background, the selector loop will detect when the
- // connection is successful
- connectSelectorLoop.register(false, true, false, false, session, clientSocket, new RegistrationCallback() {
-
- @Override
- public void done(SelectionKey selectionKey) {
- session.setSelectionKey(selectionKey);
- }
- });
- } else {
- // already connected (probably a loopback connection, or a blocking socket)
- // register for read
- connectSelectorLoop.register(false, false, true, false, session, clientSocket, new RegistrationCallback() {
-
- @Override
- public void done(SelectionKey selectionKey) {
- session.setSelectionKey(selectionKey);
- }
- });
-
- session.setConnected();
- }
-
- return connectFuture;
- }
-
- /**
- * {@inheritDoc}
- */
- public synchronized void disconnect() throws IOException {
- LOG.info("Disconnecting sessions");
-
- // Close all the existing sessions
- for (IoSession session : getManagedSessions().values()) {
- session.close(true);
- }
-
- fireServiceInactivated();
-
- // will stop the idle processor if we are the last service
- idleChecker.destroy();
- }
-}
\ No newline at end of file