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 2012/09/28 16:50:20 UTC
svn commit: r1391490 - in /mina/branches/2.0:
mina-core/src/main/java/org/apache/mina/core/filterchain/
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/...
Author: elecharny
Date: Fri Sep 28 14:50:19 2012
New Revision: 1391490
URL: http://svn.apache.org/viewvc?rev=1391490&view=rev
Log:
o Added the write(S session, WriteRequest writeRequest) method in the IoProcessor interface. It will allow a direct write into the socket if this socket is ready, and speedup the transmission of data
o Added some missing Javadoc
Modified:
mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/filterchain/DefaultIoFilterChain.java
mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingConnectionlessIoAcceptor.java
mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java
mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/service/IoProcessor.java
mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/service/SimpleIoProcessorPool.java
mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/session/AbstractIoSession.java
mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/session/DummySession.java
mina/branches/2.0/mina-core/src/main/java/org/apache/mina/transport/vmpipe/VmPipeFilterChain.java
mina/branches/2.0/mina-transport-serial/src/main/java/org/apache/mina/transport/serial/SerialSessionImpl.java
Modified: mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/filterchain/DefaultIoFilterChain.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/filterchain/DefaultIoFilterChain.java?rev=1391490&r1=1391489&r2=1391490&view=diff
==============================================================================
--- mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/filterchain/DefaultIoFilterChain.java (original)
+++ mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/filterchain/DefaultIoFilterChain.java Fri Sep 28 14:50:19 2012
@@ -33,6 +33,7 @@ import org.apache.mina.core.session.Attr
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.write.WriteRequest;
+import org.apache.mina.core.write.WriteRequestQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,8 +51,8 @@ public class DefaultIoFilterChain implem
* attribute and notifies the future when {@link #fireSessionCreated()}
* or {@link #fireExceptionCaught(Throwable)} is invoked.
*/
- public static final AttributeKey SESSION_CREATED_FUTURE = new AttributeKey(
- DefaultIoFilterChain.class, "connectFuture");
+ public static final AttributeKey SESSION_CREATED_FUTURE = new AttributeKey(DefaultIoFilterChain.class,
+ "connectFuture");
/** The associated session */
private final AbstractIoSession session;
@@ -67,7 +68,6 @@ public class DefaultIoFilterChain implem
/** The logger for this class */
private final static Logger LOGGER = LoggerFactory.getLogger(DefaultIoFilterChain.class);
-
/**
* Create a new default chain, associated with a session. It will only contain a
* HeadFilter and a TailFilter.
@@ -174,15 +174,13 @@ public class DefaultIoFilterChain implem
register(tail.prevEntry, name, filter);
}
- public synchronized void addBefore(String baseName, String name,
- IoFilter filter) {
+ public synchronized void addBefore(String baseName, String name, IoFilter filter) {
EntryImpl baseEntry = checkOldName(baseName);
checkAddable(name);
register(baseEntry.prevEntry, name, filter);
}
- public synchronized void addAfter(String baseName, String name,
- IoFilter filter) {
+ public synchronized void addAfter(String baseName, String name, IoFilter filter) {
EntryImpl baseEntry = checkOldName(baseName);
checkAddable(name);
register(baseEntry, name, filter);
@@ -203,8 +201,7 @@ public class DefaultIoFilterChain implem
}
e = e.nextEntry;
}
- throw new IllegalArgumentException("Filter not found: "
- + filter.getClass().getName());
+ throw new IllegalArgumentException("Filter not found: " + filter.getClass().getName());
}
public synchronized IoFilter remove(Class<? extends IoFilter> filterType) {
@@ -217,8 +214,7 @@ public class DefaultIoFilterChain implem
}
e = e.nextEntry;
}
- throw new IllegalArgumentException("Filter not found: "
- + filterType.getName());
+ throw new IllegalArgumentException("Filter not found: " + filterType.getName());
}
public synchronized IoFilter replace(String name, IoFilter newFilter) {
@@ -237,12 +233,10 @@ public class DefaultIoFilterChain implem
}
e = e.nextEntry;
}
- throw new IllegalArgumentException("Filter not found: "
- + oldFilter.getClass().getName());
+ throw new IllegalArgumentException("Filter not found: " + oldFilter.getClass().getName());
}
- public synchronized IoFilter replace(
- Class<? extends IoFilter> oldFilterType, IoFilter newFilter) {
+ public synchronized IoFilter replace(Class<? extends IoFilter> oldFilterType, IoFilter newFilter) {
EntryImpl e = head.nextEntry;
while (e != tail) {
if (oldFilterType.isAssignableFrom(e.getFilter().getClass())) {
@@ -252,32 +246,27 @@ public class DefaultIoFilterChain implem
}
e = e.nextEntry;
}
- throw new IllegalArgumentException("Filter not found: "
- + oldFilterType.getName());
+ throw new IllegalArgumentException("Filter not found: " + oldFilterType.getName());
}
public synchronized void clear() throws Exception {
- List<IoFilterChain.Entry> l = new ArrayList<IoFilterChain.Entry>(
- name2entry.values());
+ List<IoFilterChain.Entry> l = new ArrayList<IoFilterChain.Entry>(name2entry.values());
for (IoFilterChain.Entry entry : l) {
try {
deregister((EntryImpl) entry);
} catch (Exception e) {
- throw new IoFilterLifeCycleException("clear(): "
- + entry.getName() + " in " + getSession(), e);
+ throw new IoFilterLifeCycleException("clear(): " + entry.getName() + " in " + getSession(), e);
}
}
}
private void register(EntryImpl prevEntry, String name, IoFilter filter) {
- EntryImpl newEntry = new EntryImpl(prevEntry, prevEntry.nextEntry,
- name, filter);
+ EntryImpl newEntry = new EntryImpl(prevEntry, prevEntry.nextEntry, name, filter);
try {
filter.onPreAdd(this, name, newEntry.getNextFilter());
} catch (Exception e) {
- throw new IoFilterLifeCycleException("onPreAdd(): " + name + ':'
- + filter + " in " + getSession(), e);
+ throw new IoFilterLifeCycleException("onPreAdd(): " + name + ':' + filter + " in " + getSession(), e);
}
prevEntry.nextEntry.prevEntry = newEntry;
@@ -288,8 +277,7 @@ public class DefaultIoFilterChain implem
filter.onPostAdd(this, name, newEntry.getNextFilter());
} catch (Exception e) {
deregister0(newEntry);
- throw new IoFilterLifeCycleException("onPostAdd(): " + name + ':'
- + filter + " in " + getSession(), e);
+ throw new IoFilterLifeCycleException("onPostAdd(): " + name + ':' + filter + " in " + getSession(), e);
}
}
@@ -299,8 +287,8 @@ public class DefaultIoFilterChain implem
try {
filter.onPreRemove(this, entry.getName(), entry.getNextFilter());
} catch (Exception e) {
- throw new IoFilterLifeCycleException("onPreRemove(): "
- + entry.getName() + ':' + filter + " in " + getSession(), e);
+ throw new IoFilterLifeCycleException("onPreRemove(): " + entry.getName() + ':' + filter + " in "
+ + getSession(), e);
}
deregister0(entry);
@@ -308,8 +296,8 @@ public class DefaultIoFilterChain implem
try {
filter.onPostRemove(this, entry.getName(), entry.getNextFilter());
} catch (Exception e) {
- throw new IoFilterLifeCycleException("onPostRemove(): "
- + entry.getName() + ':' + filter + " in " + getSession(), e);
+ throw new IoFilterLifeCycleException("onPostRemove(): " + entry.getName() + ':' + filter + " in "
+ + getSession(), e);
}
}
@@ -340,8 +328,7 @@ public class DefaultIoFilterChain implem
*/
private void checkAddable(String name) {
if (name2entry.containsKey(name)) {
- throw new IllegalArgumentException(
- "Other filter is using the same name '" + name + "'");
+ throw new IllegalArgumentException("Other filter is using the same name '" + name + "'");
}
}
@@ -404,13 +391,11 @@ public class DefaultIoFilterChain implem
callNextSessionIdle(head, session, status);
}
- private void callNextSessionIdle(Entry entry, IoSession session,
- IdleStatus status) {
+ private void callNextSessionIdle(Entry entry, IoSession session, IdleStatus status) {
try {
IoFilter filter = entry.getFilter();
NextFilter nextFilter = entry.getNextFilter();
- filter.sessionIdle(nextFilter, session,
- status);
+ filter.sessionIdle(nextFilter, session, status);
} catch (Throwable e) {
fireExceptionCaught(e);
}
@@ -418,21 +403,18 @@ public class DefaultIoFilterChain implem
public void fireMessageReceived(Object message) {
if (message instanceof IoBuffer) {
- session.increaseReadBytes(((IoBuffer) message).remaining(), System
- .currentTimeMillis());
+ session.increaseReadBytes(((IoBuffer) message).remaining(), System.currentTimeMillis());
}
Entry head = this.head;
callNextMessageReceived(head, session, message);
}
- private void callNextMessageReceived(Entry entry, IoSession session,
- Object message) {
+ private void callNextMessageReceived(Entry entry, IoSession session, Object message) {
try {
IoFilter filter = entry.getFilter();
NextFilter nextFilter = entry.getNextFilter();
- filter.messageReceived(nextFilter, session,
- message);
+ filter.messageReceived(nextFilter, session, message);
} catch (Throwable e) {
fireExceptionCaught(e);
}
@@ -448,19 +430,17 @@ public class DefaultIoFilterChain implem
}
Entry head = this.head;
-
+
if (!request.isEncoded()) {
callNextMessageSent(head, session, request);
}
}
- private void callNextMessageSent(Entry entry, IoSession session,
- WriteRequest writeRequest) {
+ private void callNextMessageSent(Entry entry, IoSession session, WriteRequest writeRequest) {
try {
IoFilter filter = entry.getFilter();
NextFilter nextFilter = entry.getNextFilter();
- filter.messageSent(nextFilter, session,
- writeRequest);
+ filter.messageSent(nextFilter, session, writeRequest);
} catch (Throwable e) {
fireExceptionCaught(e);
}
@@ -471,22 +451,16 @@ public class DefaultIoFilterChain implem
callNextExceptionCaught(head, session, cause);
}
- private void callNextExceptionCaught(Entry entry, IoSession session,
- Throwable cause) {
+ private void callNextExceptionCaught(Entry entry, IoSession session, Throwable cause) {
// Notify the related future.
- ConnectFuture future = (ConnectFuture) session
- .removeAttribute(SESSION_CREATED_FUTURE);
+ ConnectFuture future = (ConnectFuture) session.removeAttribute(SESSION_CREATED_FUTURE);
if (future == null) {
try {
IoFilter filter = entry.getFilter();
NextFilter nextFilter = entry.getNextFilter();
- filter.exceptionCaught(nextFilter,
- session, cause);
+ filter.exceptionCaught(nextFilter, session, cause);
} catch (Throwable e) {
- LOGGER
- .warn(
- "Unexpected exception from exceptionCaught handler.",
- e);
+ LOGGER.warn("Unexpected exception from exceptionCaught handler.", e);
}
} else {
// Please note that this place is not the only place that
@@ -501,8 +475,7 @@ public class DefaultIoFilterChain implem
callPreviousFilterWrite(tail, session, writeRequest);
}
- private void callPreviousFilterWrite(Entry entry, IoSession session,
- WriteRequest writeRequest) {
+ private void callPreviousFilterWrite(Entry entry, IoSession session, WriteRequest writeRequest) {
try {
IoFilter filter = entry.getFilter();
NextFilter nextFilter = entry.getNextFilter();
@@ -597,8 +570,7 @@ public class DefaultIoFilterChain implem
private class HeadFilter extends IoFilterAdapter {
@SuppressWarnings("unchecked")
@Override
- public void filterWrite(NextFilter nextFilter, IoSession session,
- WriteRequest writeRequest) throws Exception {
+ public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
AbstractIoSession s = (AbstractIoSession) session;
@@ -610,6 +582,7 @@ public class DefaultIoFilterChain implem
// the buffer will be specified with messageSent event.
buffer.mark();
int remaining = buffer.remaining();
+
if (remaining == 0) {
// Zero-sized buffer means the internal message
// delimiter.
@@ -621,30 +594,36 @@ public class DefaultIoFilterChain implem
s.increaseScheduledWriteMessages();
}
- s.getWriteRequestQueue().offer(s, writeRequest);
+ WriteRequestQueue writeRequestQueue = s.getWriteRequestQueue();
+
if (!s.isWriteSuspended()) {
- s.getProcessor().flush(s);
+ if (writeRequestQueue.size() == 0) {
+ // We can write directly the message
+ s.getProcessor().write(s, writeRequest);
+ } else {
+ s.getWriteRequestQueue().offer(s, writeRequest);
+ s.getProcessor().flush(s);
+ }
+ } else {
+ s.getWriteRequestQueue().offer(s, writeRequest);
}
}
@SuppressWarnings("unchecked")
@Override
- public void filterClose(NextFilter nextFilter, IoSession session)
- throws Exception {
- ((AbstractIoSession) session).getProcessor().remove(((AbstractIoSession) session));
+ public void filterClose(NextFilter nextFilter, IoSession session) throws Exception {
+ ((AbstractIoSession) session).getProcessor().remove(session);
}
}
private static class TailFilter extends IoFilterAdapter {
@Override
- public void sessionCreated(NextFilter nextFilter, IoSession session)
- throws Exception {
+ public void sessionCreated(NextFilter nextFilter, IoSession session) throws Exception {
try {
session.getHandler().sessionCreated(session);
} finally {
// Notify the related future.
- ConnectFuture future = (ConnectFuture) session
- .removeAttribute(SESSION_CREATED_FUTURE);
+ ConnectFuture future = (ConnectFuture) session.removeAttribute(SESSION_CREATED_FUTURE);
if (future != null) {
future.setSession(session);
}
@@ -652,14 +631,12 @@ public class DefaultIoFilterChain implem
}
@Override
- public void sessionOpened(NextFilter nextFilter, IoSession session)
- throws Exception {
+ public void sessionOpened(NextFilter nextFilter, IoSession session) throws Exception {
session.getHandler().sessionOpened(session);
}
@Override
- public void sessionClosed(NextFilter nextFilter, IoSession session)
- throws Exception {
+ public void sessionClosed(NextFilter nextFilter, IoSession session) throws Exception {
AbstractIoSession s = (AbstractIoSession) session;
try {
s.getHandler().sessionClosed(session);
@@ -684,14 +661,12 @@ public class DefaultIoFilterChain implem
}
@Override
- public void sessionIdle(NextFilter nextFilter, IoSession session,
- IdleStatus status) throws Exception {
+ public void sessionIdle(NextFilter nextFilter, IoSession session, IdleStatus status) throws Exception {
session.getHandler().sessionIdle(session, status);
}
@Override
- public void exceptionCaught(NextFilter nextFilter, IoSession session,
- Throwable cause) throws Exception {
+ public void exceptionCaught(NextFilter nextFilter, IoSession session, Throwable cause) throws Exception {
AbstractIoSession s = (AbstractIoSession) session;
try {
s.getHandler().exceptionCaught(s, cause);
@@ -703,8 +678,7 @@ public class DefaultIoFilterChain implem
}
@Override
- public void messageReceived(NextFilter nextFilter, IoSession session,
- Object message) throws Exception {
+ public void messageReceived(NextFilter nextFilter, IoSession session, Object message) throws Exception {
AbstractIoSession s = (AbstractIoSession) session;
if (!(message instanceof IoBuffer)) {
s.increaseReadMessages(System.currentTimeMillis());
@@ -722,21 +696,17 @@ public class DefaultIoFilterChain implem
}
@Override
- public void messageSent(NextFilter nextFilter, IoSession session,
- WriteRequest writeRequest) throws Exception {
- session.getHandler()
- .messageSent(session, writeRequest.getMessage());
+ public void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
+ session.getHandler().messageSent(session, writeRequest.getMessage());
}
@Override
- public void filterWrite(NextFilter nextFilter, IoSession session,
- WriteRequest writeRequest) throws Exception {
+ public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
nextFilter.filterWrite(session, writeRequest);
}
@Override
- public void filterClose(NextFilter nextFilter, IoSession session)
- throws Exception {
+ public void filterClose(NextFilter nextFilter, IoSession session) throws Exception {
nextFilter.filterClose(session);
}
}
@@ -752,8 +722,7 @@ public class DefaultIoFilterChain implem
private final NextFilter nextFilter;
- private EntryImpl(EntryImpl prevEntry, EntryImpl nextEntry,
- String name, IoFilter filter) {
+ private EntryImpl(EntryImpl prevEntry, EntryImpl nextEntry, String name, IoFilter filter) {
if (filter == null) {
throw new IllegalArgumentException("filter");
}
@@ -796,14 +765,12 @@ public class DefaultIoFilterChain implem
callNextMessageReceived(nextEntry, session, message);
}
- public void messageSent(IoSession session,
- WriteRequest writeRequest) {
+ public void messageSent(IoSession session, WriteRequest writeRequest) {
Entry nextEntry = EntryImpl.this.nextEntry;
callNextMessageSent(nextEntry, session, writeRequest);
}
- public void filterWrite(IoSession session,
- WriteRequest writeRequest) {
+ public void filterWrite(IoSession session, WriteRequest writeRequest) {
Entry nextEntry = EntryImpl.this.prevEntry;
callPreviousFilterWrite(nextEntry, session, writeRequest);
}
Modified: mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingConnectionlessIoAcceptor.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingConnectionlessIoAcceptor.java?rev=1391490&r1=1391489&r2=1391490&view=diff
==============================================================================
--- mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingConnectionlessIoAcceptor.java (original)
+++ mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingConnectionlessIoAcceptor.java Fri Sep 28 14:50:19 2012
@@ -327,6 +327,74 @@ public abstract class AbstractPollingCon
/**
* {@inheritDoc}
*/
+ public void write(S session, WriteRequest writeRequest) {
+ // We will try to write the message directly
+ long currentTime = System.currentTimeMillis();
+ final WriteRequestQueue writeRequestQueue = session.getWriteRequestQueue();
+ final int maxWrittenBytes = session.getConfig().getMaxReadBufferSize()
+ + (session.getConfig().getMaxReadBufferSize() >>> 1);
+
+ int writtenBytes = 0;
+
+ try {
+ for (;;) {
+ if (writeRequest == null) {
+ writeRequest = writeRequestQueue.poll(session);
+
+ if (writeRequest == null) {
+ setInterestedInWrite(session, false);
+ break;
+ }
+
+ session.setCurrentWriteRequest(writeRequest);
+ }
+
+ IoBuffer buf = (IoBuffer) writeRequest.getMessage();
+
+ if (buf.remaining() == 0) {
+ // Clear and fire event
+ session.setCurrentWriteRequest(null);
+ buf.reset();
+ session.getFilterChain().fireMessageSent(writeRequest);
+ continue;
+ }
+
+ SocketAddress destination = writeRequest.getDestination();
+
+ if (destination == null) {
+ destination = session.getRemoteAddress();
+ }
+
+ int localWrittenBytes = send(session, buf, destination);
+
+ if ((localWrittenBytes == 0) || (writtenBytes >= maxWrittenBytes)) {
+ // Kernel buffer is full or wrote too much
+ setInterestedInWrite(session, true);
+
+ session.getWriteRequestQueue().offer(session, writeRequest);
+ scheduleFlush(session);
+ } else {
+ setInterestedInWrite(session, false);
+
+ // Clear and fire event
+ session.setCurrentWriteRequest(null);
+ writtenBytes += localWrittenBytes;
+ buf.reset();
+ session.getFilterChain().fireMessageSent(writeRequest);
+
+ break;
+ }
+ }
+ } catch (Exception e) {
+ session.getFilterChain().fireExceptionCaught(e);
+ } finally {
+ session.increaseWrittenBytes(writtenBytes, currentTime);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
public void remove(S session) {
getSessionRecycler().remove(session);
getListeners().fireSessionDestroyed(session);
Modified: mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java?rev=1391490&r1=1391489&r2=1391490&view=diff
==============================================================================
--- mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java (original)
+++ mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java Fri Sep 28 14:50:19 2012
@@ -409,6 +409,19 @@ public abstract class AbstractPollingIoP
/**
* {@inheritDoc}
*/
+ public void write(S session, WriteRequest writeRequest) {
+ WriteRequestQueue writeRequestQueue = session.getWriteRequestQueue();
+
+ writeRequestQueue.offer(session, writeRequest);
+
+ if (!session.isWriteSuspended()) {
+ this.flush(session);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
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()
Modified: mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/service/IoProcessor.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/service/IoProcessor.java?rev=1391490&r1=1391489&r2=1391490&view=diff
==============================================================================
--- mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/service/IoProcessor.java (original)
+++ mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/service/IoProcessor.java Fri Sep 28 14:50:19 2012
@@ -20,6 +20,7 @@
package org.apache.mina.core.service;
import org.apache.mina.core.session.IoSession;
+import org.apache.mina.core.write.WriteRequest;
/**
* An internal interface to represent an 'I/O processor' that performs
@@ -39,13 +40,13 @@ public interface IoProcessor<S extends I
* even after all the related resources are released.
*/
boolean isDisposing();
-
+
/**
* Returns <tt>true</tt> if and if only all resources of this processor
* have been disposed.
*/
boolean isDisposed();
-
+
/**
* Releases any resources allocated by this processor. Please note that
* the resources might not be released as long as there are any sessions
@@ -53,21 +54,33 @@ public interface IoProcessor<S extends I
* immediately and release the related resources.
*/
void dispose();
-
+
/**
* Adds the specified {@code session} to the I/O processor so that
* the I/O processor starts to perform any I/O operations related
* with the {@code session}.
+ *
+ * @param session The added session
*/
void add(S session);
/**
* Flushes the internal write request queue of the specified
* {@code session}.
+ *
+ * @param session The session we want the message to be written
*/
void flush(S session);
/**
+ * Writes the WriteRequest for the specified {@code session}.
+ *
+ * @param session The session we want the message to be written
+ * @param writeRequest the WriteRequest to write
+ */
+ void write(S session, WriteRequest writeRequest);
+
+ /**
* Controls the traffic of the specified {@code session} depending of the
* {@link IoSession#isReadSuspended()} and {@link IoSession#isWriteSuspended()}
* flags
Modified: mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/service/SimpleIoProcessorPool.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/service/SimpleIoProcessorPool.java?rev=1391490&r1=1391489&r2=1391490&view=diff
==============================================================================
--- mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/service/SimpleIoProcessorPool.java (original)
+++ mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/service/SimpleIoProcessorPool.java Fri Sep 28 14:50:19 2012
@@ -30,6 +30,7 @@ import org.apache.mina.core.RuntimeIoExc
import org.apache.mina.core.session.AbstractIoSession;
import org.apache.mina.core.session.AttributeKey;
import org.apache.mina.core.session.IoSession;
+import org.apache.mina.core.write.WriteRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -83,7 +84,7 @@ public class SimpleIoProcessorPool<S ext
private static final int DEFAULT_SIZE = Runtime.getRuntime().availableProcessors() + 1;
/** A key used to store the processor pool in the session's Attributes */
- private static final AttributeKey PROCESSOR = new AttributeKey( SimpleIoProcessorPool.class, "processor");
+ private static final AttributeKey PROCESSOR = new AttributeKey(SimpleIoProcessorPool.class, "processor");
/** The pool table */
private final IoProcessor<S>[] pool;
@@ -142,24 +143,22 @@ public class SimpleIoProcessorPool<S ext
* @param size The number of IoProcessor in the pool
*/
@SuppressWarnings("unchecked")
- public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType,
- Executor executor, int size) {
+ public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, Executor executor, int size) {
if (processorType == null) {
throw new IllegalArgumentException("processorType");
}
if (size <= 0) {
- throw new IllegalArgumentException("size: " + size
- + " (expected: positive integer)");
+ throw new IllegalArgumentException("size: " + size + " (expected: positive integer)");
}
// Create the executor if none is provided
createdExecutor = (executor == null);
-
+
if (createdExecutor) {
this.executor = Executors.newCachedThreadPool();
// Set a default reject handler
- ((ThreadPoolExecutor)this.executor).setRejectedExecutionHandler( new ThreadPoolExecutor.CallerRunsPolicy() );
+ ((ThreadPoolExecutor) this.executor).setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
} else {
this.executor = executor;
}
@@ -198,17 +197,14 @@ public class SimpleIoProcessorPool<S ext
} catch (Exception e) {
String msg = "Failed to create a new instance of " + processorType.getName() + ":" + e.getMessage();
LOGGER.error(msg, e);
- throw new RuntimeIoException(msg , e);
+ throw new RuntimeIoException(msg, e);
}
if (processorConstructor == null) {
// Raise an exception if no proper constructor is found.
- String msg = String.valueOf(processorType)
- + " must have a public constructor with one "
- + ExecutorService.class.getSimpleName()
- + " parameter, a public constructor with one "
- + Executor.class.getSimpleName()
- + " parameter or a public default constructor.";
+ String msg = String.valueOf(processorType) + " must have a public constructor with one "
+ + ExecutorService.class.getSimpleName() + " parameter, a public constructor with one "
+ + Executor.class.getSimpleName() + " parameter or a public default constructor.";
LOGGER.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -225,7 +221,7 @@ public class SimpleIoProcessorPool<S ext
// Won't happen because it has been done previously
}
}
-
+
success = true;
} finally {
if (!success) {
@@ -251,6 +247,13 @@ public class SimpleIoProcessorPool<S ext
/**
* {@inheritDoc}
*/
+ public final void write(S session, WriteRequest writeRequest) {
+ getProcessor(session).write(session, writeRequest);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
public final void remove(S session) {
getProcessor(session).remove(session);
}
@@ -287,13 +290,13 @@ public class SimpleIoProcessorPool<S ext
synchronized (disposalLock) {
if (!disposing) {
disposing = true;
-
+
for (IoProcessor<S> ioProcessor : pool) {
if (ioProcessor == null) {
// Special case if the pool has not been initialized properly
continue;
}
-
+
if (ioProcessor.isDisposing()) {
continue;
}
@@ -322,18 +325,18 @@ public class SimpleIoProcessorPool<S ext
@SuppressWarnings("unchecked")
private IoProcessor<S> getProcessor(S session) {
IoProcessor<S> processor = (IoProcessor<S>) session.getAttribute(PROCESSOR);
-
+
if (processor == null) {
if (disposed || disposing) {
throw new IllegalStateException("A disposed processor cannot be accessed.");
}
processor = pool[Math.abs((int) session.getId()) % pool.length];
-
+
if (processor == null) {
throw new IllegalStateException("A disposed processor cannot be accessed.");
}
-
+
session.setAttributeIfAbsent(PROCESSOR, processor);
}
Modified: mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/session/AbstractIoSession.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/session/AbstractIoSession.java?rev=1391490&r1=1391489&r2=1391490&view=diff
==============================================================================
--- mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/session/AbstractIoSession.java (original)
+++ mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/session/AbstractIoSession.java Fri Sep 28 14:50:19 2012
@@ -284,11 +284,11 @@ public abstract class AbstractIoSession
* {@inheritDoc}
*/
public final CloseFuture close(boolean rightNow) {
- if ( !isClosing() ) {
+ if (!isClosing()) {
if (rightNow) {
return close();
}
-
+
return closeOnFlush();
} else {
return closeFuture;
@@ -449,7 +449,7 @@ public abstract class AbstractIoSession
// We can't send a message to a connected session if we don't have
// the remote address
- if (!getTransportMetadata().isConnectionless() && ( remoteAddress != null )) {
+ if (!getTransportMetadata().isConnectionless() && (remoteAddress != null)) {
throw new UnsupportedOperationException();
}
@@ -469,7 +469,7 @@ public abstract class AbstractIoSession
// TODO: remove this code as soon as we use InputStream
// instead of Object for the message.
try {
- if (( message instanceof IoBuffer ) && !((IoBuffer) message).hasRemaining()) {
+ if ((message instanceof IoBuffer) && !((IoBuffer) message).hasRemaining()) {
// Nothing to write : probably an error in the user code
throw new IllegalArgumentException("message is empty. Forgot to call flip()?");
} else if (message instanceof FileChannel) {
@@ -753,7 +753,7 @@ public abstract class AbstractIoSession
int interval = (int) (currentTime - lastThroughputCalculationTime);
long minInterval = getConfig().getThroughputCalculationIntervalInMillis();
- if (( minInterval == 0 ) || ( interval < minInterval )) {
+ if ((minInterval == 0) || (interval < minInterval)) {
if (!force) {
return;
}
@@ -1205,19 +1205,18 @@ public abstract class AbstractIoSession
try {
remote = String.valueOf(getRemoteAddress());
- } catch ( Throwable t ) {
+ } catch (Throwable t) {
remote = "Cannot get the remote address informations: " + t.getMessage();
}
try {
local = String.valueOf(getLocalAddress());
- } catch ( Throwable t ) {
+ } catch (Throwable t) {
local = "Cannot get the local address informations: " + t.getMessage();
}
if (getService() instanceof IoAcceptor) {
- return "(" + getIdAsString() + ": " + getServiceName() + ", server, " + remote + " => " + local
- + ')';
+ return "(" + getIdAsString() + ": " + getServiceName() + ", server, " + remote + " => " + local + ')';
}
return "(" + getIdAsString() + ": " + getServiceName() + ", client, " + local + " => " + remote + ')';
@@ -1288,19 +1287,19 @@ public abstract class AbstractIoSession
IdleStatus.BOTH_IDLE, Math.max(session.getLastIoTime(), session.getLastIdleTime(IdleStatus.BOTH_IDLE)));
notifyIdleSession0(session, currentTime, session.getConfig().getIdleTimeInMillis(IdleStatus.READER_IDLE),
- IdleStatus.READER_IDLE, Math.max(session.getLastReadTime(), session
- .getLastIdleTime(IdleStatus.READER_IDLE)));
+ IdleStatus.READER_IDLE,
+ Math.max(session.getLastReadTime(), session.getLastIdleTime(IdleStatus.READER_IDLE)));
notifyIdleSession0(session, currentTime, session.getConfig().getIdleTimeInMillis(IdleStatus.WRITER_IDLE),
- IdleStatus.WRITER_IDLE, Math.max(session.getLastWriteTime(), session
- .getLastIdleTime(IdleStatus.WRITER_IDLE)));
+ IdleStatus.WRITER_IDLE,
+ Math.max(session.getLastWriteTime(), session.getLastIdleTime(IdleStatus.WRITER_IDLE)));
notifyWriteTimeout(session, currentTime);
}
private static void notifyIdleSession0(IoSession session, long currentTime, long idleTime, IdleStatus status,
long lastIoTime) {
- if (( idleTime > 0 ) && ( lastIoTime != 0 ) && ( currentTime - lastIoTime >= idleTime )) {
+ if ((idleTime > 0) && (lastIoTime != 0) && (currentTime - lastIoTime >= idleTime)) {
session.getFilterChain().fireSessionIdle(status);
}
}
@@ -1308,7 +1307,7 @@ public abstract class AbstractIoSession
private static void notifyWriteTimeout(IoSession session, long currentTime) {
long writeTimeout = session.getConfig().getWriteTimeoutInMillis();
- if (( writeTimeout > 0 ) && ( currentTime - session.getLastWriteTime() >= writeTimeout )
+ if ((writeTimeout > 0) && (currentTime - session.getLastWriteTime() >= writeTimeout)
&& !session.getWriteRequestQueue().isEmpty(session)) {
WriteRequest request = session.getCurrentWriteRequest();
if (request != null) {
Modified: mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/session/DummySession.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/session/DummySession.java?rev=1391490&r1=1391489&r2=1391490&view=diff
==============================================================================
--- mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/session/DummySession.java (original)
+++ mina/branches/2.0/mina-core/src/main/java/org/apache/mina/core/session/DummySession.java Fri Sep 28 14:50:19 2012
@@ -37,6 +37,7 @@ import org.apache.mina.core.service.IoPr
import org.apache.mina.core.service.IoService;
import org.apache.mina.core.service.TransportMetadata;
import org.apache.mina.core.write.WriteRequest;
+import org.apache.mina.core.write.WriteRequestQueue;
/**
* A dummy {@link IoSession} for unit-testing or non-network-use of
@@ -52,10 +53,8 @@ import org.apache.mina.core.write.WriteR
*/
public class DummySession extends AbstractIoSession {
- private static final TransportMetadata TRANSPORT_METADATA =
- new DefaultTransportMetadata(
- "mina", "dummy", false, false,
- SocketAddress.class, IoSessionConfig.class, Object.class);
+ private static final TransportMetadata TRANSPORT_METADATA = new DefaultTransportMetadata("mina", "dummy", false,
+ false, SocketAddress.class, IoSessionConfig.class, Object.class);
private static final SocketAddress ANONYMOUS_ADDRESS = new SocketAddress() {
private static final long serialVersionUID = -496112902353454179L;
@@ -76,11 +75,15 @@ public class DummySession extends Abstra
};
private final IoFilterChain filterChain = new DefaultIoFilterChain(this);
+
private final IoProcessor<AbstractIoSession> processor;
private volatile IoHandler handler = new IoHandlerAdapter();
+
private volatile SocketAddress localAddress = ANONYMOUS_ADDRESS;
+
private volatile SocketAddress remoteAddress = ANONYMOUS_ADDRESS;
+
private volatile TransportMetadata transportMetadata = TRANSPORT_METADATA;
/**
@@ -90,41 +93,40 @@ public class DummySession extends Abstra
super(
// Initialize dummy service.
- new AbstractIoAcceptor(
- new AbstractIoSessionConfig() {
+ new AbstractIoAcceptor(new AbstractIoSessionConfig() {
@Override
protected void doSetAll(IoSessionConfig config) {
// Do nothing
}
- },
- new Executor() {
+ }, new Executor() {
public void execute(Runnable command) {
// Do nothing
}
}) {
- @Override
- protected Set<SocketAddress> bindInternal(List<? extends SocketAddress> localAddresses) throws Exception {
- throw new UnsupportedOperationException();
- }
+ @Override
+ protected Set<SocketAddress> bindInternal(List<? extends SocketAddress> localAddresses)
+ throws Exception {
+ throw new UnsupportedOperationException();
+ }
- @Override
- protected void unbind0(List<? extends SocketAddress> localAddresses) throws Exception {
- throw new UnsupportedOperationException();
- }
+ @Override
+ protected void unbind0(List<? extends SocketAddress> localAddresses) throws Exception {
+ throw new UnsupportedOperationException();
+ }
- public IoSession newSession(SocketAddress remoteAddress, SocketAddress localAddress) {
- throw new UnsupportedOperationException();
- }
+ public IoSession newSession(SocketAddress remoteAddress, SocketAddress localAddress) {
+ throw new UnsupportedOperationException();
+ }
- public TransportMetadata getTransportMetadata() {
- return TRANSPORT_METADATA;
- }
+ public TransportMetadata getTransportMetadata() {
+ return TRANSPORT_METADATA;
+ }
- @Override
- protected void dispose0() throws Exception {
- }
- } );
+ @Override
+ protected void dispose0() throws Exception {
+ }
+ });
processor = new IoProcessor<AbstractIoSession>() {
public void add(AbstractIoSession session) {
@@ -134,7 +136,7 @@ public class DummySession extends Abstra
public void flush(AbstractIoSession session) {
DummySession s = (DummySession) session;
WriteRequest req = s.getWriteRequestQueue().poll(session);
-
+
// Chek that the request is not null. If the session has been closed,
// we may not have any pending requests.
if (req != null) {
@@ -152,6 +154,19 @@ public class DummySession extends Abstra
}
}
+ /**
+ * {@inheritDoc}
+ */
+ public void write(AbstractIoSession session, WriteRequest writeRequest) {
+ WriteRequestQueue writeRequestQueue = session.getWriteRequestQueue();
+
+ writeRequestQueue.offer(session, writeRequest);
+
+ if (!session.isWriteSuspended()) {
+ this.flush(session);
+ }
+ }
+
public void remove(AbstractIoSession session) {
if (!session.getCloseFuture().isClosed()) {
session.getFilterChain().fireSessionClosed();
@@ -288,7 +303,7 @@ public class DummySession extends Abstra
}
@Override
- public void setScheduledWriteBytes(int byteCount){
+ public void setScheduledWriteBytes(int byteCount) {
super.setScheduledWriteBytes(byteCount);
}
Modified: mina/branches/2.0/mina-core/src/main/java/org/apache/mina/transport/vmpipe/VmPipeFilterChain.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0/mina-core/src/main/java/org/apache/mina/transport/vmpipe/VmPipeFilterChain.java?rev=1391490&r1=1391489&r2=1391490&view=diff
==============================================================================
--- mina/branches/2.0/mina-core/src/main/java/org/apache/mina/transport/vmpipe/VmPipeFilterChain.java (original)
+++ mina/branches/2.0/mina-core/src/main/java/org/apache/mina/transport/vmpipe/VmPipeFilterChain.java Fri Sep 28 14:50:19 2012
@@ -43,9 +43,11 @@ import org.apache.mina.core.write.WriteT
class VmPipeFilterChain extends DefaultIoFilterChain {
private final Queue<IoEvent> eventQueue = new ConcurrentLinkedQueue<IoEvent>();
+
private final IoProcessor<VmPipeSession> processor = new VmPipeIoProcessor();
private volatile boolean flushEnabled;
+
private volatile boolean sessionOpened;
VmPipeFilterChain(AbstractIoSession session) {
@@ -86,7 +88,7 @@ class VmPipeFilterChain extends DefaultI
Object data = e.getParameter();
if (type == IoEventType.MESSAGE_RECEIVED) {
- if (sessionOpened && (! session.isReadSuspended() ) && session.getLock().tryLock()) {
+ if (sessionOpened && (!session.isReadSuspended()) && session.getLock().tryLock()) {
try {
if (session.isReadSuspended()) {
session.receivedMessageQueue.add(data);
@@ -189,11 +191,9 @@ class VmPipeFilterChain extends DefaultI
while ((req = queue.poll(session)) != null) {
Object m = req.getMessage();
pushEvent(new IoEvent(IoEventType.MESSAGE_SENT, session, req), false);
- session.getRemoteSession().getFilterChain().fireMessageReceived(
- getMessageCopy(m));
+ session.getRemoteSession().getFilterChain().fireMessageReceived(getMessageCopy(m));
if (m instanceof IoBuffer) {
- session.increaseWrittenBytes0(
- ((IoBuffer) m).remaining(), currentTime);
+ session.increaseWrittenBytes0(((IoBuffer) m).remaining(), currentTime);
}
}
} finally {
@@ -213,7 +213,7 @@ class VmPipeFilterChain extends DefaultI
if (!failedRequests.isEmpty()) {
WriteToClosedSessionException cause = new WriteToClosedSessionException(failedRequests);
- for (WriteRequest r: failedRequests) {
+ for (WriteRequest r : failedRequests) {
r.getFuture().setException(cause);
}
session.getFilterChain().fireExceptionCaught(cause);
@@ -221,6 +221,19 @@ class VmPipeFilterChain extends DefaultI
}
}
+ /**
+ * {@inheritDoc}
+ */
+ public void write(VmPipeSession session, WriteRequest writeRequest) {
+ WriteRequestQueue writeRequestQueue = session.getWriteRequestQueue();
+
+ writeRequestQueue.offer(session, writeRequest);
+
+ if (!session.isWriteSuspended()) {
+ this.flush(session);
+ }
+ }
+
private Object getMessageCopy(Object message) {
Object messageCopy = message;
if (message instanceof IoBuffer) {
@@ -252,7 +265,7 @@ class VmPipeFilterChain extends DefaultI
}
public void updateTrafficControl(VmPipeSession session) {
- if ( ! session.isReadSuspended()) {
+ if (!session.isReadSuspended()) {
List<Object> data = new ArrayList<Object>();
session.receivedMessageQueue.drainTo(data);
for (Object aData : data) {
@@ -260,7 +273,7 @@ class VmPipeFilterChain extends DefaultI
}
}
- if ( ! session.isWriteSuspended()) {
+ if (!session.isWriteSuspended()) {
flush(session);
}
}
Modified: mina/branches/2.0/mina-transport-serial/src/main/java/org/apache/mina/transport/serial/SerialSessionImpl.java
URL: http://svn.apache.org/viewvc/mina/branches/2.0/mina-transport-serial/src/main/java/org/apache/mina/transport/serial/SerialSessionImpl.java?rev=1391490&r1=1391489&r2=1391490&view=diff
==============================================================================
--- mina/branches/2.0/mina-transport-serial/src/main/java/org/apache/mina/transport/serial/SerialSessionImpl.java (original)
+++ mina/branches/2.0/mina-transport-serial/src/main/java/org/apache/mina/transport/serial/SerialSessionImpl.java Fri Sep 28 14:50:19 2012
@@ -38,6 +38,7 @@ import org.apache.mina.core.service.IoSe
import org.apache.mina.core.service.TransportMetadata;
import org.apache.mina.core.session.AbstractIoSession;
import org.apache.mina.core.write.WriteRequest;
+import org.apache.mina.core.write.WriteRequestQueue;
import org.apache.mina.util.ExceptionMonitor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -256,6 +257,22 @@ class SerialSessionImpl extends Abstract
// It's already added when the session is constructed.
}
+ /**
+ * {@inheritDoc}
+ */
+ public void write(SerialSessionImpl session, WriteRequest writeRequest) {
+ WriteRequestQueue writeRequestQueue = session.getWriteRequestQueue();
+
+ writeRequestQueue.offer(session, writeRequest);
+
+ if (!session.isWriteSuspended()) {
+ session.getProcessor().flush(session);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
public void flush(SerialSessionImpl session) {
if (writeWorker == null) {
writeWorker = new WriteWorker();