You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mina.apache.org by el...@apache.org on 2013/03/22 14:49:20 UTC
[2/5] git commit: o Removed some final keyword o Implemented the
first part of the NioUdpServer : handling new session
o Removed some final keyword
o Implemented the first part of the NioUdpServer : handling new session
Project: http://git-wip-us.apache.org/repos/asf/mina/repo
Commit: http://git-wip-us.apache.org/repos/asf/mina/commit/6426f182
Tree: http://git-wip-us.apache.org/repos/asf/mina/tree/6426f182
Diff: http://git-wip-us.apache.org/repos/asf/mina/diff/6426f182
Branch: refs/heads/trunk
Commit: 6426f182d8092d925864f8a76e7d170f82c66e46
Parents: 02f02e7
Author: Emmanuel Lécharny <el...@apache.org>
Authored: Fri Mar 22 14:04:48 2013 +0100
Committer: Emmanuel Lécharny <el...@apache.org>
Committed: Fri Mar 22 14:04:48 2013 +0100
----------------------------------------------------------------------
.../apache/mina/transport/nio/NioTcpServer.java | 24 ++--
.../apache/mina/transport/nio/NioUdpServer.java | 122 ++++++++-------
2 files changed, 80 insertions(+), 66 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mina/blob/6426f182/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
index 97601dd..46e7931 100644
--- a/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java
+++ b/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java
@@ -260,11 +260,11 @@ public class NioTcpServer extends AbstractTcpServer implements SelectorListener
}
}
- private void createSession(final SocketChannel clientSocket) throws IOException {
+ private void createSession(SocketChannel clientSocket) throws IOException {
LOG.debug("create session");
- final SocketChannel socketChannel = clientSocket;
- final TcpSessionConfig config = getSessionConfig();
- final SelectorLoop readWriteSelectorLoop = readWriteSelectorPool.getSelectorLoop();
+ SocketChannel socketChannel = clientSocket;
+ TcpSessionConfig config = getSessionConfig();
+ SelectorLoop readWriteSelectorLoop = readWriteSelectorPool.getSelectorLoop();
final NioTcpSession session = new NioTcpSession(this, socketChannel, readWriteSelectorLoop, idleChecker);
socketChannel.configureBlocking(false);
@@ -275,49 +275,49 @@ public class NioTcpServer extends AbstractTcpServer implements SelectorListener
config.getIdleTimeInMillis(IdleStatus.WRITE_IDLE));
// apply the default service socket configuration
- final Boolean keepAlive = config.isKeepAlive();
+ Boolean keepAlive = config.isKeepAlive();
if (keepAlive != null) {
session.getConfig().setKeepAlive(keepAlive);
}
- final Boolean oobInline = config.isOobInline();
+ Boolean oobInline = config.isOobInline();
if (oobInline != null) {
session.getConfig().setOobInline(oobInline);
}
- final Boolean reuseAddress = config.isReuseAddress();
+ Boolean reuseAddress = config.isReuseAddress();
if (reuseAddress != null) {
session.getConfig().setReuseAddress(reuseAddress);
}
- final Boolean tcpNoDelay = config.isTcpNoDelay();
+ Boolean tcpNoDelay = config.isTcpNoDelay();
if (tcpNoDelay != null) {
session.getConfig().setTcpNoDelay(tcpNoDelay);
}
- final Integer receiveBufferSize = config.getReadBufferSize();
+ Integer receiveBufferSize = config.getReadBufferSize();
if (receiveBufferSize != null) {
session.getConfig().setReadBufferSize(receiveBufferSize);
}
- final Integer sendBufferSize = config.getSendBufferSize();
+ Integer sendBufferSize = config.getSendBufferSize();
if (sendBufferSize != null) {
session.getConfig().setSendBufferSize(sendBufferSize);
}
- final Integer trafficClass = config.getTrafficClass();
+ Integer trafficClass = config.getTrafficClass();
if (trafficClass != null) {
session.getConfig().setTrafficClass(trafficClass);
}
- final Integer soLinger = config.getSoLinger();
+ Integer soLinger = config.getSoLinger();
if (soLinger != null) {
session.getConfig().setSoLinger(soLinger);
http://git-wip-us.apache.org/repos/asf/mina/blob/6426f182/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
index 2366e1a..9390bf5 100644
--- a/core/src/main/java/org/apache/mina/transport/nio/NioUdpServer.java
+++ b/core/src/main/java/org/apache/mina/transport/nio/NioUdpServer.java
@@ -28,6 +28,7 @@ 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;
@@ -64,19 +65,15 @@ public class NioUdpServer extends AbstractUdpServer implements SelectorListener
// 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 accept incoming connection */
- private final SelectorLoop acceptSelectorLoop;
-
- /** The read/write selectorPool */
- private final SelectorLoopPool readWriteSelectorPool;
+ /** 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), new FixedSelectorLoopPool("Server", Runtime.getRuntime()
- .availableProcessors() + 1), null);
+ this(new NioSelectorLoop("accept", 0), null);
}
/**
@@ -86,37 +83,7 @@ public class NioUdpServer extends AbstractUdpServer implements SelectorListener
* @param sessionConfig The configuration to use for this server
*/
public NioUdpServer(UdpSessionConfig config) {
- this(config, new NioSelectorLoop("accept", 0), new FixedSelectorLoopPool("Server", Runtime.getRuntime()
- .availableProcessors() + 1), null);
- }
-
- /**
- * Create a UDP 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 NioUdpServer(SelectorLoopPool selectorLoopPool, IoHandlerExecutor handlerExecutor) {
- this(selectorLoopPool.getSelectorLoop(), selectorLoopPool, handlerExecutor);
- }
-
- /**
- * Create a UDP 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 sessionConfig The configuration to use for this server
- * @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 NioUdpServer(UdpSessionConfig config, SelectorLoopPool selectorLoopPool, IoHandlerExecutor handlerExecutor) {
- this(config, selectorLoopPool.getSelectorLoop(), selectorLoopPool, handlerExecutor);
+ this(config, new NioSelectorLoop("accept", 0), null);
}
/**
@@ -128,11 +95,9 @@ public class NioUdpServer extends AbstractUdpServer implements SelectorListener
* 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 acceptSelectorLoop, SelectorLoopPool readWriteSelectorLoop,
- IoHandlerExecutor handlerExecutor) {
+ public NioUdpServer(SelectorLoop readSelectorLoop, IoHandlerExecutor handlerExecutor) {
super(handlerExecutor);
- this.acceptSelectorLoop = acceptSelectorLoop;
- this.readWriteSelectorPool = readWriteSelectorLoop;
+ this.readSelectorLoop = readSelectorLoop;
}
/**
@@ -140,16 +105,13 @@ public class NioUdpServer extends AbstractUdpServer implements SelectorListener
*
* @param sessionConfig The configuration to use for this server
* @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(UdpSessionConfig config, SelectorLoop acceptSelectorLoop,
- SelectorLoopPool readWriteSelectorLoop, IoHandlerExecutor handlerExecutor) {
+ public NioUdpServer(UdpSessionConfig config, SelectorLoop readSelectorLoop, IoHandlerExecutor handlerExecutor) {
super(config, handlerExecutor);
- this.acceptSelectorLoop = acceptSelectorLoop;
- this.readWriteSelectorPool = readWriteSelectorLoop;
+ this.readSelectorLoop = readSelectorLoop;
}
/**
@@ -201,7 +163,7 @@ public class NioUdpServer extends AbstractUdpServer implements SelectorListener
datagramChannel.socket().bind(address);
datagramChannel.configureBlocking(false);
- acceptSelectorLoop.register(false, false, true, false, this, datagramChannel, null);
+ readSelectorLoop.register(false, false, true, false, this, datagramChannel, null);
// it's the first address bound, let's fire the event
this.fireServiceActivated();
@@ -217,7 +179,7 @@ public class NioUdpServer extends AbstractUdpServer implements SelectorListener
throw new IllegalStateException("server not bound");
}
- acceptSelectorLoop.unregister(this, datagramChannel);
+ readSelectorLoop.unregister(this, datagramChannel);
datagramChannel.socket().close();
datagramChannel.close();
@@ -256,14 +218,17 @@ public class NioUdpServer extends AbstractUdpServer implements SelectorListener
LOG.debug("read {} bytes form {}", readBuffer.remaining(), source);
// let's find the corresponding session
+ if (source != null) {
+ NioUdpSession session = sessions.get(source);
- NioUdpSession session = sessions.get(source);
+ if (session == null) {
+ //session = new NioUdpSession(this, idleChecker, address, source);
- if (session == null) {
- session = new NioUdpSession(this, idleChecker, address, source);
- }
+ session = createSession(source, datagramChannel);
+ }
- session.receivedDatagram(readBuffer);
+ session.receivedDatagram(readBuffer);
+ }
} catch (final IOException ex) {
LOG.error("IOException while reading the socket", ex);
}
@@ -273,4 +238,53 @@ public class NioUdpServer extends AbstractUdpServer implements SelectorListener
}
}
+ private NioUdpSession createSession(SocketAddress remoteAddress, DatagramChannel datagramChannel)
+ throws IOException {
+ LOG.debug("create session");
+ UdpSessionConfig config = getSessionConfig();
+ final NioUdpSession session = new NioUdpSession(this, idleChecker, datagramChannel,
+ datagramChannel.getLocalAddress(), 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