Hi
I've found what I think is a pretty bad race condition in how we
handle stanzas in Vysper. In short, when we receive stanzas, we queue
them in QueuedStanzaProcessor which has an Executor that will send the
stanzas of the ProtocolWorker for further handling. However,
QueuedStanzaProcessor will not ensure stanza ordering within a
session, meaning that if a client quickly sends multiple stanzas, we
might handle them out of order.
Here's my proposed patched, making use of MINA's OrderedThreadPoolExecutor:
Index:
src/main/java/org/apache/vysper/xmpp/server/DefaultServerRuntimeContext.java
===================================================================
---
src/main/java/org/apache/vysper/xmpp/server/DefaultServerRuntimeContext.java
(revision
1068181)
+++
src/main/java/org/apache/vysper/xmpp/server/DefaultServerRuntimeContext.java
(working
copy)
@@ -97,7 +97,7 @@
/**
* 'input stream': receives stanzas issued by client sessions to
be handled by the server
*/
- private StanzaProcessor stanzaProcessor = new
QueuedStanzaProcessor(new ProtocolWorker());
+ private StanzaProcessor stanzaProcessor = new ProtocolWorker();
/**
* 'output stream': receives stanzas issued by a session, which
are going to other sessions/servers
Index: src/main/java/org/apache/vysper/mina/TCPEndpoint.java
===================================================================
--- src/main/java/org/apache/vysper/mina/TCPEndpoint.java (revision
1068181)
+++ src/main/java/org/apache/vysper/mina/TCPEndpoint.java (working copy)
@@ -21,9 +21,12 @@
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.util.concurrent.TimeUnit;
import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
+import org.apache.mina.filter.executor.ExecutorFilter;
+import org.apache.mina.filter.executor.OrderedThreadPoolExecutor;
import org.apache.mina.transport.socket.SocketAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.apache.vysper.mina.codec.XMPPProtocolCodecFactory;
@@ -60,9 +71,13 @@
NioSocketAcceptor acceptor = new NioSocketAcceptor();
DefaultIoFilterChainBuilder filterChainBuilder = new
DefaultIoFilterChainBuilder();
- //filterChainBuilder.addLast("executorFilter", new
OrderedThreadPoolExecutor());
filterChainBuilder.addLast("xmppCodec", new
ProtocolCodecFilter(new XMPPProtocolCodecFactory()));
filterChainBuilder.addLast("loggingFilter", new StanzaLoggingFilter());
+
+ int coreThreadCount = 10;
+ int maxThreadCount = 20;
+ int threadTimeoutSeconds = 2 * 60;
+ filterChainBuilder.addLast("executorFilter", new
ExecutorFilter(new OrderedThreadPoolExecutor(coreThreadCount,
maxThreadCount, threadTimeoutSeconds, TimeUnit.SECONDS)));
acceptor.setFilterChainBuilder(filterChainBuilder);
XmppIoHandlerAdapter adapter = new XmppIoHandlerAdapter();
Does this make sense or is it me not getting it? :-)
/niklas
