From 45b41e212c827905f711c188457ed6cdcb97aab3 Mon Sep 17 00:00:00 2001
From: Robert Greig <rgreig@apache.org>
Date: Mon, 29 Jan 2007 16:37:13 +0000
Subject: QPID-327 : Patch supplied by Rob Godfrey - [race condition]
 PoolingFilter : Possible race condition when completing a Job

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@501096 13f79535-47bb-0310-9956-ffa450edef68
---
 .../src/main/java/org/apache/qpid/server/Main.java   |  2 +-
 .../qpid/server/protocol/AMQMinaProtocolSession.java | 20 ++++++++++++++++++++
 2 files changed, 21 insertions(+), 1 deletion(-)

(limited to 'java/broker/src')

diff --git a/java/broker/src/main/java/org/apache/qpid/server/Main.java b/java/broker/src/main/java/org/apache/qpid/server/Main.java
index c45d1ad2c2..55009bbf49 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/Main.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/Main.java
@@ -324,7 +324,7 @@ public class Main implements ProtocolVersionList
             // implementation provided by MINA
             if (connectorConfig.enableExecutorPool)
             {
-                sconfig.setThreadModel(new ReadWriteThreadModel());
+                sconfig.setThreadModel(ReadWriteThreadModel.getInstance());
             }
 
             if (connectorConfig.enableNonSSL)
diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
index 5f9fcbdc85..460d5126ca 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
@@ -23,10 +23,13 @@ package org.apache.qpid.server.protocol;
 import org.apache.log4j.Logger;
 import org.apache.mina.common.IdleStatus;
 import org.apache.mina.common.IoSession;
+import org.apache.mina.common.IoSessionConfig;
+import org.apache.mina.common.IoServiceConfig;
 import org.apache.mina.transport.vmpipe.VmPipeAddress;
 import org.apache.qpid.AMQChannelException;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.AMQConnectionException;
+import org.apache.qpid.pool.ReadWriteThreadModel;
 import org.apache.qpid.common.ClientProperties;
 import org.apache.qpid.framing.*;
 import org.apache.qpid.protocol.AMQMethodEvent;
@@ -121,6 +124,23 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
         _codecFactory = codecFactory;
 
 
+        try
+        {
+            IoServiceConfig config = session.getServiceConfig();
+            ReadWriteThreadModel threadModel = (ReadWriteThreadModel) config.getThreadModel();
+            threadModel.getAsynchronousReadFilter().createNewJobForSession(session);
+            threadModel.getAsynchronousWriteFilter().createNewJobForSession(session);
+        }
+        catch (RuntimeException e)
+        {
+            e.printStackTrace();
+        //    throw e;
+
+        }
+
+
+
+
 //        this(session, queueRegistry, exchangeRegistry, codecFactory, new AMQStateManager());
     }
 
-- 
cgit v1.2.1