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:32 UTC
[3/5] git commit: adding BIO based UDP server (WIP)
adding BIO based UDP server (WIP)
Project: http://git-wip-us.apache.org/repos/asf/mina/repo
Commit: http://git-wip-us.apache.org/repos/asf/mina/commit/8ee4890b
Tree: http://git-wip-us.apache.org/repos/asf/mina/tree/8ee4890b
Diff: http://git-wip-us.apache.org/repos/asf/mina/diff/8ee4890b
Branch: refs/heads/trunk
Commit: 8ee4890b444f3315ab5cc960d905edc83339d82b
Parents: e249f4e
Author: jvermillard <jv...@apache.org>
Authored: Sun May 26 17:06:59 2013 +0200
Committer: jvermillard <jv...@apache.org>
Committed: Sun May 26 17:06:59 2013 +0200
----------------------------------------------------------------------
.../mina/core/nio/tcp/Mina3TcpBenchmarkClient.java | 2 +-
.../mina/core/nio/tcp/Mina3TcpBenchmarkServer.java | 2 +-
.../mina/core/nio/udp/Mina3UdpBenchmarkClient.java | 2 +-
.../mina/core/nio/udp/Mina3UdpBenchmarkServer.java | 7 +-
.../java/org/apache/mina/coap/CoapMessage.java | 42 ++-
.../org/apache/mina/coap/codec/CoapDecoder.java | 43 +-
.../org/apache/mina/coap/codec/CoapEncoder.java | 16 +-
.../main/java/org/apache/mina/api/IoSession.java | 9 +-
.../org/apache/mina/session/AbstractIoSession.java | 319 +-----------
.../java/org/apache/mina/session/SslHelper.java | 422 ---------------
.../org/apache/mina/transport/ConnectFuture.java | 52 ++
.../apache/mina/transport/bio/BioUdpServer.java | 164 ++++++
.../apache/mina/transport/bio/BioUdpSession.java | 158 ++++++
.../mina/transport/nio/AbstractNioSession.java | 331 +++++++++++
.../apache/mina/transport/nio/ConnectFuture.java | 26 -
.../apache/mina/transport/nio/NioTcpClient.java | 238 ++++++++
.../apache/mina/transport/nio/NioTcpServer.java | 345 ++++++++++++
.../apache/mina/transport/nio/NioTcpSession.java | 386 +++++++++++++
.../apache/mina/transport/nio/NioUdpClient.java | 139 +++++
.../apache/mina/transport/nio/NioUdpServer.java | 293 ++++++++++
.../apache/mina/transport/nio/NioUdpSession.java | 311 +++++++++++
.../mina/transport/nio/RegistrationCallback.java | 24 +
.../org/apache/mina/transport/nio/SslHelper.java | 421 ++++++++++++++
.../mina/transport/nio/tcp/NioTcpClient.java | 243 ---------
.../mina/transport/nio/tcp/NioTcpServer.java | 351 ------------
.../mina/transport/nio/tcp/NioTcpSession.java | 391 -------------
.../mina/transport/nio/udp/NioUdpClient.java | 142 -----
.../mina/transport/nio/udp/NioUdpServer.java | 295 ----------
.../mina/transport/nio/udp/NioUdpSession.java | 328 -----------
.../mina/transport/vmpipe/VmPipeAddress.java | 82 ---
.../mina/transport/tcp/IdleTcpServerTest.java | 2 +-
.../transport/tcp/NioTcpClientFilterEventTest.java | 2 +-
.../transport/tcp/NioTcpClientReleaseTest.java | 4 +-
.../transport/tcp/NioTcpClientTimeoutTest.java | 2 +-
.../transport/tcp/NioTcpServerFilterEventTest.java | 2 +-
.../transport/tcp/NioTcpServerHandlerTest.java | 2 +-
.../org/apache/mina/transport/tcp/SslTest.java | 2 +-
.../transport/udp/NioUdpClientFilterEventTest.java | 2 +-
.../transport/udp/NioUdpServerFilterEventTest.java | 201 +++++++
.../apache/mina/examples/coap/CoapGetServer.java | 5 +-
.../mina/examples/echoclient/NioEchoClient.java | 2 +-
.../mina/examples/echoserver/NioEchoServer.java | 2 +-
.../org/apache/mina/examples/http/HttpTest.java | 2 +-
.../org/apache/mina/examples/http/HttpsTest.java | 2 +-
.../mina/examples/udpecho/NioUdpEchoServer.java | 2 +-
45 files changed, 3158 insertions(+), 2660 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/benchmarks/src/test/java/org/apache/mina/core/nio/tcp/Mina3TcpBenchmarkClient.java
----------------------------------------------------------------------
diff --git a/benchmarks/src/test/java/org/apache/mina/core/nio/tcp/Mina3TcpBenchmarkClient.java b/benchmarks/src/test/java/org/apache/mina/core/nio/tcp/Mina3TcpBenchmarkClient.java
index b8153fd..413d3f9 100755
--- a/benchmarks/src/test/java/org/apache/mina/core/nio/tcp/Mina3TcpBenchmarkClient.java
+++ b/benchmarks/src/test/java/org/apache/mina/core/nio/tcp/Mina3TcpBenchmarkClient.java
@@ -29,7 +29,7 @@ import org.apache.mina.api.IoHandler;
import org.apache.mina.api.IoService;
import org.apache.mina.api.IoSession;
import org.apache.mina.core.BenchmarkClient;
-import org.apache.mina.transport.nio.tcp.NioTcpClient;
+import org.apache.mina.transport.nio.NioTcpClient;
/**
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/benchmarks/src/test/java/org/apache/mina/core/nio/tcp/Mina3TcpBenchmarkServer.java
----------------------------------------------------------------------
diff --git a/benchmarks/src/test/java/org/apache/mina/core/nio/tcp/Mina3TcpBenchmarkServer.java b/benchmarks/src/test/java/org/apache/mina/core/nio/tcp/Mina3TcpBenchmarkServer.java
index 52989ca..9bf550b 100755
--- a/benchmarks/src/test/java/org/apache/mina/core/nio/tcp/Mina3TcpBenchmarkServer.java
+++ b/benchmarks/src/test/java/org/apache/mina/core/nio/tcp/Mina3TcpBenchmarkServer.java
@@ -29,7 +29,7 @@ import org.apache.mina.api.IoService;
import org.apache.mina.api.IoSession;
import org.apache.mina.core.BenchmarkServer;
import org.apache.mina.session.AttributeKey;
-import org.apache.mina.transport.nio.tcp.NioTcpServer;
+import org.apache.mina.transport.nio.NioTcpServer;
/**
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/benchmarks/src/test/java/org/apache/mina/core/nio/udp/Mina3UdpBenchmarkClient.java
----------------------------------------------------------------------
diff --git a/benchmarks/src/test/java/org/apache/mina/core/nio/udp/Mina3UdpBenchmarkClient.java b/benchmarks/src/test/java/org/apache/mina/core/nio/udp/Mina3UdpBenchmarkClient.java
index 0af1ea2..286dbf8 100755
--- a/benchmarks/src/test/java/org/apache/mina/core/nio/udp/Mina3UdpBenchmarkClient.java
+++ b/benchmarks/src/test/java/org/apache/mina/core/nio/udp/Mina3UdpBenchmarkClient.java
@@ -31,7 +31,7 @@ import org.apache.mina.api.IoHandler;
import org.apache.mina.api.IoService;
import org.apache.mina.api.IoSession;
import org.apache.mina.core.BenchmarkClient;
-import org.apache.mina.transport.nio.udp.NioUdpClient;
+import org.apache.mina.transport.nio.NioUdpClient;
/**
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/benchmarks/src/test/java/org/apache/mina/core/nio/udp/Mina3UdpBenchmarkServer.java
----------------------------------------------------------------------
diff --git a/benchmarks/src/test/java/org/apache/mina/core/nio/udp/Mina3UdpBenchmarkServer.java b/benchmarks/src/test/java/org/apache/mina/core/nio/udp/Mina3UdpBenchmarkServer.java
index 1e08aa2..cfb85cb 100755
--- a/benchmarks/src/test/java/org/apache/mina/core/nio/udp/Mina3UdpBenchmarkServer.java
+++ b/benchmarks/src/test/java/org/apache/mina/core/nio/udp/Mina3UdpBenchmarkServer.java
@@ -29,7 +29,7 @@ import org.apache.mina.api.IoService;
import org.apache.mina.api.IoSession;
import org.apache.mina.core.BenchmarkServer;
import org.apache.mina.session.AttributeKey;
-import org.apache.mina.transport.nio.udp.NioUdpServer;
+import org.apache.mina.transport.nio.NioUdpServer;
import org.apache.mina.transport.udp.DefaultUdpSessionConfig;
import org.apache.mina.transport.udp.UdpSessionConfig;
@@ -60,15 +60,18 @@ public class Mina3UdpBenchmarkServer implements BenchmarkServer {
/**
* {@inheritDoc}
*/
+ @Override
public void start(int port) throws IOException {
UdpSessionConfig config = new DefaultUdpSessionConfig();
config.setReadBufferSize(65536);
udpServer = new NioUdpServer(config);
udpServer.setIoHandler(new IoHandler() {
+ @Override
public void sessionOpened(IoSession session) {
session.setAttribute(STATE_ATTRIBUTE, State.WAIT_FOR_FIRST_BYTE_LENGTH);
}
+ @Override
public void messageReceived(IoSession session, Object message) {
//System.out.println("Server Message received : " + message);
if (message instanceof ByteBuffer) {
@@ -121,6 +124,7 @@ public class Mina3UdpBenchmarkServer implements BenchmarkServer {
}
}
+ @Override
public void exceptionCaught(IoSession session, Exception cause) {
cause.printStackTrace();
}
@@ -153,6 +157,7 @@ public class Mina3UdpBenchmarkServer implements BenchmarkServer {
/**
* {@inheritedDoc}
*/
+ @Override
public void stop() throws IOException {
udpServer.unbind();
}
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/coap/src/main/java/org/apache/mina/coap/CoapMessage.java
----------------------------------------------------------------------
diff --git a/coap/src/main/java/org/apache/mina/coap/CoapMessage.java b/coap/src/main/java/org/apache/mina/coap/CoapMessage.java
index 5a1500d..a366284 100644
--- a/coap/src/main/java/org/apache/mina/coap/CoapMessage.java
+++ b/coap/src/main/java/org/apache/mina/coap/CoapMessage.java
@@ -31,19 +31,19 @@ public class CoapMessage {
private static final byte[] EMPTY_BYTE_ARRAY = new byte[] {};
- private final int version;
+ private int version;
- private final MessageType type;
+ private MessageType type;
- private final int code;
+ private int code;
- private final int id;
+ private int id;
- private final byte[] token;
+ private byte[] token;
- private final byte[] payload;
+ private byte[] payload;
- private final CoapOption[] options;
+ private CoapOption[] options;
/**
* Create a CoAP message
@@ -188,4 +188,32 @@ public class CoapMessage {
return true;
}
+ public void setVersion(int version) {
+ this.version = version;
+ }
+
+ public void setType(MessageType type) {
+ this.type = type;
+ }
+
+ public void setCode(int code) {
+ this.code = code;
+ }
+
+ public void setId(int id) {
+ this.id = id;
+ }
+
+ public void setToken(byte[] token) {
+ this.token = token;
+ }
+
+ public void setPayload(byte[] payload) {
+ this.payload = payload;
+ }
+
+ public void setOptions(CoapOption[] options) {
+ this.options = options;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/coap/src/main/java/org/apache/mina/coap/codec/CoapDecoder.java
----------------------------------------------------------------------
diff --git a/coap/src/main/java/org/apache/mina/coap/codec/CoapDecoder.java b/coap/src/main/java/org/apache/mina/coap/codec/CoapDecoder.java
index 65dc941..772a107 100644
--- a/coap/src/main/java/org/apache/mina/coap/codec/CoapDecoder.java
+++ b/coap/src/main/java/org/apache/mina/coap/codec/CoapDecoder.java
@@ -29,7 +29,6 @@ import org.apache.mina.coap.CoapOptionType;
import org.apache.mina.coap.MessageType;
import org.apache.mina.codec.ProtocolDecoderException;
import org.apache.mina.codec.StatelessProtocolDecoder;
-import org.apache.mina.util.ByteBufferDumper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -61,27 +60,27 @@ public class CoapDecoder implements StatelessProtocolDecoder<ByteBuffer, CoapMes
*/
@Override
public CoapMessage decode(ByteBuffer input, Void context) {
- LOG.debug("decode");
+ // LOG.debug("decode");
if (input.remaining() <= 0) {
- LOG.debug("nothing to decode");
+ // LOG.debug("nothing to decode");
return null;
}
int byte0 = input.get() & 0xFF;
int version = (byte0 >> 6) & 0x3;
- LOG.debug("version : {}", version);
+ // LOG.debug("version : {}", version);
MessageType type = MessageType.fromCode((byte0 >> 4) & 0x3);
- LOG.debug("type : {}", type);
+ // LOG.debug("type : {}", type);
byte[] token = new byte[byte0 & 0xF];
int code = input.get() & 0xFF;
- LOG.debug("code : {}", code);
+ // LOG.debug("code : {}", code);
int id = input.getShort() & 0xFFFF;
- LOG.debug("id : {}", id);
+ // LOG.debug("id : {}", id);
input.get(token);
- if (LOG.isDebugEnabled()) {
- LOG.debug("token : {}", ByteBufferDumper.toHex(ByteBuffer.wrap(token)));
- }
+ // if (LOG.isDebugEnabled()) {
+ // LOG.debug("token : {}", ByteBufferDumper.toHex(ByteBuffer.wrap(token)));
+ // }
// decode options
int optionCode = 0;
@@ -92,28 +91,28 @@ public class CoapDecoder implements StatelessProtocolDecoder<ByteBuffer, CoapMes
// start of payload ?
if (next == 0xFF) {
- LOG.debug("start of payload");
+ // LOG.debug("start of payload");
payload = new byte[input.remaining()];
input.get(payload);
- if (LOG.isDebugEnabled()) {
- LOG.debug("payload : {}", ByteBufferDumper.dump(ByteBuffer.wrap(payload)));
- }
+ // if (LOG.isDebugEnabled()) {
+ // LOG.debug("payload : {}", ByteBufferDumper.dump(ByteBuffer.wrap(payload)));
+ // }
break;
} else {
int optionDeltaQuartet = (next >> 4) & 0xF;
// decode the option type
optionCode += optionFromQuartet(optionDeltaQuartet, input);
- LOG.debug("optionCode : {}", optionCode);
+ // LOG.debug("optionCode : {}", optionCode);
// decode the option length
int optionLenQuartet = next & 0x0F;
int optionLength = optionFromQuartet(optionLenQuartet, input);
- LOG.debug("optionLength : {}", optionLength);
+ // LOG.debug("optionLength : {}", optionLength);
// create the option DTO
CoapOptionType optType = CoapOptionType.fromCode(optionCode);
- LOG.debug("option type : {}", optType);
+ // LOG.debug("option type : {}", optType);
// get the value
byte[] optionValue = new byte[optionLength];
@@ -138,14 +137,14 @@ public class CoapDecoder implements StatelessProtocolDecoder<ByteBuffer, CoapMes
}
private int optionFromQuartet(int value, ByteBuffer input) {
- LOG.debug("quartet value : {}", value);
+ // LOG.debug("quartet value : {}", value);
if (value < 13) {
return value;
} else if (value == 13) {
- if (LOG.isDebugEnabled()) {
- int val = input.get(input.position()) & 0xFF;
- LOG.debug("byte : {}", val);
- }
+ // if (LOG.isDebugEnabled()) {
+ int val = input.get(input.position()) & 0xFF;
+ // LOG.debug("byte : {}", val);
+ // }
return (input.get() & 0xFF) + 13;
} else if (value == 14) {
return (input.getShort() & 0xFFFF) + 269;
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/coap/src/main/java/org/apache/mina/coap/codec/CoapEncoder.java
----------------------------------------------------------------------
diff --git a/coap/src/main/java/org/apache/mina/coap/codec/CoapEncoder.java b/coap/src/main/java/org/apache/mina/coap/codec/CoapEncoder.java
index ce0e0d8..1e15da3 100644
--- a/coap/src/main/java/org/apache/mina/coap/codec/CoapEncoder.java
+++ b/coap/src/main/java/org/apache/mina/coap/codec/CoapEncoder.java
@@ -55,7 +55,7 @@ public class CoapEncoder implements StatelessProtocolEncoder<CoapMessage, ByteBu
@Override
public ByteBuffer encode(CoapMessage message, Void context) {
- LOG.debug("encoding {}", message);
+ // LOG.debug("encoding {}", message);
// compute size of the needed buffer
int size = HEADER_SIZE + message.getToken().length;
@@ -94,7 +94,7 @@ public class CoapEncoder implements StatelessProtocolEncoder<CoapMessage, ByteBu
size += message.getPayload().length;
}
- LOG.debug("computed size : {}", size);
+ // LOG.debug("computed size : {}", size);
ByteBuffer buffer = ByteBuffer.allocate(size);
buffer.order(ByteOrder.BIG_ENDIAN);
@@ -110,14 +110,14 @@ public class CoapEncoder implements StatelessProtocolEncoder<CoapMessage, ByteBu
int lastOptCode = 0;
for (CoapOption opt : message.getOptions()) {
- LOG.debug("encode option {}", opt);
+ // LOG.debug("encode option {}", opt);
int optionDelta = opt.getType().getCode() - lastOptCode;
int deltaQuartet = getQuartet(optionDelta);
int optionLength = opt.getData().length;
int optionQuartet = getQuartet(optionLength);
- LOG.debug("optionQuartet : {}", optionQuartet);
+ // LOG.debug("optionQuartet : {}", optionQuartet);
buffer.put((byte) ((deltaQuartet << 4) | optionQuartet));
@@ -128,11 +128,11 @@ public class CoapEncoder implements StatelessProtocolEncoder<CoapMessage, ByteBu
buffer.putShort((short) (optionDelta - 269));
}
- LOG.debug("option length : {}", optionLength);
+ // LOG.debug("option length : {}", optionLength);
if (optionQuartet == 13) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("length byte : " + (optionLength - 13));
- }
+ // if (LOG.isDebugEnabled()) {
+ // LOG.debug("length byte : " + (optionLength - 13));
+ // }
buffer.put((byte) (optionLength - 13));
} else if (optionQuartet == 14) {
buffer.putShort((short) (optionLength - 269));
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/api/IoSession.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/api/IoSession.java b/core/src/main/java/org/apache/mina/api/IoSession.java
index 09ba321..4b4cf64 100644
--- a/core/src/main/java/org/apache/mina/api/IoSession.java
+++ b/core/src/main/java/org/apache/mina/api/IoSession.java
@@ -21,16 +21,15 @@ package org.apache.mina.api;
import java.net.SocketAddress;
import java.util.Collections;
-import java.util.Queue;
import java.util.Set;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLException;
import org.apache.mina.session.AttributeKey;
-import org.apache.mina.session.SslHelper;
import org.apache.mina.session.WriteRequest;
import org.apache.mina.transport.nio.SelectorLoop;
+import org.apache.mina.transport.nio.SslHelper;
/**
* A handle which represents a connection between two end-points regardless of transport types.
@@ -363,10 +362,4 @@ public interface IoSession {
*/
WriteRequest enqueueWriteRequest(WriteRequest writeRequest);
- /**
- * Get the {@link Queue} of this session. The write queue contains the pending writes.
- *
- * @return the write queue of this session
- */
- Queue<WriteRequest> getWriteQueue();
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/session/AbstractIoSession.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/session/AbstractIoSession.java b/core/src/main/java/org/apache/mina/session/AbstractIoSession.java
index 80b9664..ab78858 100644
--- a/core/src/main/java/org/apache/mina/session/AbstractIoSession.java
+++ b/core/src/main/java/org/apache/mina/session/AbstractIoSession.java
@@ -19,14 +19,9 @@
*/
package org.apache.mina.session;
-import java.io.IOException;
import java.nio.ByteBuffer;
-import java.nio.channels.SelectableChannel;
-import java.nio.channels.SocketChannel;
import java.util.Collections;
-import java.util.Queue;
import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
@@ -51,9 +46,8 @@ import org.apache.mina.service.executor.OpenEvent;
import org.apache.mina.service.executor.ReceiveEvent;
import org.apache.mina.service.executor.SentEvent;
import org.apache.mina.service.idlechecker.IdleChecker;
-import org.apache.mina.transport.nio.SelectorListener;
import org.apache.mina.transport.nio.SelectorLoop;
-import org.apache.mina.util.AbstractIoFuture;
+import org.apache.mina.transport.nio.SslHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -66,9 +60,6 @@ public abstract class AbstractIoSession implements IoSession, ReadFilterChainCon
/** The logger for this class */
private static final Logger LOG = LoggerFactory.getLogger(AbstractIoSession.class);
- /** the NIO channel for this session */
- protected final SelectableChannel channel;
-
/** unique identifier generator */
private static final AtomicInteger NEXT_ID = new AtomicInteger(0);
@@ -125,16 +116,6 @@ public abstract class AbstractIoSession implements IoSession, ReadFilterChainCon
/** Tells if the session is secured or not */
protected volatile boolean secured;
- /** is this session registered for being polled for write ready events */
- private final AtomicBoolean registeredForWrite = new AtomicBoolean();
-
- // ------------------------------------------------------------------------
- // Write queue
- // ------------------------------------------------------------------------
-
- /** the queue of pending writes for the session, to be dequeued by the {@link SelectorLoop} */
- private final Queue<WriteRequest> writeQueue = new DefaultWriteQueue();
-
// ------------------------------------------------------------------------
// Filter chain
// ------------------------------------------------------------------------
@@ -153,9 +134,9 @@ public abstract class AbstractIoSession implements IoSession, ReadFilterChainCon
* {@link org.apache.mina.api.IoSession#getId()}) and an associated {@link IoService}
*
* @param service the service this session is associated with
- * @param selectorLoop the selector loop in charge of processing this session read/write events
+ * @param idleChecker the checker for idle session
*/
- public AbstractIoSession(final IoService service, SelectableChannel channel, final IdleChecker idleChecker) {
+ public AbstractIoSession(final IoService service, final IdleChecker idleChecker) {
// generated a unique id
id = NEXT_ID.getAndIncrement();
creationTime = System.currentTimeMillis();
@@ -163,7 +144,6 @@ public abstract class AbstractIoSession implements IoSession, ReadFilterChainCon
this.chain = service.getFilters();
this.idleChecker = idleChecker;
this.config = service.getSessionConfig();
- this.channel = channel;
LOG.debug("Created new session with id : {}", id);
@@ -536,173 +516,6 @@ public abstract class AbstractIoSession implements IoSession, ReadFilterChainCon
processMessageWriting(writeRequest, future);
}
- /**
- * Writes the message immediately. If we can't write all the message, we will get back the number of written bytes.
- *
- * @param message the message to write
- * @return the number of written bytes
- */
- protected abstract int writeDirect(Object message);
-
- /**
- * Copy the HeapBuffer into a DirectBuffer, if needed.
- *
- * @param writeRequest The request containing the HeapBuffer
- * @param createNew A flag to force the creation of a DirectBuffer
- * @return A DirectBuffer
- */
- protected abstract ByteBuffer convertToDirectBuffer(WriteRequest writeRequest, boolean createNew);
-
- /**
- * {@inheritDoc}
- */
- @Override
- public WriteRequest enqueueWriteRequest(WriteRequest writeRequest) {
- if (isConnectedSecured()) {
- // SSL/TLS : we have to encrypt the message
- final SslHelper sslHelper = getAttribute(SSL_HELPER, null);
-
- if (sslHelper == null) {
- throw new IllegalStateException();
- }
-
- writeRequest = sslHelper.processWrite(this, writeRequest.getMessage(), writeQueue);
- }
-
- synchronized (writeQueue) {
- ByteBuffer message = (ByteBuffer) writeRequest.getMessage();
-
- if (writeQueue.isEmpty()) {
- // Transfer the buffer in a DirectByteBuffer if it's a HeapByteBuffer and if it's too big
- message = convertToDirectBuffer(writeRequest, false);
-
- // We don't have anything in the writeQueue, let's try to write the
- // data in the channel immediately if we can
- int written = writeDirect(writeRequest.getMessage());
-
- LOG.debug("wrote {} bytes to {}", written, this);
-
- if (written > 0) {
- incrementWrittenBytes(written);
- }
-
- // Update the idle status for this session
- idleChecker.sessionWritten(this, System.currentTimeMillis());
- int remaining = message.remaining();
-
- if ((written < 0) || (remaining > 0)) {
- // Create a DirectBuffer unconditionally
- convertToDirectBuffer(writeRequest, true);
-
- // We have to push the request on the writeQueue
- writeQueue.add(writeRequest);
-
- // If it wasn't, we register this session as interested to write.
- // It's done in atomic fashion for avoiding two concurrent registering.
- if (!registeredForWrite.getAndSet(true)) {
- flushWriteQueue();
- }
- } else {
- // The message has been fully written : update the stats, and signal the handler
- // generate the message sent event
- // complete the future if we have one (we should...)
- final DefaultWriteFuture future = (DefaultWriteFuture) writeRequest.getFuture();
-
- if (future != null) {
- future.complete();
- }
-
- final Object highLevel = ((DefaultWriteRequest) writeRequest).getOriginalMessage();
-
- if (highLevel != null) {
- processMessageSent(highLevel);
- }
- }
- } else {
- // Transfer the buffer in a DirectByteBuffer if it's a HeapByteBuffer
- message = convertToDirectBuffer(writeRequest, true);
-
- // We have to push the request on the writeQueue
- writeQueue.add(writeRequest);
- }
- }
-
- return writeRequest;
- }
-
- public abstract void flushWriteQueue();
-
- public void setNotRegisteredForWrite() {
- registeredForWrite.set(false);
- }
-
- protected boolean isRegisteredForWrite() {
- return registeredForWrite.get();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public Queue<WriteRequest> getWriteQueue() {
- return writeQueue;
- }
-
- // ------------------------------------------------------------------------
- // Close session management
- // ------------------------------------------------------------------------
-
- /** we pre-allocate a close future for lock-less {@link #close(boolean)} */
- private final IoFuture<Void> closeFuture = new AbstractIoFuture<Void>() {
-
- /**
- * {@inheritDoc}
- */
- @Override
- protected boolean cancelOwner(final boolean mayInterruptIfRunning) {
- // we don't cancel close
- return false;
- }
- };
-
- /**
- * {@inheritDoc}
- */
- @Override
- public IoFuture<Void> close(final boolean immediately) {
- switch (state) {
- case CREATED:
- LOG.error("Session {} not opened", this);
- throw new IllegalStateException("cannot close an not opened session");
- case CONNECTED:
- state = SessionState.CLOSING;
- if (immediately) {
- channelClose();
- processSessionClosed();
- } else {
- // flush this session the flushing code will close the session
- flushWriteQueue();
- }
- break;
- case CLOSING:
- // return the same future
- LOG.warn("Already closing session {}", this);
- break;
- case CLOSED:
- LOG.warn("Already closed session {}", this);
- break;
- default:
- throw new IllegalStateException("not implemented session state : " + state);
- }
-
- return closeFuture;
- }
-
- /**
- * Close the inner socket channel
- */
- protected abstract void channelClose();
-
// ------------------------------------------------------------------------
// Event processing using the filter chain
// ------------------------------------------------------------------------
@@ -746,35 +559,6 @@ public abstract class AbstractIoSession implements IoSession, ReadFilterChainCon
}
/**
- * process session open event using the filter chain. To be called by the session {@link SelectorLoop} .
- */
- public void processSessionCreated() {
- LOG.debug("processing session open event");
-
- try {
-
- for (final IoFilter filter : chain) {
- filter.sessionOpened(this);
- }
-
- final IoHandler handler = getService().getIoHandler();
-
- if (handler != null) {
- IoHandlerExecutor executor = getService().getIoHandlerExecutor();
- if (executor != null) {
- // asynchronous event
- executor.execute(new OpenEvent(this));
- } else {
- // synchronous call (in the I/O loop)
- handler.sessionOpened(this);
- }
- }
- } catch (final RuntimeException e) {
- processException(e);
- }
- }
-
- /**
* process session closed event using the filter chain. To be called by the session {@link SelectorLoop} .
*/
public void processSessionClosed() {
@@ -1000,101 +784,4 @@ public abstract class AbstractIoSession implements IoSession, ReadFilterChainCon
readChainPosition--;
}
- /**
- * Process a write operation. This will be executed only because the session has something to write into the
- * channel.
- */
- public void processWrite(SelectorLoop selectorLoop) {
- try {
- LOG.debug("ready for write");
- LOG.debug("writable session : {}", this);
-
- do {
- // get a write request from the queue. We left it in the queue,
- // just in case we can't write all of the message content into
- // the channel : we will have to retrieve the message later
- final WriteRequest writeRequest = writeQueue.peek();
-
- if (writeRequest == null) {
- // Nothing to write : we are done
- break;
- }
-
- // The message is necessarily a ByteBuffer at this point
- final ByteBuffer buf = (ByteBuffer) writeRequest.getMessage();
-
- // Note that if the connection is secured, the buffer
- // already contains encrypted data.
-
- // Try to write the data, and get back the number of bytes
- // actually written
- final int written = ((SocketChannel) channel).write(buf);
- LOG.debug("wrote {} bytes to {}", written, this);
-
- if (written > 0) {
- incrementWrittenBytes(written);
- }
-
- // Update the idle status for this session
- idleChecker.sessionWritten(this, System.currentTimeMillis());
-
- // Ok, we may not have written everything. Check that.
- if (buf.remaining() == 0) {
- // completed write request, let's remove it (we use poll() instead
- // of remove(), because remove() may throw an exception if the
- // queue is empty.
- writeQueue.poll();
-
- // complete the future if we have one (we should...)
- final DefaultWriteFuture future = (DefaultWriteFuture) writeRequest.getFuture();
-
- if (future != null) {
- future.complete();
- }
-
- // generate the message sent event
- final Object highLevel = ((DefaultWriteRequest) writeRequest).getOriginalMessage();
-
- if (highLevel != null) {
- processMessageSent(highLevel);
- }
- } else {
- // output socket buffer is full, we need
- // to give up until next selection for
- // writing.
- break;
- }
- } while (!writeQueue.isEmpty());
-
- // We may have exited from the loop for some other reason
- // that an empty queue
- // if the session is no more interested in writing, we need
- // to stop listening for OP_WRITE events
- //
- // IMPORTANT : this section is synchronized so that the OP_WRITE flag
- // can be set safely by both the selector thread and the writer thread.
- synchronized (writeQueue) {
- if (writeQueue.isEmpty()) {
- if (isClosing()) {
- LOG.debug("closing session {} have empty write queue, so we close it", this);
- // we was flushing writes, now we to the close
- channelClose();
- } else {
- // no more write event needed
- selectorLoop.modifyRegistration(false, !isReadSuspended(), false, (SelectorListener) this,
- channel, false);
-
- // Reset the flag in IoSession too
- setNotRegisteredForWrite();
- }
- } else {
- // We have some more data to write : the channel OP_WRITE interest remains
- // as it was.
- }
- }
- } catch (final IOException e) {
- LOG.error("Exception while writing : ", e);
- processException(e);
- }
- }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/session/SslHelper.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/session/SslHelper.java b/core/src/main/java/org/apache/mina/session/SslHelper.java
deleted file mode 100644
index 26a7f63..0000000
--- a/core/src/main/java/org/apache/mina/session/SslHelper.java
+++ /dev/null
@@ -1,422 +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.session;
-
-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.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
- */
- /* no qualifier */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/ConnectFuture.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/ConnectFuture.java b/core/src/main/java/org/apache/mina/transport/ConnectFuture.java
new file mode 100644
index 0000000..2f707e5
--- /dev/null
+++ b/core/src/main/java/org/apache/mina/transport/ConnectFuture.java
@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+import java.util.concurrent.Future;
+
+import org.apache.mina.api.IoSession;
+import org.apache.mina.util.AbstractIoFuture;
+
+/**
+ * {@link Future} for waiting session connection.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ */
+public class ConnectFuture extends AbstractIoFuture<IoSession> {
+
+ @Override
+ protected boolean cancelOwner(boolean mayInterruptIfRunning) {
+ return false;
+ }
+
+ /**
+ * session connected
+ */
+ public void complete(IoSession session) {
+ setResult(session);
+ }
+
+ /**
+ * connection error
+ */
+ public void error(Exception e) {
+ setException(e);
+ }
+}
\ 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/bio/BioUdpServer.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/bio/BioUdpServer.java b/core/src/main/java/org/apache/mina/transport/bio/BioUdpServer.java
new file mode 100644
index 0000000..448327f
--- /dev/null
+++ b/core/src/main/java/org/apache/mina/transport/bio/BioUdpServer.java
@@ -0,0 +1,164 @@
+/*
+ * 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.bio;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.mina.service.executor.IoHandlerExecutor;
+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;
+
+/**
+ * A java blocking I/O based UDP server.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ *
+ */
+public class BioUdpServer extends AbstractUdpServer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BioUdpServer.class);
+
+ private SocketAddress boundAddress = null;
+
+ // the inner server socket/channel
+ DatagramChannel channel;
+
+ // thread in charge of reading the server socket
+ private Worker worker;
+ private boolean bound = false;
+
+ // for detecting idle sessions
+ private IdleChecker idleChecker = new IndexedIdleChecker();
+
+ // list of all the sessions by remote socket address
+ private final Map<SocketAddress /* remote socket address */, BioUdpSession> sessions = new ConcurrentHashMap<SocketAddress, BioUdpSession>();
+
+ public BioUdpServer(IoHandlerExecutor executor) {
+ super(executor);
+ }
+
+ public BioUdpServer(UdpSessionConfig config, IoHandlerExecutor executor) {
+ super(config, executor);
+ }
+
+ @Override
+ public SocketAddress getBoundAddress() {
+ return boundAddress;
+ }
+
+ @Override
+ public void bind(SocketAddress localAddress) throws IOException {
+ LOG.info("binding to address {}", localAddress);
+ if (bound) {
+ throw new IllegalStateException("already bound");
+ }
+ boundAddress = localAddress;
+ channel = DatagramChannel.open();
+ channel.socket().bind(localAddress);
+ Boolean reuseAddress = config.isReuseAddress();
+
+ if (reuseAddress != null) {
+ channel.socket().setReuseAddress(reuseAddress);
+ }
+
+ Integer readBufferSize = config.getReadBufferSize();
+
+ if (readBufferSize != null) {
+ channel.socket().setReceiveBufferSize(readBufferSize);
+ }
+
+ Integer sendBufferSize = config.getSendBufferSize();
+
+ if (sendBufferSize != null) {
+ channel.socket().setSendBufferSize(sendBufferSize);
+ }
+
+ Integer trafficClass = config.getTrafficClass();
+
+ if (trafficClass != null) {
+ channel.socket().setTrafficClass(trafficClass);
+ }
+
+ worker = new Worker();
+ bound = true;
+ worker.start();
+ }
+
+ @Override
+ public void bind(int port) throws IOException {
+ bind(new InetSocketAddress(5683));
+ }
+
+ @Override
+ public void unbind() throws IOException {
+ if (!bound) {
+ throw new IllegalStateException("not bound");
+ }
+ bound = false;
+ try {
+ worker.join();
+ channel.close();
+ boundAddress = null;
+ } catch (InterruptedException e) {
+ LOG.error("exception", e);
+ }
+
+ }
+
+ private class Worker extends Thread {
+
+ @Override
+ public void run() {
+ /** 64k receive buffer */
+ ByteBuffer rcvdBuffer = ByteBuffer.allocate(64 * 1024);
+ while (bound) {
+ // I/O !
+ rcvdBuffer.clear();
+ try {
+ SocketAddress from = channel.receive(rcvdBuffer);
+ BioUdpSession session = sessions.get(from);
+ if (session == null) {
+ // create the session
+ session = new BioUdpSession(from, BioUdpServer.this, idleChecker);
+ sessions.put(from, session);
+
+ // fire open
+ session.processSessionOpen();
+ }
+ rcvdBuffer.flip();
+ session.processMessageReceived(rcvdBuffer);
+
+ } catch (IOException e) {
+ LOG.error("Exception while reading", e);
+ }
+ }
+ }
+ }
+}
\ 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/bio/BioUdpSession.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/bio/BioUdpSession.java b/core/src/main/java/org/apache/mina/transport/bio/BioUdpSession.java
new file mode 100644
index 0000000..3d01539
--- /dev/null
+++ b/core/src/main/java/org/apache/mina/transport/bio/BioUdpSession.java
@@ -0,0 +1,158 @@
+/*
+ * 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.bio;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+
+import org.apache.mina.api.IoFuture;
+import org.apache.mina.api.IoSession;
+import org.apache.mina.api.IoSessionConfig;
+import org.apache.mina.service.idlechecker.IdleChecker;
+import org.apache.mina.session.AbstractIoSession;
+import org.apache.mina.session.DefaultWriteRequest;
+import org.apache.mina.session.WriteRequest;
+
+/**
+ * A {@link IoSession} for {@link BioUdpServer}
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ */
+public class BioUdpSession extends AbstractIoSession {
+
+ private SocketAddress remoteAddress;
+
+ public BioUdpSession(SocketAddress remoteAddress, BioUdpServer service, IdleChecker idleChecker) {
+ super(service, idleChecker);
+ this.remoteAddress = remoteAddress;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public SocketAddress getRemoteAddress() {
+ return remoteAddress;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public SocketAddress getLocalAddress() {
+ return ((BioUdpServer) getService()).getBoundAddress();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public IoFuture<Void> close(boolean immediately) {
+ processSessionClosed();
+ return null;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void suspendRead() {
+ throw new IllegalStateException("not supported");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void suspendWrite() {
+ throw new IllegalStateException("not supported");
+
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void resumeRead() {
+ throw new IllegalStateException("not supported");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void resumeWrite() {
+ throw new IllegalStateException("not supported");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isReadSuspended() {
+ return false;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isWriteSuspended() {
+ return false;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public IoSessionConfig getConfig() {
+ return config;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public WriteRequest enqueueWriteRequest(WriteRequest writeRequest) {
+
+ try {
+ int written = ((BioUdpServer) getService()).channel.send((ByteBuffer) writeRequest.getMessage(),
+ remoteAddress);
+ if (written > 0) {
+ incrementWrittenBytes(written);
+
+ final Object highLevel = ((DefaultWriteRequest) writeRequest).getOriginalMessage();
+
+ if (highLevel != null) {
+ processMessageSent(highLevel);
+ }
+
+ } else {
+ processException(new RuntimeException("unable to write"));
+ }
+ } catch (IOException e) {
+ processException(e);
+ }
+
+ return writeRequest;
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/transport/nio/AbstractNioSession.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/nio/AbstractNioSession.java b/core/src/main/java/org/apache/mina/transport/nio/AbstractNioSession.java
new file mode 100644
index 0000000..85cf170
--- /dev/null
+++ b/core/src/main/java/org/apache/mina/transport/nio/AbstractNioSession.java
@@ -0,0 +1,331 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.nio.channels.SelectableChannel;
+import java.nio.channels.SocketChannel;
+import java.util.Queue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.mina.api.IoFuture;
+import org.apache.mina.api.IoService;
+import org.apache.mina.api.IoSession;
+import org.apache.mina.service.idlechecker.IdleChecker;
+import org.apache.mina.session.AbstractIoSession;
+import org.apache.mina.session.DefaultWriteFuture;
+import org.apache.mina.session.DefaultWriteQueue;
+import org.apache.mina.session.DefaultWriteRequest;
+import org.apache.mina.session.WriteRequest;
+import org.apache.mina.util.AbstractIoFuture;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Common ancestor for NIO based {@link IoSession} implmentation.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ */
+public abstract class AbstractNioSession extends AbstractIoSession {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractNioSession.class);
+
+ /** the NIO channel for this session */
+ protected final SelectableChannel channel;
+
+ /** is this session registered for being polled for write ready events */
+ private final AtomicBoolean registeredForWrite = new AtomicBoolean();
+
+ /** the queue of pending writes for the session, to be dequeued by the {@link SelectorLoop} */
+ private final Queue<WriteRequest> writeQueue = new DefaultWriteQueue();
+
+ public AbstractNioSession(IoService service, SelectableChannel channel, IdleChecker idleChecker) {
+ super(service, idleChecker);
+ this.channel = channel;
+ }
+
+ /**
+ * Writes the message immediately. If we can't write all the message, we will get back the number of written bytes.
+ *
+ * @param message the message to write
+ * @return the number of written bytes
+ */
+ protected abstract int writeDirect(Object message);
+
+ /**
+ * Copy the HeapBuffer into a DirectBuffer, if needed.
+ *
+ * @param writeRequest The request containing the HeapBuffer
+ * @param createNew A flag to force the creation of a DirectBuffer
+ * @return A DirectBuffer
+ */
+ protected abstract ByteBuffer convertToDirectBuffer(WriteRequest writeRequest, boolean createNew);
+
+ // ------------------------------------------------------------------------
+ // Close session management
+ // ------------------------------------------------------------------------
+
+ /** we pre-allocate a close future for lock-less {@link #close(boolean)} */
+ private final IoFuture<Void> closeFuture = new AbstractIoFuture<Void>() {
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected boolean cancelOwner(final boolean mayInterruptIfRunning) {
+ // we don't cancel close
+ return false;
+ }
+ };
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public IoFuture<Void> close(final boolean immediately) {
+ switch (state) {
+ case CREATED:
+ LOG.error("Session {} not opened", this);
+ throw new IllegalStateException("cannot close an not opened session");
+ case CONNECTED:
+ state = SessionState.CLOSING;
+ if (immediately) {
+ channelClose();
+ processSessionClosed();
+ } else {
+ // flush this session the flushing code will close the session
+ flushWriteQueue();
+ }
+ break;
+ case CLOSING:
+ // return the same future
+ LOG.warn("Already closing session {}", this);
+ break;
+ case CLOSED:
+ LOG.warn("Already closed session {}", this);
+ break;
+ default:
+ throw new IllegalStateException("not implemented session state : " + state);
+ }
+
+ return closeFuture;
+ }
+
+ /**
+ * Close the inner socket channel
+ */
+ protected abstract void channelClose();
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public WriteRequest enqueueWriteRequest(WriteRequest writeRequest) {
+ LOG.debug("enqueueWriteRequest {}", writeRequest);
+ if (isConnectedSecured()) {
+ // SSL/TLS : we have to encrypt the message
+ final SslHelper sslHelper = getAttribute(SSL_HELPER, null);
+
+ if (sslHelper == null) {
+ throw new IllegalStateException();
+ }
+
+ writeRequest = sslHelper.processWrite(this, writeRequest.getMessage(), writeQueue);
+ }
+
+ synchronized (writeQueue) {
+ ByteBuffer message = (ByteBuffer) writeRequest.getMessage();
+
+ if (writeQueue.isEmpty()) {
+ // Transfer the buffer in a DirectByteBuffer if it's a HeapByteBuffer and if it's too big
+ message = convertToDirectBuffer(writeRequest, false);
+
+ // We don't have anything in the writeQueue, let's try to write the
+ // data in the channel immediately if we can
+ int written = writeDirect(writeRequest.getMessage());
+
+ LOG.debug("wrote {} bytes to {}", written, this);
+
+ if (written > 0) {
+ incrementWrittenBytes(written);
+ }
+
+ // Update the idle status for this session
+ idleChecker.sessionWritten(this, System.currentTimeMillis());
+ int remaining = message.remaining();
+
+ if ((written < 0) || (remaining > 0)) {
+ // Create a DirectBuffer unconditionally
+ convertToDirectBuffer(writeRequest, true);
+
+ // We have to push the request on the writeQueue
+ writeQueue.add(writeRequest);
+
+ // If it wasn't, we register this session as interested to write.
+ // It's done in atomic fashion for avoiding two concurrent registering.
+ if (!registeredForWrite.getAndSet(true)) {
+ flushWriteQueue();
+ }
+ } else {
+ // The message has been fully written : update the stats, and signal the handler
+ // generate the message sent event
+ // complete the future if we have one (we should...)
+ final DefaultWriteFuture future = (DefaultWriteFuture) writeRequest.getFuture();
+
+ if (future != null) {
+ future.complete();
+ }
+
+ final Object highLevel = ((DefaultWriteRequest) writeRequest).getOriginalMessage();
+
+ if (highLevel != null) {
+ processMessageSent(highLevel);
+ }
+ }
+ } else {
+ // Transfer the buffer in a DirectByteBuffer if it's a HeapByteBuffer
+ message = convertToDirectBuffer(writeRequest, true);
+
+ // We have to push the request on the writeQueue
+ writeQueue.add(writeRequest);
+ }
+ }
+
+ return writeRequest;
+ }
+
+ public abstract void flushWriteQueue();
+
+ public void setNotRegisteredForWrite() {
+ registeredForWrite.set(false);
+ }
+
+ protected boolean isRegisteredForWrite() {
+ return registeredForWrite.get();
+ }
+
+ /**
+ * Get the {@link Queue} of this session. The write queue contains the pending writes.
+ *
+ * @return the write queue of this session
+ */
+ public Queue<WriteRequest> getWriteQueue() {
+ return writeQueue;
+ }
+
+ /**
+ * Process a write operation. This will be executed only because the session has something to write into the
+ * channel.
+ */
+ public void processWrite(SelectorLoop selectorLoop) {
+ try {
+ LOG.debug("ready for write");
+ LOG.debug("writable session : {}", this);
+
+ do {
+ // get a write request from the queue. We left it in the queue,
+ // just in case we can't write all of the message content into
+ // the channel : we will have to retrieve the message later
+ final WriteRequest writeRequest = writeQueue.peek();
+
+ if (writeRequest == null) {
+ // Nothing to write : we are done
+ break;
+ }
+
+ // The message is necessarily a ByteBuffer at this point
+ final ByteBuffer buf = (ByteBuffer) writeRequest.getMessage();
+
+ // Note that if the connection is secured, the buffer
+ // already contains encrypted data.
+
+ // Try to write the data, and get back the number of bytes
+ // actually written
+ final int written = ((SocketChannel) channel).write(buf);
+ LOG.debug("wrote {} bytes to {}", written, this);
+
+ if (written > 0) {
+ incrementWrittenBytes(written);
+ }
+
+ // Update the idle status for this session
+ idleChecker.sessionWritten(this, System.currentTimeMillis());
+
+ // Ok, we may not have written everything. Check that.
+ if (buf.remaining() == 0) {
+ // completed write request, let's remove it (we use poll() instead
+ // of remove(), because remove() may throw an exception if the
+ // queue is empty.
+ writeQueue.poll();
+
+ // complete the future if we have one (we should...)
+ final DefaultWriteFuture future = (DefaultWriteFuture) writeRequest.getFuture();
+
+ if (future != null) {
+ future.complete();
+ }
+
+ // generate the message sent event
+ final Object highLevel = ((DefaultWriteRequest) writeRequest).getOriginalMessage();
+
+ if (highLevel != null) {
+ processMessageSent(highLevel);
+ }
+ } else {
+ // output socket buffer is full, we need
+ // to give up until next selection for
+ // writing.
+ break;
+ }
+ } while (!writeQueue.isEmpty());
+
+ // We may have exited from the loop for some other reason
+ // that an empty queue
+ // if the session is no more interested in writing, we need
+ // to stop listening for OP_WRITE events
+ //
+ // IMPORTANT : this section is synchronized so that the OP_WRITE flag
+ // can be set safely by both the selector thread and the writer thread.
+ synchronized (writeQueue) {
+ if (writeQueue.isEmpty()) {
+ if (isClosing()) {
+ LOG.debug("closing session {} have empty write queue, so we close it", this);
+ // we was flushing writes, now we to the close
+ channelClose();
+ } else {
+ // no more write event needed
+ selectorLoop.modifyRegistration(false, !isReadSuspended(), false, (SelectorListener) this,
+ channel, false);
+
+ // Reset the flag in IoSession too
+ setNotRegisteredForWrite();
+ }
+ } else {
+ // We have some more data to write : the channel OP_WRITE interest remains
+ // as it was.
+ }
+ }
+ } catch (final IOException e) {
+ LOG.error("Exception while writing : ", e);
+ processException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/mina/blob/8ee4890b/core/src/main/java/org/apache/mina/transport/nio/ConnectFuture.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/nio/ConnectFuture.java b/core/src/main/java/org/apache/mina/transport/nio/ConnectFuture.java
deleted file mode 100644
index 8c33702..0000000
--- a/core/src/main/java/org/apache/mina/transport/nio/ConnectFuture.java
+++ /dev/null
@@ -1,26 +0,0 @@
-package org.apache.mina.transport.nio;
-
-import org.apache.mina.api.IoSession;
-import org.apache.mina.util.AbstractIoFuture;
-
-public class ConnectFuture extends AbstractIoFuture<IoSession> {
-
- @Override
- protected boolean cancelOwner(boolean mayInterruptIfRunning) {
- return false;
- }
-
- /**
- * session connected
- */
- public void complete(IoSession session) {
- setResult(session);
- }
-
- /**
- * connection error
- */
- public void error(Exception e) {
- setException(e);
- }
-}
\ 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/NioTcpClient.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mina/transport/nio/NioTcpClient.java b/core/src/main/java/org/apache/mina/transport/nio/NioTcpClient.java
new file mode 100644
index 0000000..802bae9
--- /dev/null
+++ b/core/src/main/java/org/apache/mina/transport/nio/NioTcpClient.java
@@ -0,0 +1,238 @@
+/*
+ * 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.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.ConnectFuture;
+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