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 2012/05/25 13:58:21 UTC
svn commit: r1342586 - in /mina/trunk:
core/src/main/java/org/apache/mina/session/
core/src/main/java/org/apache/mina/transport/nio/
core/src/main/java/org/apache/mina/transport/tcp/
core/src/main/java/org/apache/mina/transport/tcp/nio/ core/src/main/j...
Author: jvermillard
Date: Fri May 25 11:58:20 2012
New Revision: 1342586
URL: http://svn.apache.org/viewvc?rev=1342586&view=rev
Log:
* flattened the nio hierarchy for sharing bits between UDP and TCP NIO implementation
* moved some logic to the servers & session
* read datagram events
Added:
mina/trunk/core/src/main/java/org/apache/mina/transport/nio/
mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioSelectorProcessor.java (contents, props changed)
- copied, changed from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/NioSelectorProcessor.java
mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpClient.java
- copied, changed from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/nio/NioTcpClient.java
mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java (contents, props changed)
- copied, changed from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/nio/NioTcpServer.java
mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpSession.java (contents, props changed)
- copied, changed from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/NioTcpSession.java
mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpClient.java
- copied, changed from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/udp/nio/NioUdpClient.java
mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpServer.java
- copied, changed from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/udp/nio/NioUdpServer.java
mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpSession.java (with props)
mina/trunk/core/src/main/java/org/apache/mina/transport/nio/SelectorEventListener.java (with props)
Removed:
mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/NioSelectorProcessor.java
mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/NioTcpSession.java
mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/nio/NioTcpClient.java
mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/nio/NioTcpServer.java
mina/trunk/core/src/main/java/org/apache/mina/transport/udp/nio/NioUdpClient.java
mina/trunk/core/src/main/java/org/apache/mina/transport/udp/nio/NioUdpServer.java
Modified:
mina/trunk/core/src/main/java/org/apache/mina/session/AbstractIoSession.java
mina/trunk/core/src/test/java/org/apache/mina/service/idlecheker/IndexedIdleChekerTest.java
mina/trunk/core/src/test/java/org/apache/mina/session/AbstractIoSessionTest.java
mina/trunk/examples/src/main/java/org/apache/mina/examples/echoserver/NioEchoServer.java
mina/trunk/examples/src/main/java/org/apache/mina/examples/http/HttpTest.java
mina/trunk/examples/src/main/java/org/apache/mina/examples/http/HttpsTest.java
mina/trunk/examples/src/main/java/org/apache/mina/examples/ldap/LdapTest.java
mina/trunk/examples/src/main/java/org/apache/mina/examples/udpecho/NioUdpEchoServer.java
Modified: mina/trunk/core/src/main/java/org/apache/mina/session/AbstractIoSession.java
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/main/java/org/apache/mina/session/AbstractIoSession.java?rev=1342586&r1=1342585&r2=1342586&view=diff
==============================================================================
--- mina/trunk/core/src/main/java/org/apache/mina/session/AbstractIoSession.java (original)
+++ mina/trunk/core/src/main/java/org/apache/mina/session/AbstractIoSession.java Fri May 25 11:58:20 2012
@@ -40,6 +40,7 @@ import org.apache.mina.api.IoSession;
import org.apache.mina.filterchain.ReadFilterChainController;
import org.apache.mina.filterchain.WriteFilterChainController;
import org.apache.mina.service.SelectorProcessor;
+import org.apache.mina.service.idlechecker.IdleChecker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,6 +71,9 @@ public abstract class AbstractIoSession
/** The {@link SelectorProcessor} used for handling this session writing */
protected SelectorProcessor writeProcessor;
+ /** the {@link IdleChecker} in charge of detecting idle event for this session */
+ protected final IdleChecker idleChecker;
+
//------------------------------------------------------------------------
// Basic statistics
//------------------------------------------------------------------------
@@ -106,14 +110,14 @@ public abstract class AbstractIoSession
protected volatile boolean secured;
/** is this session registered for being polled for write ready events */
- private AtomicBoolean registeredForWrite = new AtomicBoolean();
+ private final AtomicBoolean registeredForWrite = new AtomicBoolean();
//------------------------------------------------------------------------
// Write queue
//------------------------------------------------------------------------
/** the queue of pending writes for the session, to be dequeued by the {@link SelectorProcessor} */
- private Queue<WriteRequest> writeQueue = new DefaultWriteQueue();
+ private final Queue<WriteRequest> writeQueue = new DefaultWriteQueue();
/** A lock to protect the access to the write queue */
private final ReadWriteLock writeQueueLock = new ReentrantReadWriteLock();
@@ -147,13 +151,14 @@ public abstract class AbstractIoSession
* @param service the service this session is associated with
* @param writeProcessor the processor in charge of processing this session write queue
*/
- public AbstractIoSession(IoService service, SelectorProcessor writeProcessor) {
+ public AbstractIoSession(IoService service, SelectorProcessor writeProcessor, IdleChecker idleChecker) {
// generated a unique id
id = NEXT_ID.getAndIncrement();
creationTime = System.currentTimeMillis();
this.service = service;
this.writeProcessor = writeProcessor;
this.chain = service.getFilters();
+ this.idleChecker = idleChecker;
LOG.debug("Created new session with id : {}", id);
@@ -250,6 +255,7 @@ public abstract class AbstractIoSession
/**
* {@inheritDoc}
*/
+ @Override
public void changeState(SessionState to) throws IllegalStateException {
try {
stateWriteLock.lock();
@@ -346,6 +352,7 @@ public abstract class AbstractIoSession
/**
* {@inheritDoc}
*/
+ @Override
public void initSecure(SSLContext sslContext) throws SSLException {
SslHelper sslHelper = new SslHelper(this, sslContext);
sslHelper.init();
@@ -524,6 +531,7 @@ public abstract class AbstractIoSession
/**
* {@inheritDoc}
*/
+ @Override
public WriteRequest enqueueWriteRequest(Object message) {
WriteRequest request = null;
@@ -681,6 +689,7 @@ public abstract class AbstractIoSession
* process session message received event using the filter chain. To be called by the session {@link SelectorProcessor} .
* @param message the received message
*/
+ @Override
public void callWriteNextFilter(Object message) {
LOG.debug("calling next filter for writing for message '{}' position : {}", message, writeChainPosition);
Copied: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioSelectorProcessor.java (from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/NioSelectorProcessor.java)
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioSelectorProcessor.java?p2=mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioSelectorProcessor.java&p1=mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/NioSelectorProcessor.java&r1=1341901&r2=1342586&rev=1342586&view=diff
==============================================================================
--- mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/NioSelectorProcessor.java (original)
+++ mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioSelectorProcessor.java Fri May 25 11:58:20 2012
@@ -18,22 +18,18 @@
*
*/
-package org.apache.mina.transport.tcp;
-
-import static org.apache.mina.api.IoSession.SSL_HELPER;
+package org.apache.mina.transport.nio;
import java.io.IOException;
-import java.net.SocketAddress;
import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
-import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Queue;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -50,11 +46,7 @@ import org.apache.mina.service.SelectorS
import org.apache.mina.service.idlechecker.IdleChecker;
import org.apache.mina.service.idlechecker.IndexedIdleChecker;
import org.apache.mina.session.AbstractIoSession;
-import org.apache.mina.session.DefaultWriteFuture;
-import org.apache.mina.session.SslHelper;
-import org.apache.mina.session.WriteRequest;
-import org.apache.mina.transport.tcp.nio.NioTcpServer;
-import org.apache.mina.transport.udp.nio.NioUdpServer;
+import org.apache.mina.transport.tcp.TcpSessionConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -64,620 +56,479 @@ import org.slf4j.LoggerFactory;
*
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
*/
-public class NioSelectorProcessor implements SelectorProcessor<NioTcpServer,NioUdpServer> {
-
- /** A logger for this class */
- private static final Logger LOGGER = LoggerFactory
- .getLogger(NioSelectorProcessor.class);
-
- /**
- * A timeout used for the select, as we need to get out to deal with idle
- * sessions
- */
- private static final long SELECT_TIMEOUT = 1000L;
-
- private SelectorStrategy<NioSelectorProcessor> strategy;
-
- private final Map<SocketAddress, ServerSocketChannel> serverSocketChannels = new ConcurrentHashMap<SocketAddress, ServerSocketChannel>();
-
- /** Read buffer for all the incoming bytes (default to 64Kb) */
- private final ByteBuffer readBuffer = ByteBuffer.allocate(64 * 1024);
-
- /** the thread polling and processing the I/O events */
- private SelectorWorker worker = null;
-
- /** helper for detecting idleing sessions */
- private final IdleChecker idleChecker = new IndexedIdleChecker();
-
-
- /** A queue containing the servers to bind to this selector */
- private final Queue<NioTcpServer> tcpServersToadd = new ConcurrentLinkedQueue<NioTcpServer>();
-
- /** A queue containing the servers to bind to this selector */
- private final Queue<NioUdpServer> udpServersToadd = new ConcurrentLinkedQueue<NioUdpServer>();
-
- /** server to remove of the selector */
- private final Queue<NioTcpServer> tcpServersToRemove = new ConcurrentLinkedQueue<NioTcpServer>();
-
- /** server to remove of the selector */
- private final Queue<NioUdpServer> udpServersToRemove = new ConcurrentLinkedQueue<NioUdpServer>();
-
- /**
- * new session freshly accepted, placed here for being added to the selector
- */
- private final Queue<NioTcpSession> sessionsToConnect = new ConcurrentLinkedQueue<NioTcpSession>();
-
- /** session to be removed of the selector */
- private final Queue<NioTcpSession> sessionsToClose = new ConcurrentLinkedQueue<NioTcpSession>();
-
- /** A queue used to store the sessions to be flushed */
- private final Queue<NioTcpSession> flushingSessions = new ConcurrentLinkedQueue<NioTcpSession>();
-
- private Selector selector;
-
- // Lock for Selector worker, using default. can look into fairness later.
- // We need to think about a lock less mechanism here.
- private final Lock workerLock = new ReentrantLock();
-
- /**
- * {@inheritDoc}
- */
- @SuppressWarnings("unchecked")
- @Override
- public void setStrategy(SelectorStrategy<?> strategy) {
- this.strategy = (SelectorStrategy<NioSelectorProcessor>)strategy;
- }
-
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void addServer(NioTcpServer server) {
- tcpServersToadd.add(server);
- this.wakeupWorker();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void addServer(NioUdpServer server) {
- udpServersToadd.add(server);
- this.wakeupWorker();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void removeServer(NioTcpServer server) {
- tcpServersToRemove.add(server);
- this.wakeupWorker();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void removeServer(NioUdpServer server) {
- udpServersToRemove.add(server);
- this.wakeupWorker();
- }
-
- /**
- * Wake the I/O worker thread and if none exists, create a new one FIXME :
- * too much locking there ?
- */
- private void wakeupWorker() {
- this.workerLock.lock();
- try {
- if (this.worker == null) {
- this.worker = new SelectorWorker();
- this.worker.start();
- }
- } finally {
- this.workerLock.unlock();
- }
-
- if (this.selector != null) {
- this.selector.wakeup();
- }
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void createSession(final IoService service, final Object clientSocket)
- throws SSLException {
- LOGGER.debug("create session");
- final SocketChannel socketChannel = (SocketChannel) clientSocket;
- final TcpSessionConfig config = (TcpSessionConfig) service
- .getSessionConfig();
- final NioTcpSession session = new NioTcpSession(service, socketChannel,
- this.strategy.getSelectorForNewSession(this));
-
- try {
- socketChannel.configureBlocking(false);
- } catch (IOException e) {
- LOGGER.error(
- "Unexpected exception, while configuring socket as non blocking",
- e);
- throw new RuntimeIoException(
- "cannot configure socket as non-blocking", e);
- }
- // 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.getReceiveBufferSize();
-
- if (receiveBufferSize != null) {
- session.getConfig().setReceiveBufferSize(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());
- }
-
- // event session created
- session.processSessionCreated();
-
- // add the session to the queue for being added to the selector
- this.sessionsToConnect.add(session);
- this.wakeupWorker();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void flush(final AbstractIoSession session) {
- LOGGER.debug("scheduling session {} for writing", session);
- // add the session to the list of session to be registered for writing
- this.flushingSessions.add((NioTcpSession) session);
- // wake the selector for unlocking the I/O thread
- this.wakeupWorker();
- }
-
- /**
- * The worker processing incoming session creation, session destruction
- * requests, session write and reads. It will also bind new servers.
- */
- private class SelectorWorker extends Thread {
- // map for finding the keys associated with a given server
- private final Map<ServerSocketChannel, SelectionKey> serverKey = new HashMap<ServerSocketChannel, SelectionKey>();
-
- // map for finding read keys associated with a given session
- private final Map<NioTcpSession, SelectionKey> sessionReadKey = new HashMap<NioTcpSession, SelectionKey>();
-
- @Override
- public void run() {
- try {
- if (NioSelectorProcessor.this.selector == null) {
- LOGGER.debug("opening a new selector");
-
- try {
- NioSelectorProcessor.this.selector = Selector.open();
- } catch (IOException e) {
- LOGGER.error(
- "IOException while opening a new Selector", e);
- }
- }
-
- for (;;) {
- try {
- // pop server sockets for removing
- if (NioSelectorProcessor.this.tcpServersToRemove.size() > 0 || NioSelectorProcessor.this.udpServersToRemove.size() > 0) {
- this.processServerRemove();
- }
-
- // pop new server sockets for accepting
- if (NioSelectorProcessor.this.tcpServersToadd.size() > 0 || NioSelectorProcessor.this.udpServersToadd.size() > 0) {
- this.processServerAdd();
- }
-
- // pop new session for starting read/write
- if (NioSelectorProcessor.this.sessionsToConnect.size() > 0) {
- this.processConnectSessions();
- }
-
- // pop session for close, if any
- if (NioSelectorProcessor.this.sessionsToClose.size() > 0) {
- this.processCloseSessions();
- }
-
- LOGGER.debug("selecting...");
- int readyCount = NioSelectorProcessor.this.selector
- .select(SELECT_TIMEOUT);
- LOGGER.debug("... done selecting : {}", readyCount);
-
- if (readyCount > 0) {
- // process selected keys
- Iterator<SelectionKey> selectedKeys = NioSelectorProcessor.this.selector
- .selectedKeys().iterator();
-
- // Loop on each SelectionKey and process any valid
- // action
- while (selectedKeys.hasNext()) {
- SelectionKey key = selectedKeys.next();
- selectedKeys.remove();
-
- if (!key.isValid()) {
- continue;
- }
-
- NioSelectorProcessor.this.selector
- .selectedKeys().remove(key);
-
- if (key.isAcceptable()) {
- this.processAccept(key);
- }
-
- if (key.isReadable()) {
- this.processRead(key);
- }
-
- if (key.isWritable()) {
- this.processWrite(key);
- }
-
- }
- }
-
- // registering session with data in the write queue for
- // writing
- while (!NioSelectorProcessor.this.flushingSessions
- .isEmpty()) {
- this.processFlushSessions();
- }
- } catch (IOException e) {
- LOGGER.error("IOException while selecting selector", e);
- }
-
- // stop the worker if needed
- NioSelectorProcessor.this.workerLock.lock();
-
- try {
- if (NioSelectorProcessor.this.selector.keys().isEmpty()) {
- NioSelectorProcessor.this.worker = null;
- break;
- }
- } finally {
- NioSelectorProcessor.this.workerLock.unlock();
- }
-
- // check for idle events
- NioSelectorProcessor.this.idleChecker
- .processIdleSession(System.currentTimeMillis());
- }
- } catch (Exception e) {
- LOGGER.error("Unexpected exception : ", e);
- }
- }
-
-
- /**
- * Handles the servers addition
- */
- private void processServerAdd() throws IOException {
-
- NioTcpServer tcpServer;
- while ( ( tcpServer = tcpServersToadd.poll() ) != null ) {
- // register for accept
- SelectionKey key = tcpServer.getServerSocketChannel().register(
- NioSelectorProcessor.this.selector,
- SelectionKey.OP_ACCEPT);
- key.attach(tcpServer);
- tcpServer.setAcceptKey(key);
- LOGGER.debug("registered for accept : {}",tcpServer);
- }
-
- NioUdpServer udpServer;
- while ( ( udpServer = udpServersToadd.poll() ) != null ) {
- // register for read
- SelectionKey key = udpServer.getDatagramChannel().register(
- NioSelectorProcessor.this.selector,
- SelectionKey.OP_READ);
- key.attach(udpServer);
- udpServer.setReadKey(key);
- LOGGER.debug("registered for accept : {}",udpServer);
- }
- }
-
- /**
- * Handles the servers removal
- */
- private void processServerRemove() {
- NioTcpServer tcpServer;
- while ( ( tcpServer = tcpServersToRemove.poll() ) != null ) {
- // find the server key and cancel it
- SelectionKey key = tcpServer.getAcceptKey();
- key.cancel();
- tcpServer.setAcceptKey(null);
- key.attach(null);
- }
- NioUdpServer udpServer;
- while ( ( udpServer = udpServersToRemove.poll() ) != null ) {
- // find the server key and cancel it
- SelectionKey key = udpServer.getReadKey();
- key.cancel();
- udpServer.setReadKey(null);
- key.attach(null);
- }
- }
-
- /**
- * Handles all the sessions that must be connected
- */
- private void processConnectSessions() throws IOException {
- while (!NioSelectorProcessor.this.sessionsToConnect.isEmpty()) {
- NioTcpSession session = NioSelectorProcessor.this.sessionsToConnect
- .poll();
- SelectionKey key = session.getSocketChannel().register(
- NioSelectorProcessor.this.selector,
- SelectionKey.OP_READ);
- key.attach(session);
- this.sessionReadKey.put(session, key);
-
- // Switch to CONNECTED, only if the session is not secured, as
- // the SSL Handshake
- // will occur later.
- if (!session.isSecured()) {
- session.setConnected();
-
- // fire the event
- ((AbstractIoService) session.getService())
- .fireSessionCreated(session);
- session.processSessionOpened();
- long time = System.currentTimeMillis();
- NioSelectorProcessor.this.idleChecker.sessionRead(session,
- time);
- NioSelectorProcessor.this.idleChecker.sessionWritten(
- session, time);
- }
- }
- }
-
- /**
- * Handles all the sessions that must be closed
- */
- private void processCloseSessions() throws IOException {
- while (!NioSelectorProcessor.this.sessionsToClose.isEmpty()) {
- NioTcpSession session = NioSelectorProcessor.this.sessionsToClose
- .poll();
-
- SelectionKey key = this.sessionReadKey.remove(session);
- key.cancel();
-
- // closing underlying socket
- session.getSocketChannel().close();
- // fire the event
- session.processSessionClosed();
- ((AbstractIoService) session.getService())
- .fireSessionDestroyed(session);
- }
- }
-
- /**
- * Processes the Accept action for the given SelectionKey
- */
- private void processAccept(final SelectionKey key) throws IOException {
- LOGGER.debug("acceptable new client {}", key);
- NioTcpServer server = (NioTcpServer) key.attachment();
-
- // accepted connection
- SocketChannel newClientChannel = server.getServerSocketChannel().accept();
- LOGGER.debug("client accepted");
- // and give it's to the strategy
- NioSelectorProcessor.this.strategy.getSelectorForNewSession(
- NioSelectorProcessor.this).createSession(server,
- newClientChannel);
- }
-
- /**
- * Processes the Read action for the given SelectionKey
- */
- private void processRead(final SelectionKey key) throws IOException {
- LOGGER.debug("readable client {}", key);
- NioTcpSession session = (NioTcpSession) key.attachment();
- SocketChannel channel = session.getSocketChannel();
- NioSelectorProcessor.this.readBuffer.clear();
- int readCount = channel.read(NioSelectorProcessor.this.readBuffer);
-
- LOGGER.debug("read {} bytes", readCount);
-
- if (readCount < 0) {
- // session closed by the remote peer
- LOGGER.debug("session closed by the remote peer");
- NioSelectorProcessor.this.sessionsToClose.add(session);
- } else {
- // we have read some data
- // limit at the current position & rewind buffer back to start &
- // push to the chain
- NioSelectorProcessor.this.readBuffer.flip();
-
- if (session.isSecured()) {
- // We are reading data over a SSL/TLS encrypted connection.
- // Redirect
- // the processing to the SslHelper class.
- SslHelper sslHelper = session
- .getAttribute(SSL_HELPER, null);
-
- if (sslHelper == null) {
- throw new IllegalStateException();
- }
-
- sslHelper.processRead(session,
- NioSelectorProcessor.this.readBuffer);
- } else {
- // Plain message, not encrypted : go directly to the chain
- session.processMessageReceived(NioSelectorProcessor.this.readBuffer);
- }
-
- NioSelectorProcessor.this.idleChecker.sessionRead(session,
- System.currentTimeMillis());
- }
- }
-
- /**
- * Processes the Write action for the given SelectionKey
- */
- private void processWrite(final SelectionKey key) throws IOException {
- NioTcpSession session = (NioTcpSession) key.attachment();
-
- LOGGER.debug("writable session : {}", session);
-
- session.setNotRegisteredForWrite();
-
- // write from the session write queue
- boolean isEmpty = false;
-
- try {
- Queue<WriteRequest> queue = session.acquireWriteQueue();
-
- do {
- // get a write request from the queue
- WriteRequest wreq = queue.peek();
-
- if (wreq == null) {
- break;
- }
-
- ByteBuffer buf = (ByteBuffer) wreq.getMessage();
-
- // Note that if the connection is secured, the buffer
- // already
- // contains encrypted data.
- int wrote = session.getSocketChannel().write(buf);
- session.incrementWrittenBytes(wrote);
- LOGGER.debug("wrote {} bytes to {}", wrote, session);
-
- NioSelectorProcessor.this.idleChecker.sessionWritten(
- session, System.currentTimeMillis());
-
- if (buf.remaining() == 0) {
- // completed write request, let's remove it
- queue.remove();
- // complete the future
- DefaultWriteFuture future = (DefaultWriteFuture) wreq
- .getFuture();
-
- if (future != null) {
- future.complete();
- }
- } else {
- // output socket buffer is full, we need
- // to give up until next selection for
- // writing
- break;
- }
- } while (!queue.isEmpty());
-
- isEmpty = queue.isEmpty();
- } finally {
- session.releaseWriteQueue();
- }
-
- // if the session is no more interested in writing, we need
- // to stop listening for OP_WRITE events
- if (isEmpty) {
- if (session.isClosing()) {
- LOGGER.debug(
- "closing session {} have empty write queue, so we close it",
- session);
- // we was flushing writes, now we to the close
- session.getSocketChannel().close();
- } else {
- // a key registered for read ? (because we can have a
- // Selector for reads and another for the writes
- SelectionKey readKey = this.sessionReadKey.get(session);
-
- if (readKey != null) {
- LOGGER.debug("registering key for only reading");
- SelectionKey mykey = session.getSocketChannel()
- .register(NioSelectorProcessor.this.selector,
- SelectionKey.OP_READ, session);
- this.sessionReadKey.put(session, mykey);
- } else {
- LOGGER.debug("cancel key for writing");
- session.getSocketChannel()
- .keyFor(NioSelectorProcessor.this.selector)
- .cancel();
- }
- }
- }
- }
-
- /**
- * Flushes the sessions
- */
- private void processFlushSessions() throws IOException {
- NioTcpSession session = NioSelectorProcessor.this.flushingSessions
- .poll();
- // a key registered for read ? (because we can have a
- // Selector for reads and another for the writes
- SelectionKey readKey = this.sessionReadKey.get(session);
-
- if (readKey != null) {
- // register for read/write
- SelectionKey key = session.getSocketChannel().register(
- NioSelectorProcessor.this.selector,
- SelectionKey.OP_READ | SelectionKey.OP_WRITE, session);
-
- this.sessionReadKey.put(session, key);
- } else {
- session.getSocketChannel().register(
- NioSelectorProcessor.this.selector,
- SelectionKey.OP_WRITE, session);
- }
- }
- }
+public class NioSelectorProcessor implements SelectorProcessor<NioTcpServer, NioUdpServer> {
+
+ /** A logger for this class */
+ private static final Logger LOGGER = LoggerFactory.getLogger(NioSelectorProcessor.class);
+
+ /**
+ * A timeout used for the select, as we need to get out to deal with idle
+ * sessions
+ */
+ private static final long SELECT_TIMEOUT = 1000L;
+
+ private SelectorStrategy<NioSelectorProcessor> strategy;
+
+ /** Read buffer for all the incoming bytes (default to 64Kb) */
+ private final ByteBuffer readBuffer = ByteBuffer.allocate(64 * 1024);
+
+ /** the thread polling and processing the I/O events */
+ private SelectorWorker worker = null;
+
+ /** helper for detecting idleing sessions */
+ private final IdleChecker idleChecker = new IndexedIdleChecker();
+
+ /** A queue containing the servers to bind to this selector */
+ private final Queue<NioTcpServer> tcpServersToadd = new ConcurrentLinkedQueue<NioTcpServer>();
+
+ /** A queue containing the servers to bind to this selector */
+ private final Queue<NioUdpServer> udpServersToadd = new ConcurrentLinkedQueue<NioUdpServer>();
+
+ /** server to remove of the selector */
+ private final Queue<NioTcpServer> tcpServersToRemove = new ConcurrentLinkedQueue<NioTcpServer>();
+
+ /** server to remove of the selector */
+ private final Queue<NioUdpServer> udpServersToRemove = new ConcurrentLinkedQueue<NioUdpServer>();
+
+ /**
+ * new session freshly accepted, placed here for being added to the selector
+ */
+ private final Queue<NioTcpSession> sessionsToConnect = new ConcurrentLinkedQueue<NioTcpSession>();
+
+ /** session to be removed of the selector */
+ private final Queue<NioTcpSession> sessionsToClose = new ConcurrentLinkedQueue<NioTcpSession>();
+
+ /** A queue used to store the sessions to be flushed */
+ private final Queue<NioTcpSession> flushingSessions = new ConcurrentLinkedQueue<NioTcpSession>();
+
+ private Selector selector;
+
+ // Lock for Selector worker, using default. can look into fairness later.
+ // We need to think about a lock less mechanism here.
+ private final Lock workerLock = new ReentrantLock();
+
+ /**
+ * {@inheritDoc}
+ */
+ @SuppressWarnings("unchecked")
+ @Override
+ public void setStrategy(SelectorStrategy<?> strategy) {
+ this.strategy = (SelectorStrategy<NioSelectorProcessor>) strategy;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void addServer(NioTcpServer server) {
+ tcpServersToadd.add(server);
+ this.wakeupWorker();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void addServer(NioUdpServer server) {
+ udpServersToadd.add(server);
+ this.wakeupWorker();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void removeServer(NioTcpServer server) {
+ tcpServersToRemove.add(server);
+ this.wakeupWorker();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void removeServer(NioUdpServer server) {
+ udpServersToRemove.add(server);
+ this.wakeupWorker();
+ }
+
+ /**
+ * Wake the I/O worker thread and if none exists, create a new one FIXME :
+ * too much locking there ?
+ */
+ private void wakeupWorker() {
+ this.workerLock.lock();
+ try {
+ if (this.worker == null) {
+ this.worker = new SelectorWorker();
+ this.worker.start();
+ }
+ } finally {
+ this.workerLock.unlock();
+ }
+
+ if (this.selector != null) {
+ this.selector.wakeup();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void createSession(final IoService service, final Object clientSocket) throws SSLException {
+ LOGGER.debug("create session");
+ final SocketChannel socketChannel = (SocketChannel) clientSocket;
+ final TcpSessionConfig config = (TcpSessionConfig) service.getSessionConfig();
+ final NioTcpSession session = new NioTcpSession(service, socketChannel,
+ this.strategy.getSelectorForNewSession(this), idleChecker);
+
+ try {
+ socketChannel.configureBlocking(false);
+ } catch (IOException e) {
+ LOGGER.error("Unexpected exception, while configuring socket as non blocking", e);
+ throw new RuntimeIoException("cannot configure socket as non-blocking", e);
+ }
+ // 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.getReceiveBufferSize();
+
+ if (receiveBufferSize != null) {
+ session.getConfig().setReceiveBufferSize(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());
+ }
+
+ // event session created
+ session.processSessionCreated();
+
+ // add the session to the queue for being added to the selector
+ this.sessionsToConnect.add(session);
+ this.wakeupWorker();
+ }
+
+ public IdleChecker getIdleChecker() {
+ return idleChecker;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void flush(final AbstractIoSession session) {
+ LOGGER.debug("scheduling session {} for writing", session);
+ // add the session to the list of session to be registered for writing
+ this.flushingSessions.add((NioTcpSession) session);
+ // wake the selector for unlocking the I/O thread
+ this.wakeupWorker();
+ }
+
+ /**
+ * Add the session to list of session to close and remove
+ * @param session
+ */
+ public void addSessionToClose(NioTcpSession session) {
+ sessionsToClose.add(session);
+ }
+
+ public void cancelKeyForWritting(NioTcpSession session) {
+
+ // a key registered for read ? (because we can have a
+ // Selector for reads and another for the writes
+ SelectionKey readKey = this.sessionReadKey.get(session);
+
+ if (readKey != null) {
+ LOGGER.debug("registering key for only reading");
+ SelectionKey mykey;
+ try {
+ mykey = session.getSocketChannel().register(selector, SelectionKey.OP_READ, session);
+ this.sessionReadKey.put(session, mykey);
+ } catch (ClosedChannelException e) {
+ LOGGER.error("already closed session", e);
+ }
+ } else {
+ LOGGER.debug("cancel key for writing");
+ session.getSocketChannel().keyFor(NioSelectorProcessor.this.selector).cancel();
+ }
+ }
+
+ // map for finding read keys associated with a given session
+ private Map<NioTcpSession, SelectionKey> sessionReadKey;
+
+ /**
+ * The worker processing incoming session creation, session destruction
+ * requests, session write and reads. It will also bind new servers.
+ */
+ private class SelectorWorker extends Thread {
+
+ public SelectorWorker() {
+ sessionReadKey = new HashMap<NioTcpSession, SelectionKey>();
+ }
+
+ @Override
+ public void run() {
+ try {
+ if (NioSelectorProcessor.this.selector == null) {
+ LOGGER.debug("opening a new selector");
+
+ try {
+ NioSelectorProcessor.this.selector = Selector.open();
+ } catch (IOException e) {
+ LOGGER.error("IOException while opening a new Selector", e);
+ }
+ }
+
+ for (;;) {
+ try {
+ // pop server sockets for removing
+ if (NioSelectorProcessor.this.tcpServersToRemove.size() > 0
+ || NioSelectorProcessor.this.udpServersToRemove.size() > 0) {
+ this.processServerRemove();
+ }
+
+ // pop new server sockets for accepting
+ if (NioSelectorProcessor.this.tcpServersToadd.size() > 0
+ || NioSelectorProcessor.this.udpServersToadd.size() > 0) {
+ this.processServerAdd();
+ }
+
+ // pop new session for starting read/write
+ if (NioSelectorProcessor.this.sessionsToConnect.size() > 0) {
+ this.processConnectSessions();
+ }
+
+ // pop session for close, if any
+ if (NioSelectorProcessor.this.sessionsToClose.size() > 0) {
+ this.processCloseSessions();
+ }
+
+ LOGGER.debug("selecting...");
+ int readyCount = NioSelectorProcessor.this.selector.select(SELECT_TIMEOUT);
+ LOGGER.debug("... done selecting : {}", readyCount);
+
+ if (readyCount > 0) {
+ // process selected keys
+ Iterator<SelectionKey> selectedKeys = NioSelectorProcessor.this.selector.selectedKeys()
+ .iterator();
+
+ // Loop on each SelectionKey and process any valid
+ // action
+ while (selectedKeys.hasNext()) {
+ SelectionKey key = selectedKeys.next();
+ selectedKeys.remove();
+
+ if (!key.isValid()) {
+ continue;
+ }
+
+ NioSelectorProcessor.this.selector.selectedKeys().remove(key);
+
+ if (key.isAcceptable()) {
+ ((SelectorEventListener) key.attachment()).acceptReady(NioSelectorProcessor.this);
+ }
+
+ if (key.isReadable()) {
+ ((SelectorEventListener) key.attachment()).readReady(NioSelectorProcessor.this,
+ readBuffer);
+ }
+
+ if (key.isWritable()) {
+ ((SelectorEventListener) key.attachment()).writeReady(NioSelectorProcessor.this);
+ }
+
+ }
+ }
+
+ // registering session with data in the write queue for
+ // writing
+ while (!NioSelectorProcessor.this.flushingSessions.isEmpty()) {
+ this.processFlushSessions();
+ }
+ } catch (IOException e) {
+ LOGGER.error("IOException while selecting selector", e);
+ }
+
+ // stop the worker if needed
+ NioSelectorProcessor.this.workerLock.lock();
+
+ try {
+ if (NioSelectorProcessor.this.selector.keys().isEmpty()) {
+ NioSelectorProcessor.this.worker = null;
+ break;
+ }
+ } finally {
+ NioSelectorProcessor.this.workerLock.unlock();
+ }
+
+ // check for idle events
+ NioSelectorProcessor.this.idleChecker.processIdleSession(System.currentTimeMillis());
+ }
+ } catch (Exception e) {
+ LOGGER.error("Unexpected exception : ", e);
+ }
+ }
+
+ /**
+ * Handles the servers addition
+ */
+ private void processServerAdd() throws IOException {
+
+ NioTcpServer tcpServer;
+ while ((tcpServer = tcpServersToadd.poll()) != null) {
+ // register for accept
+ SelectionKey key = tcpServer.getServerSocketChannel().register(NioSelectorProcessor.this.selector,
+ SelectionKey.OP_ACCEPT);
+ key.attach(tcpServer);
+ tcpServer.setAcceptKey(key);
+ LOGGER.debug("registered for accept : {}", tcpServer);
+ }
+
+ NioUdpServer udpServer;
+ while ((udpServer = udpServersToadd.poll()) != null) {
+ // register for read
+ SelectionKey key = udpServer.getDatagramChannel().register(NioSelectorProcessor.this.selector,
+ SelectionKey.OP_READ);
+ key.attach(udpServer);
+ udpServer.setReadKey(key);
+ LOGGER.debug("registered for accept : {}", udpServer);
+ }
+ }
+
+ /**
+ * Handles the servers removal
+ */
+ private void processServerRemove() {
+ NioTcpServer tcpServer;
+ while ((tcpServer = tcpServersToRemove.poll()) != null) {
+ // find the server key and cancel it
+ SelectionKey key = tcpServer.getAcceptKey();
+ key.cancel();
+ tcpServer.setAcceptKey(null);
+ key.attach(null);
+ }
+ NioUdpServer udpServer;
+ while ((udpServer = udpServersToRemove.poll()) != null) {
+ // find the server key and cancel it
+ SelectionKey key = udpServer.getReadKey();
+ key.cancel();
+ udpServer.setReadKey(null);
+ key.attach(null);
+ }
+ }
+
+ /**
+ * Handles all the sessions that must be connected
+ */
+ private void processConnectSessions() throws IOException {
+ while (!NioSelectorProcessor.this.sessionsToConnect.isEmpty()) {
+ NioTcpSession session = NioSelectorProcessor.this.sessionsToConnect.poll();
+ SelectionKey key = session.getSocketChannel().register(NioSelectorProcessor.this.selector,
+ SelectionKey.OP_READ);
+ key.attach(session);
+
+ sessionReadKey.put(session, key);
+
+ // Switch to CONNECTED, only if the session is not secured, as
+ // the SSL Handshake
+ // will occur later.
+ if (!session.isSecured()) {
+ session.setConnected();
+
+ // fire the event
+ ((AbstractIoService) session.getService()).fireSessionCreated(session);
+ session.processSessionOpened();
+ long time = System.currentTimeMillis();
+ NioSelectorProcessor.this.idleChecker.sessionRead(session, time);
+ NioSelectorProcessor.this.idleChecker.sessionWritten(session, time);
+ }
+ }
+ }
+
+ /**
+ * Handles all the sessions that must be closed
+ */
+ private void processCloseSessions() throws IOException {
+ while (!NioSelectorProcessor.this.sessionsToClose.isEmpty()) {
+ NioTcpSession session = NioSelectorProcessor.this.sessionsToClose.poll();
+
+ SelectionKey key = sessionReadKey.remove(session);
+ key.cancel();
+
+ // closing underlying socket
+ session.getSocketChannel().close();
+ // fire the event
+ session.processSessionClosed();
+ ((AbstractIoService) session.getService()).fireSessionDestroyed(session);
+ }
+ }
+
+ /**
+ * Flushes the sessions
+ */
+ private void processFlushSessions() throws IOException {
+ NioTcpSession session = NioSelectorProcessor.this.flushingSessions.poll();
+ // a key registered for read ? (because we can have a
+ // Selector for reads and another for the writes
+ SelectionKey readKey = sessionReadKey.get(session);
+
+ if (readKey != null) {
+ // register for read/write
+ SelectionKey key = session.getSocketChannel().register(NioSelectorProcessor.this.selector,
+ SelectionKey.OP_READ | SelectionKey.OP_WRITE, session);
+
+ sessionReadKey.put(session, key);
+ } else {
+ session.getSocketChannel().register(NioSelectorProcessor.this.selector, SelectionKey.OP_WRITE, session);
+ }
+ }
+ }
}
Propchange: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioSelectorProcessor.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Copied: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpClient.java (from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/nio/NioTcpClient.java)
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpClient.java?p2=mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpClient.java&p1=mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/nio/NioTcpClient.java&r1=1341901&r2=1342586&rev=1342586&view=diff
==============================================================================
--- mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/nio/NioTcpClient.java (original)
+++ mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpClient.java Fri May 25 11:58:20 2012
@@ -17,7 +17,7 @@
* under the License.
*
*/
-package org.apache.mina.transport.tcp.nio;
+package org.apache.mina.transport.nio;
import org.apache.mina.api.IoSessionConfig;
import org.apache.mina.transport.tcp.AbstractTcpClient;
Copied: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java (from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/nio/NioTcpServer.java)
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java?p2=mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java&p1=mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/nio/NioTcpServer.java&r1=1341901&r2=1342586&rev=1342586&view=diff
==============================================================================
--- mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/nio/NioTcpServer.java (original)
+++ mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java Fri May 25 11:58:20 2012
@@ -17,16 +17,17 @@
* under the License.
*
*/
-package org.apache.mina.transport.tcp.nio;
+package org.apache.mina.transport.nio;
import java.io.IOException;
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.service.SelectorStrategy;
import org.apache.mina.transport.tcp.AbstractTcpServer;
-import org.apache.mina.transport.tcp.NioSelectorProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,39 +36,40 @@ import org.slf4j.LoggerFactory;
*
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
*/
-public class NioTcpServer extends AbstractTcpServer {
+public class NioTcpServer extends AbstractTcpServer implements SelectorEventListener {
static final Logger LOG = LoggerFactory.getLogger(NioTcpServer.class);
-
+
// the strategy for dispatching servers and client to selector threads.
private final SelectorStrategy<NioSelectorProcessor> strategy;
// the bound local address
private SocketAddress address = null;
-
+
private NioSelectorProcessor acceptProcessor = null;
-
+
// the key used for selecting accept event
private SelectionKey acceptKey = null;
-
+
// the server socket for accepting clients
- private ServerSocketChannel serverChannel = null;
-
+ private ServerSocketChannel serverChannel = null;
+
public NioTcpServer(final SelectorStrategy<NioSelectorProcessor> strategy) {
super();
this.strategy = strategy;
}
-
+
/**
* Get the inner Server socket for accepting new client connections
* @return
*/
public ServerSocketChannel getServerSocketChannel() {
- return this.serverChannel;
+ return this.serverChannel;
}
public void setServerSocketChannel(ServerSocketChannel serverChannel) {
- this.serverChannel = serverChannel;
+ this.serverChannel = serverChannel;
}
+
/**
* {@inheritDoc}
*/
@@ -85,16 +87,16 @@ public class NioTcpServer extends Abstra
LOG.info("binding address {}", localAddress);
this.address = localAddress;
-
+
serverChannel = ServerSocketChannel.open();
serverChannel.socket().setReuseAddress(isReuseAddress());
serverChannel.socket().bind(address);
serverChannel.configureBlocking(false);
acceptProcessor = this.strategy.getSelectorForBindNewAddress();
-
+
acceptProcessor.addServer(this);
-
+
// it's the first address bound, let's fire the event
this.fireServiceActivated();
}
@@ -104,7 +106,7 @@ public class NioTcpServer extends Abstra
*/
@Override
public SocketAddress getBoundAddress() {
- return address;
+ return address;
}
/**
@@ -112,30 +114,60 @@ public class NioTcpServer extends Abstra
*/
@Override
public synchronized void unbind() throws IOException {
- LOG.info("unbinding {}", address);
- if( this.address == null) {
- throw new IllegalStateException("server not bound");
+ LOG.info("unbinding {}", address);
+ if (this.address == null) {
+ throw new IllegalStateException("server not bound");
}
serverChannel.socket().close();
serverChannel.close();
- acceptProcessor.removeServer(this);
-
+ acceptProcessor.removeServer(this);
+
this.address = null;
this.fireServiceInactivated();
}
- /**
- * @return the acceptKey
- */
- public SelectionKey getAcceptKey() {
- return acceptKey;
- }
-
- /**
- * @param acceptKey the acceptKey to set
- */
- public void setAcceptKey(SelectionKey acceptKey) {
- this.acceptKey = acceptKey;
- }
-
+ /**
+ * @return the acceptKey
+ */
+ public SelectionKey getAcceptKey() {
+ return acceptKey;
+ }
+
+ /**
+ * @param acceptKey the acceptKey to set
+ */
+ public void setAcceptKey(SelectionKey acceptKey) {
+ this.acceptKey = acceptKey;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void acceptReady(NioSelectorProcessor processor) throws IOException {
+ LOG.debug("acceptable new client");
+
+ // accepted connection
+ SocketChannel newClientChannel = getServerSocketChannel().accept();
+ LOG.debug("client accepted");
+
+ // and give it's to the strategy
+ strategy.getSelectorForNewSession(processor).createSession(this, newClientChannel);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void readReady(NioSelectorProcessor processor, ByteBuffer readBuffer) {
+ throw new IllegalStateException("read event should never occur on NioTcpServer");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void writeReady(NioSelectorProcessor processor) {
+ throw new IllegalStateException("write event should never occur on NioTcpServer");
+ }
}
\ No newline at end of file
Propchange: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpServer.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Copied: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpSession.java (from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/NioTcpSession.java)
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpSession.java?p2=mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpSession.java&p1=mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/NioTcpSession.java&r1=1341901&r2=1342586&rev=1342586&view=diff
==============================================================================
--- mina/trunk/core/src/main/java/org/apache/mina/transport/tcp/NioTcpSession.java (original)
+++ mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpSession.java Fri May 25 11:58:20 2012
@@ -17,20 +17,25 @@
* under the License.
*
*/
-package org.apache.mina.transport.tcp;
+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.SocketChannel;
+import java.util.Queue;
import org.apache.mina.api.IoFuture;
import org.apache.mina.api.IoService;
import org.apache.mina.api.RuntimeIoException;
-import org.apache.mina.service.SelectorProcessor;
+import org.apache.mina.service.idlechecker.IdleChecker;
import org.apache.mina.session.AbstractIoSession;
-import org.apache.mina.transport.tcp.nio.NioTcpClient;
-import org.apache.mina.transport.tcp.nio.NioTcpServer;
+import org.apache.mina.session.DefaultWriteFuture;
+import org.apache.mina.session.SslHelper;
+import org.apache.mina.session.WriteRequest;
+import org.apache.mina.transport.tcp.ProxyTcpSessionConfig;
+import org.apache.mina.transport.tcp.TcpSessionConfig;
import org.apache.mina.util.AbstractIoFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,19 +47,16 @@ import org.slf4j.LoggerFactory;
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
*
*/
-public class NioTcpSession extends AbstractIoSession {
+public class NioTcpSession extends AbstractIoSession implements SelectorEventListener {
private static final Logger LOG = LoggerFactory.getLogger(NioTcpSession.class);
/** the NIO socket channel for this TCP session */
- private SocketChannel channel;
+ private final SocketChannel channel;
/** the socket configuration */
private final TcpSessionConfig configuration;
- // this session requested to close
- private volatile boolean closeRequested = false;
-
/** we pre-allocate a close future for lock-less {@link #close(boolean)} */
private final IoFuture<Void> closeFuture = new AbstractIoFuture<Void>() {
@@ -68,8 +70,8 @@ public class NioTcpSession extends Abstr
}
};
- NioTcpSession(IoService service, SocketChannel channel, SelectorProcessor writeProcessor) {
- super(service, writeProcessor);
+ NioTcpSession(IoService service, SocketChannel channel, NioSelectorProcessor writeProcessor, IdleChecker idleChecker) {
+ super(service, writeProcessor, idleChecker);
this.channel = channel;
this.configuration = new ProxyTcpSessionConfig(channel.socket());
}
@@ -127,7 +129,6 @@ public class NioTcpSession extends Abstr
LOG.error("Session {} not opened", this);
throw new RuntimeIoException("cannot close an not opened session");
case CONNECTED:
- closeRequested = true;
state = SessionState.CLOSING;
if (immediately) {
try {
@@ -226,4 +227,125 @@ public class NioTcpSession extends Abstr
state = SessionState.CONNECTED;
}
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void acceptReady(NioSelectorProcessor processor) {
+ // should never happen
+ throw new IllegalStateException("accept event should never occur on NioTcpSession");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void readReady(NioSelectorProcessor processor, ByteBuffer readBuffer) throws IOException {
+ LOG.debug("readable session : {}", this);
+ readBuffer.clear();
+ int readCount = 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");
+ processor.addSessionToClose(this);
+ } else {
+ // 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.
+ SslHelper sslHelper = getAttribute(SSL_HELPER, null);
+
+ if (sslHelper == null) {
+ throw new IllegalStateException();
+ }
+
+ sslHelper.processRead(this, readBuffer);
+ } else {
+ // Plain message, not encrypted : go directly to the chain
+ processMessageReceived(readBuffer);
+ }
+
+ idleChecker.sessionRead(this, System.currentTimeMillis());
+ }
+
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void writeReady(NioSelectorProcessor processor) throws IOException {
+ LOG.debug("writable session : {}", this);
+
+ setNotRegisteredForWrite();
+
+ // write from the session write queue
+ boolean isEmpty = false;
+
+ try {
+ Queue<WriteRequest> queue = acquireWriteQueue();
+
+ do {
+ // get a write request from the queue
+ WriteRequest wreq = queue.peek();
+
+ if (wreq == null) {
+ break;
+ }
+
+ ByteBuffer buf = (ByteBuffer) wreq.getMessage();
+
+ // Note that if the connection is secured, the buffer
+ // already
+ // contains encrypted data.
+ int wrote = getSocketChannel().write(buf);
+ incrementWrittenBytes(wrote);
+ LOG.debug("wrote {} bytes to {}", wrote, this);
+
+ idleChecker.sessionWritten(this, System.currentTimeMillis());
+
+ if (buf.remaining() == 0) {
+ // completed write request, let's remove it
+ queue.remove();
+ // complete the future
+ DefaultWriteFuture future = (DefaultWriteFuture) wreq.getFuture();
+
+ if (future != null) {
+ future.complete();
+ }
+ } else {
+ // output socket buffer is full, we need
+ // to give up until next selection for
+ // writing
+ break;
+ }
+ } while (!queue.isEmpty());
+
+ isEmpty = queue.isEmpty();
+ } finally {
+ this.releaseWriteQueue();
+ }
+
+ // if the session is no more interested in writing, we need
+ // to stop listening for OP_WRITE events
+ if (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
+ getSocketChannel().close();
+ } else {
+ // no more write event needed
+ processor.cancelKeyForWritting(this);
+ }
+ }
+ }
}
Propchange: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioTcpSession.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Copied: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpClient.java (from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/udp/nio/NioUdpClient.java)
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpClient.java?p2=mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpClient.java&p1=mina/trunk/core/src/main/java/org/apache/mina/transport/udp/nio/NioUdpClient.java&r1=1341901&r2=1342586&rev=1342586&view=diff
==============================================================================
--- mina/trunk/core/src/main/java/org/apache/mina/transport/udp/nio/NioUdpClient.java (original)
+++ mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpClient.java Fri May 25 11:58:20 2012
@@ -17,7 +17,7 @@
* under the License.
*
*/
-package org.apache.mina.transport.udp.nio;
+package org.apache.mina.transport.nio;
import org.apache.mina.api.IoSessionConfig;
import org.apache.mina.transport.udp.AbstractUdpClient;
Copied: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpServer.java (from r1341901, mina/trunk/core/src/main/java/org/apache/mina/transport/udp/nio/NioUdpServer.java)
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpServer.java?p2=mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpServer.java&p1=mina/trunk/core/src/main/java/org/apache/mina/transport/udp/nio/NioUdpServer.java&r1=1341901&r2=1342586&rev=1342586&view=diff
==============================================================================
--- mina/trunk/core/src/main/java/org/apache/mina/transport/udp/nio/NioUdpServer.java (original)
+++ mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpServer.java Fri May 25 11:58:20 2012
@@ -17,16 +17,18 @@
* under the License.
*
*/
-package org.apache.mina.transport.udp.nio;
+package org.apache.mina.transport.nio;
import java.io.IOException;
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.IoSessionConfig;
import org.apache.mina.service.SelectorStrategy;
-import org.apache.mina.transport.tcp.NioSelectorProcessor;
import org.apache.mina.transport.udp.AbstractUdpServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,25 +38,28 @@ import org.slf4j.LoggerFactory;
*
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
*/
-public class NioUdpServer extends AbstractUdpServer {
-
- static final Logger LOG = LoggerFactory.getLogger(NioUdpServer.class);
+public class NioUdpServer extends AbstractUdpServer implements SelectorEventListener {
+
+ static final Logger LOG = LoggerFactory.getLogger(NioUdpServer.class);
// the bound local address
- private SocketAddress address = null;
-
+ private SocketAddress address = null;
+
// the strategy for dispatching servers and client to selector threads.
private final SelectorStrategy<NioSelectorProcessor> strategy;
// the processor used for read and write this server
private NioSelectorProcessor processor;
-
+
// 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>();
+
/**
* Create a new instance of NioUdpServer
*/
@@ -70,9 +75,9 @@ public class NioUdpServer extends Abstra
* @return the datagram channel bound to this {@link NioUdpServer}.
*/
public DatagramChannel getDatagramChannel() {
- return datagramChannel;
+ return datagramChannel;
}
-
+
/**
* {@inheritDoc}
*/
@@ -82,15 +87,14 @@ public class NioUdpServer extends Abstra
return null;
}
-
/**
* {@inheritDoc}
*/
@Override
public SocketAddress getBoundAddress() {
- return address;
+ return address;
}
-
+
/**
* {@inheritDoc}
*/
@@ -106,53 +110,94 @@ public class NioUdpServer extends Abstra
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);
processor = this.strategy.getSelectorForBindNewAddress();
-
+
processor.addServer(this);
-
+
// 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");
+ LOG.info("unbinding {}", address);
+ if (this.address == null) {
+ throw new IllegalStateException("server not bound");
}
datagramChannel.socket().close();
datagramChannel.close();
-
- processor.removeServer(this);
-
+
+ processor.removeServer(this);
+
this.address = null;
this.fireServiceInactivated();
}
- /**
- * @return the readKey
- */
- public SelectionKey getReadKey() {
- return readKey;
- }
-
- /**
- * @param readKey the readKey to set
- */
- public void setReadKey(SelectionKey readKey) {
- this.readKey = readKey;
- }
-
+ /**
+ * @return the readKey
+ */
+ public SelectionKey getReadKey() {
+ return readKey;
+ }
+
+ /**
+ * @param readKey the readKey to set
+ */
+ public void setReadKey(SelectionKey readKey) {
+ this.readKey = readKey;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void acceptReady(NioSelectorProcessor processor) {
+ throw new IllegalStateException("accept event should never occur on NioUdpServer");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void readReady(NioSelectorProcessor processor, ByteBuffer readBuffer) throws IOException {
+ LOG.debug("readable datagram for UDP service : {}", this);
+ readBuffer.clear();
+
+ SocketAddress source = datagramChannel.receive(readBuffer);
+ readBuffer.flip();
+
+ LOG.debug("read {} bytes form {}", readBuffer.remaining(), source);
+
+ // let's find the corresponding session
+
+ NioUdpSession session = sessions.get(source);
+ if (session == null) {
+ session = new NioUdpSession(this, strategy.getSelectorForNewSession(processor), processor.getIdleChecker(),
+ address, source);
+ }
+
+ session.receivedDatagram(readBuffer);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void writeReady(NioSelectorProcessor processor) throws IOException {
+ // TODO : flush the sessions
+
+ }
+
}
\ No newline at end of file
Added: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpSession.java
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpSession.java?rev=1342586&view=auto
==============================================================================
--- mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpSession.java (added)
+++ mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpSession.java Fri May 25 11:58:20 2012
@@ -0,0 +1,185 @@
+/**
+ * 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.net.SocketAddress;
+import java.nio.ByteBuffer;
+
+import org.apache.mina.api.IoFuture;
+import org.apache.mina.api.IoService;
+import org.apache.mina.api.IoSessionConfig;
+import org.apache.mina.api.RuntimeIoException;
+import org.apache.mina.service.idlechecker.IdleChecker;
+import org.apache.mina.session.AbstractIoSession;
+import org.apache.mina.util.AbstractIoFuture;
+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 AbstractIoSession {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NioUdpSession.class);
+
+ private final SocketAddress localAddress;
+
+ private final SocketAddress remoteAddress;
+
+ /** 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(boolean mayInterruptIfRunning) {
+ // we don't cancel close
+ return false;
+ }
+ };
+
+ /**
+ * @param service
+ * @param writeProcessor
+ * @param idleChecker
+ */
+ public NioUdpSession(IoService service, NioSelectorProcessor writeProcessor, IdleChecker idleChecker,
+ SocketAddress localAddress, SocketAddress remoteAddress) {
+ super(service, writeProcessor, idleChecker);
+ this.localAddress = localAddress;
+ this.remoteAddress = remoteAddress;
+ }
+
+ /**
+ * {@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 RuntimeIoException("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 RuntimeException("Not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void suspendWrite() {
+ // TODO
+ throw new RuntimeException("Not implemented");
+ }
+
+ /**
+ * {@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
+ throw new RuntimeException("Not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isWriteSuspended() {
+ // TODO
+ throw new RuntimeException("Not implemented");
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public IoSessionConfig getConfig() {
+ return null;
+ }
+
+ /**
+ * Called when the session received a datagram.
+ * @param readBuffer the received datagram
+ */
+ void receivedDatagram(ByteBuffer readBuffer) {
+ processMessageReceived(readBuffer);
+ idleChecker.sessionRead(this, System.currentTimeMillis());
+ }
+}
Propchange: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/NioUdpSession.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/SelectorEventListener.java
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/main/java/org/apache/mina/transport/nio/SelectorEventListener.java?rev=1342586&view=auto
==============================================================================
--- mina/trunk/core/src/main/java/org/apache/mina/transport/nio/SelectorEventListener.java (added)
+++ mina/trunk/core/src/main/java/org/apache/mina/transport/nio/SelectorEventListener.java Fri May 25 11:58:20 2012
@@ -0,0 +1,49 @@
+/**
+ * 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;
+
+/**
+ * Listener for I/O events generated by {@link NioSelectorProcessor}.
+ *
+ * @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ */
+public interface SelectorEventListener {
+
+ /**
+ * A read ready event
+ * @param processor the processor generating this event
+ * @param readBuffer the buffer to be used for reading
+ */
+ void readReady(NioSelectorProcessor processor, ByteBuffer readBuffer) throws IOException;
+
+ /**
+ * A write ready event
+ * @param processor the processor generating this event
+ */
+ void writeReady(NioSelectorProcessor processor) throws IOException;
+
+ /**
+ * An accept ready event
+ * @param processor the processor generating this event
+ */
+ void acceptReady(NioSelectorProcessor processor) throws IOException;
+}
Propchange: mina/trunk/core/src/main/java/org/apache/mina/transport/nio/SelectorEventListener.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: mina/trunk/core/src/test/java/org/apache/mina/service/idlecheker/IndexedIdleChekerTest.java
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/test/java/org/apache/mina/service/idlecheker/IndexedIdleChekerTest.java?rev=1342586&r1=1342585&r2=1342586&view=diff
==============================================================================
--- mina/trunk/core/src/test/java/org/apache/mina/service/idlecheker/IndexedIdleChekerTest.java (original)
+++ mina/trunk/core/src/test/java/org/apache/mina/service/idlecheker/IndexedIdleChekerTest.java Fri May 25 11:58:20 2012
@@ -31,6 +31,7 @@ import org.apache.mina.api.IoFuture;
import org.apache.mina.api.IoService;
import org.apache.mina.api.IoSessionConfig;
import org.apache.mina.service.SelectorProcessor;
+import org.apache.mina.service.idlechecker.IdleChecker;
import org.apache.mina.service.idlechecker.IndexedIdleChecker;
import org.apache.mina.session.AbstractIoSession;
import org.apache.mina.session.AbstractIoSessionConfig;
@@ -43,9 +44,9 @@ import org.junit.Test;
*/
public class IndexedIdleChekerTest {
- private IndexedIdleChecker idleChecker = new IndexedIdleChecker();
+ private final IndexedIdleChecker idleChecker = new IndexedIdleChecker();
- private long now = System.currentTimeMillis();
+ private final long now = System.currentTimeMillis();
@Test
public void process_on_empty_index() {
@@ -55,7 +56,7 @@ public class IndexedIdleChekerTest {
@Test
public void read_event() {
IoService service = mock(IoService.class);
- DummySession session = new DummySession(service);
+ DummySession session = new DummySession(service, idleChecker);
session.getConfig().setIdleTimeInMillis(IdleStatus.READ_IDLE, 1000L);
@@ -72,7 +73,7 @@ public class IndexedIdleChekerTest {
@Test
public void write_event() {
IoService service = mock(IoService.class);
- DummySession session = new DummySession(service);
+ DummySession session = new DummySession(service, idleChecker);
session.getConfig().setIdleTimeInMillis(IdleStatus.WRITE_IDLE, 1000L);
@@ -84,7 +85,7 @@ public class IndexedIdleChekerTest {
assertEquals(1, session.writeIdleCount);
}
- private SelectorProcessor processor = mock(SelectorProcessor.class);
+ private final SelectorProcessor processor = mock(SelectorProcessor.class);
private class DummySession extends AbstractIoSession {
@@ -92,8 +93,8 @@ public class IndexedIdleChekerTest {
int writeIdleCount = 0;
- private DummySession(IoService service) {
- super(service, processor);
+ private DummySession(IoService service, IdleChecker checker) {
+ super(service, processor, checker);
}
@Override
Modified: mina/trunk/core/src/test/java/org/apache/mina/session/AbstractIoSessionTest.java
URL: http://svn.apache.org/viewvc/mina/trunk/core/src/test/java/org/apache/mina/session/AbstractIoSessionTest.java?rev=1342586&r1=1342585&r2=1342586&view=diff
==============================================================================
--- mina/trunk/core/src/test/java/org/apache/mina/session/AbstractIoSessionTest.java (original)
+++ mina/trunk/core/src/test/java/org/apache/mina/session/AbstractIoSessionTest.java Fri May 25 11:58:20 2012
@@ -18,9 +18,16 @@
*/
package org.apache.mina.session;
-import static junit.framework.Assert.*;
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.*;
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertNotNull;
+import static junit.framework.Assert.assertNull;
+import static junit.framework.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
@@ -45,11 +52,11 @@ import org.junit.Test;
*/
public class AbstractIoSessionTest {
- private SelectorProcessor processor = mock(SelectorProcessor.class);
+ private final SelectorProcessor processor = mock(SelectorProcessor.class);
private class DummySession extends AbstractIoSession {
private DummySession(IoService service) {
- super(service, processor);
+ super(service, processor, null);
}
@Override
@@ -122,11 +129,11 @@ public class AbstractIoSessionTest {
private IoService service = null;
- private IoFilter filter1 = spy(new PassthruFilter());
+ private final IoFilter filter1 = spy(new PassthruFilter());
- private IoFilter filter2 = spy(new PassthruFilter());
+ private final IoFilter filter2 = spy(new PassthruFilter());
- private IoFilter filter3 = spy(new PassthruFilter());
+ private final IoFilter filter3 = spy(new PassthruFilter());
@Before
public void setup() {
Modified: mina/trunk/examples/src/main/java/org/apache/mina/examples/echoserver/NioEchoServer.java
URL: http://svn.apache.org/viewvc/mina/trunk/examples/src/main/java/org/apache/mina/examples/echoserver/NioEchoServer.java?rev=1342586&r1=1342585&r2=1342586&view=diff
==============================================================================
--- mina/trunk/examples/src/main/java/org/apache/mina/examples/echoserver/NioEchoServer.java (original)
+++ mina/trunk/examples/src/main/java/org/apache/mina/examples/echoserver/NioEchoServer.java Fri May 25 11:58:20 2012
@@ -34,8 +34,8 @@ import org.apache.mina.filter.logging.Lo
import org.apache.mina.filterchain.ReadFilterChainController;
import org.apache.mina.filterchain.WriteFilterChainController;
import org.apache.mina.service.OneThreadSelectorStrategy;
-import org.apache.mina.transport.tcp.NioSelectorProcessor;
-import org.apache.mina.transport.tcp.nio.NioTcpServer;
+import org.apache.mina.transport.nio.NioSelectorProcessor;
+import org.apache.mina.transport.nio.NioTcpServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Modified: mina/trunk/examples/src/main/java/org/apache/mina/examples/http/HttpTest.java
URL: http://svn.apache.org/viewvc/mina/trunk/examples/src/main/java/org/apache/mina/examples/http/HttpTest.java?rev=1342586&r1=1342585&r2=1342586&view=diff
==============================================================================
--- mina/trunk/examples/src/main/java/org/apache/mina/examples/http/HttpTest.java (original)
+++ mina/trunk/examples/src/main/java/org/apache/mina/examples/http/HttpTest.java Fri May 25 11:58:20 2012
@@ -39,8 +39,8 @@ import org.apache.mina.http.api.HttpRequ
import org.apache.mina.http.api.HttpStatus;
import org.apache.mina.http.api.HttpVersion;
import org.apache.mina.service.OneThreadSelectorStrategy;
-import org.apache.mina.transport.tcp.NioSelectorProcessor;
-import org.apache.mina.transport.tcp.nio.NioTcpServer;
+import org.apache.mina.transport.nio.NioSelectorProcessor;
+import org.apache.mina.transport.nio.NioTcpServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Modified: mina/trunk/examples/src/main/java/org/apache/mina/examples/http/HttpsTest.java
URL: http://svn.apache.org/viewvc/mina/trunk/examples/src/main/java/org/apache/mina/examples/http/HttpsTest.java?rev=1342586&r1=1342585&r2=1342586&view=diff
==============================================================================
--- mina/trunk/examples/src/main/java/org/apache/mina/examples/http/HttpsTest.java (original)
+++ mina/trunk/examples/src/main/java/org/apache/mina/examples/http/HttpsTest.java Fri May 25 11:58:20 2012
@@ -39,8 +39,8 @@ import org.apache.mina.http.api.HttpRequ
import org.apache.mina.http.api.HttpStatus;
import org.apache.mina.http.api.HttpVersion;
import org.apache.mina.service.OneThreadSelectorStrategy;
-import org.apache.mina.transport.tcp.NioSelectorProcessor;
-import org.apache.mina.transport.tcp.nio.NioTcpServer;
+import org.apache.mina.transport.nio.NioSelectorProcessor;
+import org.apache.mina.transport.nio.NioTcpServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Modified: mina/trunk/examples/src/main/java/org/apache/mina/examples/ldap/LdapTest.java
URL: http://svn.apache.org/viewvc/mina/trunk/examples/src/main/java/org/apache/mina/examples/ldap/LdapTest.java?rev=1342586&r1=1342585&r2=1342586&view=diff
==============================================================================
--- mina/trunk/examples/src/main/java/org/apache/mina/examples/ldap/LdapTest.java (original)
+++ mina/trunk/examples/src/main/java/org/apache/mina/examples/ldap/LdapTest.java Fri May 25 11:58:20 2012
@@ -40,8 +40,8 @@ import org.apache.mina.filter.logging.Lo
import org.apache.mina.filterchain.ReadFilterChainController;
import org.apache.mina.ldap.LdapCodec;
import org.apache.mina.service.OneThreadSelectorStrategy;
-import org.apache.mina.transport.tcp.NioSelectorProcessor;
-import org.apache.mina.transport.tcp.nio.NioTcpServer;
+import org.apache.mina.transport.nio.NioSelectorProcessor;
+import org.apache.mina.transport.nio.NioTcpServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Modified: mina/trunk/examples/src/main/java/org/apache/mina/examples/udpecho/NioUdpEchoServer.java
URL: http://svn.apache.org/viewvc/mina/trunk/examples/src/main/java/org/apache/mina/examples/udpecho/NioUdpEchoServer.java?rev=1342586&r1=1342585&r2=1342586&view=diff
==============================================================================
--- mina/trunk/examples/src/main/java/org/apache/mina/examples/udpecho/NioUdpEchoServer.java (original)
+++ mina/trunk/examples/src/main/java/org/apache/mina/examples/udpecho/NioUdpEchoServer.java Fri May 25 11:58:20 2012
@@ -33,8 +33,8 @@ import org.apache.mina.filter.logging.Lo
import org.apache.mina.filterchain.ReadFilterChainController;
import org.apache.mina.filterchain.WriteFilterChainController;
import org.apache.mina.service.OneThreadSelectorStrategy;
-import org.apache.mina.transport.tcp.NioSelectorProcessor;
-import org.apache.mina.transport.udp.nio.NioUdpServer;
+import org.apache.mina.transport.nio.NioSelectorProcessor;
+import org.apache.mina.transport.nio.NioUdpServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;