QPID-443 : Fix to transactionality of message publishing

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2@525862 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 35 insertions, 0 deletions

diff --git a/java/common/src/main/java/org/apache/qpid/pool/Event.java b/java/common/src/main/java/org/apache/qpid/pool/Event.java
index 09890a103d..7300ec8c3f 100644
--- a/java/common/src/main/java/org/apache/qpid/pool/Event.java
+++ b/java/common/src/main/java/org/apache/qpid/pool/Event.java
@@ -85,4 +85,27 @@ abstract public class Event
     }
 
 
+
+    public static final class CloseEvent extends Event
+    {
+        private final IoFilter.NextFilter _nextFilter;
+
+        public CloseEvent(final IoFilter.NextFilter nextFilter)
+        {
+            super();
+            _nextFilter = nextFilter;
+        }
+
+
+        public void process(IoSession session)
+        {
+            _nextFilter.sessionClosed(session);
+        }
+
+        public IoFilter.NextFilter getNextFilter()
+        {
+            return _nextFilter;
+        }
+    }
+
 }
diff --git a/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java b/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java
index 8126ca4bc8..552ecf6b66 100644
--- a/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java
+++ b/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java
@@ -20,6 +20,8 @@
  */
 package org.apache.qpid.pool;
 
+import org.apache.qpid.pool.Event.CloseEvent;
+
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -206,6 +208,10 @@ public class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionH
             fireAsynchEvent(session, new Event.ReceivedEvent(nextFilter, message));
         }
 
+        public void sessionClosed(final NextFilter nextFilter, final IoSession session) throws Exception
+        {
+            fireAsynchEvent(session, new CloseEvent(nextFilter));
+        }
 
     }
 
@@ -224,6 +230,12 @@ public class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionH
             fireAsynchEvent(session, new Event.WriteEvent(nextFilter, writeRequest));
         }
 
+        public void sessionClosed(final NextFilter nextFilter, final IoSession session) throws Exception
+        {
+            fireAsynchEvent(session, new CloseEvent(nextFilter));
+        }
+        
+
     }
 
     public static PoolingFilter createAynschReadPoolingFilter(ReferenceCountingExecutorService refCountingPool,String name)