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 2010/10/09 19:06:30 UTC
svn commit: r1006194 - in /mina/branches/2.0.1:
mina-core/src/main/java/org/apache/mina/core/polling/
mina-core/src/main/java/org/apache/mina/core/service/
mina-core/src/main/java/org/apache/mina/transport/socket/nio/
mina-core/src/test/java/org/apache...
Author: elecharny
Date: Sat Oct 9 17:06:30 2010
New Revision: 1006194
URL: http://svn.apache.org/viewvc?rev=1006194&view=rev
Log:
o Added some missing Javadoc
o Replaced the <T> parameter by a <S>, as <S> is closer to Session than <T>
Modified:
mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingConnectionlessIoAcceptor.java
mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java
mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/service/IoProcessor.java
mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/service/SimpleIoProcessorPool.java
mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/transport/socket/nio/NioProcessor.java
mina/branches/2.0.1/mina-core/src/test/java/org/apache/mina/transport/socket/nio/PollingIoProcessorTest.java
mina/branches/2.0.1/mina-transport-apr/src/main/java/org/apache/mina/transport/socket/apr/AprIoProcessor.java
Modified: mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingConnectionlessIoAcceptor.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingConnectionlessIoAcceptor.java?rev=1006194&r1=1006193&r2=1006194&view=diff
==============================================================================
--- mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingConnectionlessIoAcceptor.java (original)
+++ mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingConnectionlessIoAcceptor.java Sat Oct 9 17:06:30 2010
@@ -54,8 +54,10 @@ import org.apache.mina.util.ExceptionMon
*
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
* @org.apache.xbean.XBean
- */
-public abstract class AbstractPollingConnectionlessIoAcceptor<T extends AbstractIoSession, H>
+ *
+ * @param <S> the type of the {@link IoSession} this processor can handle
+*/
+public abstract class AbstractPollingConnectionlessIoAcceptor<S extends AbstractIoSession, H>
extends AbstractIoAcceptor {
private static final IoSessionRecycler DEFAULT_RECYCLER = new ExpiringSessionRecycler();
@@ -67,12 +69,14 @@ public abstract class AbstractPollingCon
private static final long SELECT_TIMEOUT = 1000L;
private final Object lock = new Object();
- private final IoProcessor<T> processor = new ConnectionlessAcceptorProcessor();
+
+ private final IoProcessor<S> processor = new ConnectionlessAcceptorProcessor();
private final Queue<AcceptorOperationFuture> registerQueue =
new ConcurrentLinkedQueue<AcceptorOperationFuture>();
private final Queue<AcceptorOperationFuture> cancelQueue =
new ConcurrentLinkedQueue<AcceptorOperationFuture>();
- private final Queue<T> flushingSessions = new ConcurrentLinkedQueue<T>();
+
+ private final Queue<S> flushingSessions = new ConcurrentLinkedQueue<S>();
private final Map<String, H> boundHandles =
Collections.synchronizedMap(new HashMap<String, H>());
@@ -152,9 +156,12 @@ public abstract class AbstractPollingCon
protected abstract boolean isReadable(H handle);
protected abstract boolean isWritable(H handle);
protected abstract SocketAddress receive(H handle, IoBuffer buffer) throws Exception;
- protected abstract int send(T session, IoBuffer buffer, SocketAddress remoteAddress) throws Exception;
- protected abstract T newSession(IoProcessor<T> processor, H handle, SocketAddress remoteAddress) throws Exception;
- protected abstract void setInterestedInWrite(T session, boolean interested) throws Exception;
+
+ protected abstract int send(S session, IoBuffer buffer, SocketAddress remoteAddress) throws Exception;
+
+ protected abstract S newSession(IoProcessor<S> processor, H handle, SocketAddress remoteAddress) throws Exception;
+
+ protected abstract void setInterestedInWrite(S session, boolean interested) throws Exception;
/**
* {@inheritDoc}
@@ -275,7 +282,7 @@ public abstract class AbstractPollingCon
}
// If a new session needs to be created.
- T newSession = newSession(processor, handle, remoteAddress);
+ S newSession = newSession(processor, handle, remoteAddress);
getSessionRecycler().put(newSession);
session = newSession;
}
@@ -311,23 +318,23 @@ public abstract class AbstractPollingCon
}
}
- private class ConnectionlessAcceptorProcessor implements IoProcessor<T> {
+ private class ConnectionlessAcceptorProcessor implements IoProcessor<S> {
- public void add(T session) {
+ public void add(S session) {
}
- public void flush(T session) {
+ public void flush(S session) {
if (scheduleFlush(session)) {
wakeup();
}
}
- public void remove(T session) {
+ public void remove(S session) {
getSessionRecycler().remove(session);
getListeners().fireSessionDestroyed(session);
}
- public void updateTrafficControl(T session) {
+ public void updateTrafficControl(S session) {
throw new UnsupportedOperationException();
}
@@ -361,7 +368,7 @@ public abstract class AbstractPollingCon
}
}
- private boolean scheduleFlush(T session) {
+ private boolean scheduleFlush(S session) {
// Set the schedule for flush flag if the session
// has not already be added to the flushingSessions
// queue
@@ -443,7 +450,7 @@ public abstract class AbstractPollingCon
if (isWritable(h)) {
for (IoSession session : getManagedSessions().values()) {
- scheduleFlush((T) session);
+ scheduleFlush((S) session);
}
}
} catch (Throwable t) {
@@ -474,7 +481,7 @@ public abstract class AbstractPollingCon
private void flushSessions(long currentTime) {
for (;;) {
- T session = flushingSessions.poll();
+ S session = flushingSessions.poll();
if (session == null) {
break;
@@ -496,7 +503,7 @@ public abstract class AbstractPollingCon
}
}
- private boolean flush(T session, long currentTime) throws Exception {
+ private boolean flush(S session, long currentTime) throws Exception {
// Clear OP_WRITE
setInterestedInWrite(session, false);
Modified: mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java?rev=1006194&r1=1006193&r2=1006194&view=diff
==============================================================================
--- mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java (original)
+++ mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java Sat Oct 9 17:06:30 2010
@@ -60,9 +60,10 @@ import org.slf4j.LoggerFactory;
* operation is possible.
*
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
+ *
+ * @param <S> the type of the {@link IoSession} this processor can handle
*/
-public abstract class AbstractPollingIoProcessor<T extends AbstractIoSession>
- implements IoProcessor<T> {
+public abstract class AbstractPollingIoProcessor<S extends AbstractIoSession> implements IoProcessor<S> {
/** A logger for this class */
private final static Logger LOG = LoggerFactory.getLogger(IoProcessor.class);
@@ -93,19 +94,19 @@ public abstract class AbstractPollingIoP
private final Executor executor;
/** A Session queue containing the newly created sessions */
- private final Queue<T> newSessions = new ConcurrentLinkedQueue<T>();
+ private final Queue<S> newSessions = new ConcurrentLinkedQueue<S>();
/** A queue used to store the sessions to be removed */
- private final Queue<T> removingSessions = new ConcurrentLinkedQueue<T>();
+ private final Queue<S> removingSessions = new ConcurrentLinkedQueue<S>();
/** A queue used to store the sessions to be flushed */
- private final Queue<T> flushingSessions = new ConcurrentLinkedQueue<T>();
+ private final Queue<S> flushingSessions = new ConcurrentLinkedQueue<S>();
/**
* A queue used to store the sessions which have a trafficControl to be
* updated
*/
- private final Queue<T> trafficControllingSessions = new ConcurrentLinkedQueue<T>();
+ private final Queue<S> trafficControllingSessions = new ConcurrentLinkedQueue<S>();
/** The processor thread : it handles the incoming messages */
private Processor processor;
@@ -254,14 +255,14 @@ public abstract class AbstractPollingIoP
*
* @return {@link Iterator} of {@link IoSession}
*/
- protected abstract Iterator<T> allSessions();
+ protected abstract Iterator<S> allSessions();
/**
* Get an {@link Iterator} for the list of {@link IoSession} found selected
* by the last call of {@link AbstractPollingIoProcessor#select(int)
* @return {@link Iterator} of {@link IoSession} read for I/Os operation
*/
- protected abstract Iterator<T> selectedSessions();
+ protected abstract Iterator<S> selectedSessions();
/**
* Get the state of a session (preparing, open, closed)
@@ -270,7 +271,7 @@ public abstract class AbstractPollingIoP
* the {@link IoSession} to inspect
* @return the state of the session
*/
- protected abstract SessionState getState(T session);
+ protected abstract SessionState getState(S session);
/**
* Is the session ready for writing
@@ -279,7 +280,7 @@ public abstract class AbstractPollingIoP
* the session queried
* @return true is ready, false if not ready
*/
- protected abstract boolean isWritable(T session);
+ protected abstract boolean isWritable(S session);
/**
* Is the session ready for reading
@@ -288,7 +289,7 @@ public abstract class AbstractPollingIoP
* the session queried
* @return true is ready, false if not ready
*/
- protected abstract boolean isReadable(T session);
+ protected abstract boolean isReadable(S session);
/**
* register a session for writing
@@ -298,7 +299,7 @@ public abstract class AbstractPollingIoP
* @param isInterested
* true for registering, false for removing
*/
- protected abstract void setInterestedInWrite(T session, boolean isInterested)
+ protected abstract void setInterestedInWrite(S session, boolean isInterested)
throws Exception;
/**
@@ -309,7 +310,7 @@ public abstract class AbstractPollingIoP
* @param isInterested
* true for registering, false for removing
*/
- protected abstract void setInterestedInRead(T session, boolean isInterested)
+ protected abstract void setInterestedInRead(S session, boolean isInterested)
throws Exception;
/**
@@ -319,7 +320,7 @@ public abstract class AbstractPollingIoP
* the session queried
* @return true is registered for reading
*/
- protected abstract boolean isInterestedInRead(T session);
+ protected abstract boolean isInterestedInRead(S session);
/**
* is this session registered for writing
@@ -328,7 +329,7 @@ public abstract class AbstractPollingIoP
* the session queried
* @return true is registered for writing
*/
- protected abstract boolean isInterestedInWrite(T session);
+ protected abstract boolean isInterestedInWrite(S session);
/**
* Initialize the polling of a session. Add it to the polling process.
@@ -336,7 +337,7 @@ public abstract class AbstractPollingIoP
* @param session the {@link IoSession} to add to the polling
* @throws Exception any exception thrown by the underlying system calls
*/
- protected abstract void init(T session) throws Exception;
+ protected abstract void init(S session) throws Exception;
/**
* Destroy the underlying client socket handle
@@ -346,7 +347,7 @@ public abstract class AbstractPollingIoP
* @throws Exception
* any exception thrown by the underlying system calls
*/
- protected abstract void destroy(T session) throws Exception;
+ protected abstract void destroy(S session) throws Exception;
/**
* Reads a sequence of bytes from a {@link IoSession} into the given
@@ -360,7 +361,7 @@ public abstract class AbstractPollingIoP
* @throws Exception
* any exception thrown by the underlying system calls
*/
- protected abstract int read(T session, IoBuffer buf) throws Exception;
+ protected abstract int read(S session, IoBuffer buf) throws Exception;
/**
* Write a sequence of bytes to a {@link IoSession}, means to be called when
@@ -377,7 +378,7 @@ public abstract class AbstractPollingIoP
* @throws Exception
* any exception thrown by the underlying system calls
*/
- protected abstract int write(T session, IoBuffer buf, int length)
+ protected abstract int write(S session, IoBuffer buf, int length)
throws Exception;
/**
@@ -396,13 +397,13 @@ public abstract class AbstractPollingIoP
* @throws Exception
* any exception thrown by the underlying system calls
*/
- protected abstract int transferFile(T session, FileRegion region, int length)
+ protected abstract int transferFile(S session, FileRegion region, int length)
throws Exception;
/**
* {@inheritDoc}
*/
- public final void add(T session) {
+ public final void add(S session) {
if (isDisposing()) {
throw new IllegalStateException("Already disposed.");
}
@@ -415,19 +416,19 @@ public abstract class AbstractPollingIoP
/**
* {@inheritDoc}
*/
- public final void remove(T session) {
+ public final void remove(S session) {
scheduleRemove(session);
startupProcessor();
}
- private void scheduleRemove(T session) {
+ private void scheduleRemove(S session) {
removingSessions.add(session);
}
/**
* {@inheritDoc}
*/
- public final void flush(T session) {
+ public final void flush(S session) {
// add the session to the queue if it's not already
// in the queue, then wake up the select()
if (session.setScheduledForFlush( true )) {
@@ -436,7 +437,7 @@ public abstract class AbstractPollingIoP
}
}
- private void scheduleFlush(T session) {
+ private void scheduleFlush(S session) {
// add the session to the queue if it's not already
// in the queue
if (session.setScheduledForFlush(true)) {
@@ -447,7 +448,7 @@ public abstract class AbstractPollingIoP
/**
* {@inheritDoc}
*/
- public final void updateTrafficMask(T session) {
+ public final void updateTrafficMask(S session) {
trafficControllingSessions.add(session);
wakeup();
}
@@ -470,27 +471,6 @@ public abstract class AbstractPollingIoP
}
/**
- * In the case we are using the java select() method, this method is used to
- * trash the buggy selector and create a new one, registring all the sockets
- * on it.
- *
- * @throws IOException
- * If we got an exception
- */
- abstract protected void registerNewSelector() throws IOException;
-
- /**
- * Check that the select() has not exited immediately just because of a
- * broken connection. In this case, this is a standard case, and we just
- * have to loop.
- *
- * @return true if a connection has been brutally closed.
- * @throws IOException
- * If we got an exception
- */
- abstract protected boolean isBrokenConnection() throws IOException;
-
- /**
* Loops over the new sessions blocking queue and returns the number of
* sessions which are effectively created
*
@@ -499,7 +479,7 @@ public abstract class AbstractPollingIoP
private int handleNewSessions() {
int addedSessions = 0;
- for (T session = newSessions.poll(); session != null; session = newSessions.poll()) {
+ for (S session = newSessions.poll(); session != null; session = newSessions.poll()) {
if (addNow(session)) {
// A new session has been created
addedSessions++;
@@ -518,7 +498,7 @@ public abstract class AbstractPollingIoP
* @param session The session to create
* @return true if the session has been registered
*/
- private boolean addNow(T session) {
+ private boolean addNow(S session) {
boolean registered = false;
try {
@@ -552,7 +532,7 @@ public abstract class AbstractPollingIoP
private int removeSessions() {
int removedSessions = 0;
- for (T session = removingSessions.poll();session != null;session = removingSessions.poll()) {
+ for (S session = removingSessions.poll(); session != null; session = removingSessions.poll()) {
SessionState state = getState(session);
// Now deal with the removal accordingly to the session's state
@@ -588,7 +568,7 @@ public abstract class AbstractPollingIoP
return removedSessions;
}
- private boolean removeNow(T session) {
+ private boolean removeNow(S session) {
clearWriteRequestQueue(session);
try {
@@ -605,7 +585,7 @@ public abstract class AbstractPollingIoP
return false;
}
- private void clearWriteRequestQueue(T session) {
+ private void clearWriteRequestQueue(S session) {
WriteRequestQueue writeRequestQueue = session.getWriteRequestQueue();
WriteRequest req;
@@ -652,8 +632,8 @@ public abstract class AbstractPollingIoP
}
private void process() throws Exception {
- for (Iterator<T> i = selectedSessions(); i.hasNext();) {
- T session = i.next();
+ for (Iterator<S> i = selectedSessions(); i.hasNext();) {
+ S session = i.next();
process(session);
i.remove();
}
@@ -662,7 +642,7 @@ public abstract class AbstractPollingIoP
/**
* Deal with session ready for the read or write operations, or both.
*/
- private void process(T session) {
+ private void process(S session) {
// Process Reads
if (isReadable(session) && !session.isReadSuspended()) {
read(session);
@@ -677,7 +657,7 @@ public abstract class AbstractPollingIoP
}
}
- private void read(T session) {
+ private void read(S session) {
IoSessionConfig config = session.getConfig();
int bufferSize = config.getReadBufferSize();
IoBuffer buf = IoBuffer.allocate(bufferSize);
@@ -780,7 +760,7 @@ public abstract class AbstractPollingIoP
}
do {
- T session = flushingSessions.poll(); // the same one with firstSession
+ S session = flushingSessions.poll(); // the same one with firstSession
if (session == null) {
// Just in case ... It should not happen.
@@ -829,7 +809,7 @@ public abstract class AbstractPollingIoP
} while (!flushingSessions.isEmpty());
}
- private boolean flushNow(T session, long currentTime) {
+ private boolean flushNow(S session, long currentTime) {
if (!session.isConnected()) {
scheduleRemove(session);
return false;
@@ -932,7 +912,7 @@ public abstract class AbstractPollingIoP
return true;
}
- private int writeBuffer(T session, WriteRequest req,
+ private int writeBuffer(S session, WriteRequest req,
boolean hasFragmentation, int maxLength, long currentTime)
throws Exception {
IoBuffer buf = (IoBuffer) req.getMessage();
@@ -965,7 +945,7 @@ public abstract class AbstractPollingIoP
return localWrittenBytes;
}
- private int writeFile(T session, WriteRequest req,
+ private int writeFile(S session, WriteRequest req,
boolean hasFragmentation, int maxLength, long currentTime)
throws Exception {
int localWrittenBytes;
@@ -997,7 +977,7 @@ public abstract class AbstractPollingIoP
return localWrittenBytes;
}
- private void fireMessageSent(T session, WriteRequest req) {
+ private void fireMessageSent(S session, WriteRequest req) {
session.setCurrentWriteRequest(null);
IoFilterChain filterChain = session.getFilterChain();
filterChain.fireMessageSent(req);
@@ -1010,7 +990,7 @@ public abstract class AbstractPollingIoP
int queueSize = trafficControllingSessions.size();
while (queueSize > 0) {
- T session = trafficControllingSessions.poll();
+ S session = trafficControllingSessions.poll();
if (session == null) {
// We are done with this queue.
@@ -1051,7 +1031,7 @@ public abstract class AbstractPollingIoP
/**
* {@inheritDoc}
*/
- public void updateTrafficControl(T session) {
+ public void updateTrafficControl(S session) {
//
try {
setInterestedInRead(session, !session.isReadSuspended());
@@ -1087,45 +1067,7 @@ public abstract class AbstractPollingIoP
// idle session when we get out of the select every
// second. (note : this is a hack to avoid creating
// a dedicated thread).
- long t0 = System.currentTimeMillis();
int selected = select(SELECT_TIMEOUT);
- long t1 = System.currentTimeMillis();
- long delta = (t1 - t0);
-
- if ((selected == 0) && !wakeupCalled.get() && (delta < 100)) {
- // Last chance : the select() may have been
- // interrupted because we have had an closed channel.
- if (isBrokenConnection()) {
- LOG.warn("Broken connection");
-
- // we can reselect immediately
- // set back the flag to false
- wakeupCalled.getAndSet(false);
-
- continue;
- } else {
- LOG.warn("Create a new selector. Selected is 0, delta = "
- + (t1 - t0));
- // Ok, we are hit by the nasty epoll
- // spinning.
- // Basically, there is a race condition
- // which causes a closing file descriptor not to be
- // considered as available as a selected channel, but
- // it stopped the select. The next time we will
- // call select(), it will exit immediately for the same
- // reason, and do so forever, consuming 100%
- // CPU.
- // We have to destroy the selector, and
- // register all the socket on a new one.
- registerNewSelector();
- }
-
- // Set back the flag to false
- wakeupCalled.getAndSet(false);
-
- // and continue the loop
- continue;
- }
// Manage newly created session first
nSessions += handleNewSessions();
@@ -1163,7 +1105,7 @@ public abstract class AbstractPollingIoP
// Disconnect all sessions immediately if disposal has been
// requested so that we exit this loop eventually.
if (isDisposing()) {
- for (Iterator<T> i = allSessions(); i.hasNext();) {
+ for (Iterator<S> i = allSessions(); i.hasNext();) {
scheduleRemove(i.next());
}
Modified: mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/service/IoProcessor.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/service/IoProcessor.java?rev=1006194&r1=1006193&r2=1006194&view=diff
==============================================================================
--- mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/service/IoProcessor.java (original)
+++ mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/service/IoProcessor.java Sat Oct 9 17:06:30 2010
@@ -29,9 +29,9 @@ import org.apache.mina.core.session.IoSe
*
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
*
- * @param <T> the type of the {@link IoSession} this processor can handle
+ * @param <S> the type of the {@link IoSession} this processor can handle
*/
-public interface IoProcessor<T extends IoSession> {
+public interface IoProcessor<S extends IoSession> {
/**
* Returns <tt>true</tt> if and if only {@link #dispose()} method has
@@ -59,20 +59,20 @@ public interface IoProcessor<T extends I
* the I/O processor starts to perform any I/O operations related
* with the {@code session}.
*/
- void add(T session);
+ void add(S session);
/**
* Flushes the internal write request queue of the specified
* {@code session}.
*/
- void flush(T session);
+ void flush(S session);
/**
* Controls the traffic of the specified {@code session} depending of the
* {@link IoSession#isReadSuspended()} and {@link IoSession#isWriteSuspended()}
* flags
*/
- void updateTrafficControl(T session);
+ void updateTrafficControl(S session);
/**
* Removes and closes the specified {@code session} from the I/O
@@ -80,5 +80,5 @@ public interface IoProcessor<T extends I
* associated with the {@code session} and releases any other related
* resources.
*/
- void remove(T session);
+ void remove(S session);
}
Modified: mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/service/SimpleIoProcessorPool.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/service/SimpleIoProcessorPool.java?rev=1006194&r1=1006193&r2=1006194&view=diff
==============================================================================
--- mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/service/SimpleIoProcessorPool.java (original)
+++ mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/core/service/SimpleIoProcessorPool.java Sat Oct 9 17:06:30 2010
@@ -70,10 +70,10 @@ import org.slf4j.LoggerFactory;
*
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
*
- * @param <T> the type of the {@link IoSession} to be managed by the specified
+ * @param <S> the type of the {@link IoSession} to be managed by the specified
* {@link IoProcessor}.
*/
-public class SimpleIoProcessorPool<T extends AbstractIoSession> implements IoProcessor<T> {
+public class SimpleIoProcessorPool<S extends AbstractIoSession> implements IoProcessor<S> {
/** A logger for this class */
private final static Logger LOGGER = LoggerFactory.getLogger(SimpleIoProcessorPool.class);
@@ -84,7 +84,7 @@ public class SimpleIoProcessorPool<T ext
private static final AttributeKey PROCESSOR = new AttributeKey( SimpleIoProcessorPool.class, "processor");
/** The pool table */
- private final IoProcessor<T>[] pool;
+ private final IoProcessor<S>[] pool;
/** The contained which is passed to the IoProcessor when they are created */
private final Executor executor;
@@ -107,7 +107,7 @@ public class SimpleIoProcessorPool<T ext
*
* @param processorType The type of IoProcessor to use
*/
- public SimpleIoProcessorPool(Class<? extends IoProcessor<T>> processorType) {
+ public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType) {
this(processorType, null, DEFAULT_SIZE);
}
@@ -118,7 +118,7 @@ public class SimpleIoProcessorPool<T ext
* @param processorType The type of IoProcessor to use
* @param size The number of IoProcessor in the pool
*/
- public SimpleIoProcessorPool(Class<? extends IoProcessor<T>> processorType, int size) {
+ public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, int size) {
this(processorType, null, size);
}
@@ -128,7 +128,7 @@ public class SimpleIoProcessorPool<T ext
* @param processorType The type of IoProcessor to use
* @param executor The {@link Executor}
*/
- public SimpleIoProcessorPool(Class<? extends IoProcessor<T>> processorType, Executor executor) {
+ public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, Executor executor) {
this(processorType, executor, DEFAULT_SIZE);
}
@@ -140,7 +140,7 @@ public class SimpleIoProcessorPool<T ext
* @param size The number of IoProcessor in the pool
*/
@SuppressWarnings("unchecked")
- public SimpleIoProcessorPool(Class<? extends IoProcessor<T>> processorType,
+ public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType,
Executor executor, int size) {
if (processorType == null) {
throw new IllegalArgumentException("processorType");
@@ -163,7 +163,7 @@ public class SimpleIoProcessorPool<T ext
pool = new IoProcessor[size];
boolean success = false;
- Constructor<? extends IoProcessor<T>> processorConstructor = null;
+ Constructor<? extends IoProcessor<S>> processorConstructor = null;
boolean usesExecutorArg = true;
try {
@@ -233,28 +233,28 @@ public class SimpleIoProcessorPool<T ext
/**
* {@inheritDoc}
*/
- public final void add(T session) {
+ public final void add(S session) {
getProcessor(session).add(session);
}
/**
* {@inheritDoc}
*/
- public final void flush(T session) {
+ public final void flush(S session) {
getProcessor(session).flush(session);
}
/**
* {@inheritDoc}
*/
- public final void remove(T session) {
+ public final void remove(S session) {
getProcessor(session).remove(session);
}
/**
* {@inheritDoc}
*/
- public final void updateTrafficControl(T session) {
+ public final void updateTrafficControl(S session) {
getProcessor(session).updateTrafficControl(session);
}
@@ -316,8 +316,8 @@ public class SimpleIoProcessorPool<T ext
* the session's attributes, pick a new processor and stores it.
*/
@SuppressWarnings("unchecked")
- private IoProcessor<T> getProcessor(T session) {
- IoProcessor<T> processor = (IoProcessor<T>) session.getAttribute(PROCESSOR);
+ private IoProcessor<S> getProcessor(S session) {
+ IoProcessor<S> processor = (IoProcessor<S>) session.getAttribute(PROCESSOR);
if (processor == null) {
processor = nextProcessor(session);
@@ -330,7 +330,7 @@ public class SimpleIoProcessorPool<T ext
/**
* Get a new Processor in the pool, using a round-robin algorithm.
*/
- private IoProcessor<T> nextProcessor(T session) {
+ private IoProcessor<S> nextProcessor(S session) {
if (disposed) {
throw new IllegalStateException(
"A disposed processor cannot be accessed.");
Modified: mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/transport/socket/nio/NioProcessor.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/transport/socket/nio/NioProcessor.java?rev=1006194&r1=1006193&r2=1006194&view=diff
==============================================================================
--- mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/transport/socket/nio/NioProcessor.java (original)
+++ mina/branches/2.0.1/mina-core/src/main/java/org/apache/mina/transport/socket/nio/NioProcessor.java Sat Oct 9 17:06:30 2010
@@ -21,11 +21,9 @@ package org.apache.mina.transport.socket
import java.io.IOException;
import java.nio.channels.ByteChannel;
-import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
-import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.Executor;
@@ -118,67 +116,6 @@ public final class NioProcessor extends
}
/**
- * In the case we are using the java select() method, this method is used to
- * trash the buggy selector and create a new one, registering all the
- * sockets on it.
- */
- protected void registerNewSelector() throws IOException {
- synchronized (selector) {
- Set<SelectionKey> keys = selector.keys();
-
- // Open a new selector
- Selector newSelector = Selector.open();
-
- // Loop on all the registered keys, and register them on the new selector
- for (SelectionKey key : keys) {
- SelectableChannel ch = key.channel();
-
- // Don't forget to attache the session, and back !
- NioSession session = (NioSession)key.attachment();
- SelectionKey newKey = ch.register(newSelector, key.interestOps(), session);
- session.setSelectionKey( newKey );
- }
-
- // Now we can close the old selector and switch it
- selector.close();
- selector = newSelector;
- }
- }
-
- /**
- * {@inheritDoc}
- */
- protected boolean isBrokenConnection() throws IOException {
- // A flag set to true if we find a broken session
- boolean brokenSession = false;
-
- synchronized (selector) {
- // Get the selector keys
- Set<SelectionKey> keys = selector.keys();
-
- // Loop on all the keys to see if one of them
- // has a closed channel
- for (SelectionKey key : keys) {
- SelectableChannel channel = key.channel();
-
- if ((((channel instanceof DatagramChannel) && ((DatagramChannel) channel)
- .isConnected()))
- || ((channel instanceof SocketChannel) && ((SocketChannel) channel)
- .isConnected())) {
- // The channel is not connected anymore. Cancel
- // the associated key then.
- key.cancel();
-
- // Set the flag to true to avoid a selector switch
- brokenSession = true;
- }
- }
- }
-
- return brokenSession;
- }
-
- /**
* {@inheritDoc}
*/
@Override
Modified: mina/branches/2.0.1/mina-core/src/test/java/org/apache/mina/transport/socket/nio/PollingIoProcessorTest.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0.1/mina-core/src/test/java/org/apache/mina/transport/socket/nio/PollingIoProcessorTest.java?rev=1006194&r1=1006193&r2=1006194&view=diff
==============================================================================
--- mina/branches/2.0.1/mina-core/src/test/java/org/apache/mina/transport/socket/nio/PollingIoProcessorTest.java (original)
+++ mina/branches/2.0.1/mina-core/src/test/java/org/apache/mina/transport/socket/nio/PollingIoProcessorTest.java Sat Oct 9 17:06:30 2010
@@ -21,7 +21,6 @@ package org.apache.mina.transport.socket
import static org.junit.Assert.assertNotNull;
-import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.NoRouteToHostException;
import java.util.Iterator;
@@ -158,17 +157,6 @@ public class PollingIoProcessorTest {
throw new NoRouteToHostException(
"No Route To Host Test");
}
-
- @Override
- protected boolean isBrokenConnection() throws IOException {
- return proc.isBrokenConnection();
- }
-
- @Override
- protected void registerNewSelector() throws IOException {
- proc.registerNewSelector();
- }
-
});
connector.setHandler(new IoHandlerAdapter());
Modified: mina/branches/2.0.1/mina-transport-apr/src/main/java/org/apache/mina/transport/socket/apr/AprIoProcessor.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0.1/mina-transport-apr/src/main/java/org/apache/mina/transport/socket/apr/AprIoProcessor.java?rev=1006194&r1=1006193&r2=1006194&view=diff
==============================================================================
--- mina/branches/2.0.1/mina-transport-apr/src/main/java/org/apache/mina/transport/socket/apr/AprIoProcessor.java (original)
+++ mina/branches/2.0.1/mina-transport-apr/src/main/java/org/apache/mina/transport/socket/apr/AprIoProcessor.java Sat Oct 9 17:06:30 2010
@@ -466,19 +466,4 @@ public final class AprIoProcessor extend
private void throwException(int code) throws IOException {
throw new IOException(org.apache.tomcat.jni.Error.strerror(-code) + " (code: " + code + ")");
}
-
- /**
- * {@inheritDoc}
- */
- protected void registerNewSelector() {
- // Do nothing
- }
-
- /**
- * {@inheritDoc}
- */
- protected boolean isBrokenConnection() throws IOException {
- // Here, we assume that this is the case.
- return true;
- }
}
\ No newline at end of file