QPID-4541 Added the ability to turn off replay at the time the session

is created. The XASessionImpl will use this feature to turn off replay
on the underlying JMS session. This prevents messages being replayed
outside the boundaries of the XA transaction.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1438725 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 36 insertions, 8 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/XASessionImpl.java b/java/client/src/main/java/org/apache/qpid/client/XASessionImpl.java
index 05bd121bbd..e01ec8578d 100644
--- a/java/client/src/main/java/org/apache/qpid/client/XASessionImpl.java
+++ b/java/client/src/main/java/org/apache/qpid/client/XASessionImpl.java
@@ -88,7 +88,7 @@ public class XASessionImpl extends AMQSession_0_10 implements XASession, XATopic
      */
     public void createSession()
     {
-        _qpidDtxSession = getQpidConnection().createSession(0);
+        _qpidDtxSession = getQpidConnection().createSession(0,true);
         _qpidDtxSession.setSessionListener(this);
         _qpidDtxSession.dtxSelect();
     }
diff --git a/java/client/src/test/java/org/apache/qpid/client/AMQSession_0_10Test.java b/java/client/src/test/java/org/apache/qpid/client/AMQSession_0_10Test.java
index 59181c3a13..d4e0ee60a6 100644
--- a/java/client/src/test/java/org/apache/qpid/client/AMQSession_0_10Test.java
+++ b/java/client/src/test/java/org/apache/qpid/client/AMQSession_0_10Test.java
@@ -597,7 +597,7 @@ public class AMQSession_0_10Test extends QpidTestCase
         connection.setSessionFactory(new SessionFactory()
         {
 
-            public Session newSession(Connection conn, Binary name, long expiry)
+            public Session newSession(Connection conn, Binary name, long expiry, boolean isNoReplay)
             {
                 return new MockSession(conn, new SessionDelegate(), name, expiry, throwException);
             }
diff --git a/java/common/src/main/java/org/apache/qpid/transport/Connection.java b/java/common/src/main/java/org/apache/qpid/transport/Connection.java
index 5ae2f1ceb2..cdca726148 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/Connection.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/Connection.java
@@ -86,15 +86,15 @@ public class Connection extends ConnectionInvoker
 
     public static interface SessionFactory
     {
-        Session newSession(Connection conn, Binary name, long expiry);
+        Session newSession(Connection conn, Binary name, long expiry, boolean isNoReplay);
     }
 
     private static final class DefaultSessionFactory implements SessionFactory
     {
 
-        public Session newSession(final Connection conn, final Binary name, final long expiry)
+        public Session newSession(final Connection conn, final Binary name, final long expiry, final boolean isNoReplay)
         {
-            return new Session(conn, name, expiry);
+            return new Session(conn, name, expiry, isNoReplay);
         }
     }
 
@@ -296,7 +296,12 @@ public class Connection extends ConnectionInvoker
 
     public Session createSession(long expiry)
     {
-        return createSession(UUID.randomUUID().toString(), expiry);
+        return createSession(expiry, false);
+    }
+
+    public Session createSession(long expiry, boolean isNoReplay)
+    {
+        return createSession(UUID.randomUUID().toString(), expiry, isNoReplay);
     }
 
     public Session createSession(String name)
@@ -309,6 +314,11 @@ public class Connection extends ConnectionInvoker
         return createSession(Strings.toUTF8(name), expiry);
     }
 
+    public Session createSession(String name, long expiry,boolean isNoReplay)
+    {
+        return createSession(new Binary(Strings.toUTF8(name)), expiry, isNoReplay);
+    }
+
     public Session createSession(byte[] name, long expiry)
     {
         return createSession(new Binary(name), expiry);
@@ -316,6 +326,11 @@ public class Connection extends ConnectionInvoker
 
     public Session createSession(Binary name, long expiry)
     {
+        return createSession(name, expiry, false);
+    }
+
+    public Session createSession(Binary name, long expiry, boolean isNoReplay)
+    {
         synchronized (lock)
         {
             Waiter w = new Waiter(lock, timeout);
@@ -329,7 +344,7 @@ public class Connection extends ConnectionInvoker
                 throw new ConnectionException("Timed out waiting for connection to be ready. Current state is :" + state);
             }
 
-            Session ssn = _sessionFactory.newSession(this, name, expiry);
+            Session ssn = _sessionFactory.newSession(this, name, expiry, isNoReplay);
             registerSession(ssn);
             map(ssn);
             ssn.attach();
diff --git a/java/common/src/main/java/org/apache/qpid/transport/Session.java b/java/common/src/main/java/org/apache/qpid/transport/Session.java
index d66415c659..8b29d6e424 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/Session.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/Session.java
@@ -131,19 +131,31 @@ public class Session extends SessionInvoker
     private final Object stateLock = new Object();
 
     private final AtomicBoolean _failoverRequired = new AtomicBoolean(false);
+    private boolean _isNoReplay = false;
 
     protected Session(Connection connection, Binary name, long expiry)
     {
         this(connection, new SessionDelegate(), name, expiry);
     }
 
+    protected Session(Connection connection, Binary name, long expiry, boolean noReplay)
+    {
+        this(connection, new SessionDelegate(), name, expiry, noReplay);
+    }
+
     protected Session(Connection connection, SessionDelegate delegate, Binary name, long expiry)
     {
+        this(connection, delegate, name, expiry,false);
+    }
+
+    protected Session(Connection connection, SessionDelegate delegate, Binary name, long expiry, boolean noReplay)
+    {
         this.connection = connection;
         this.delegate = delegate;
         this.name = name;
         this.expiry = expiry;
         this.closing = false;
+        this._isNoReplay = noReplay;
         initReceiver();
     }
 
@@ -281,6 +293,7 @@ public class Session extends SessionInvoker
     void resume()
     {
         _failoverRequired.set(false);
+
         synchronized (commandsLock)
         {
             attach();
@@ -739,7 +752,7 @@ public class Session extends SessionInvoker
                     sessionCommandPoint(0, 0);
                 }
 
-                boolean replayTransfer = !closing && !transacted &&
+                boolean replayTransfer = !_isNoReplay && !closing && !transacted &&
                                          m instanceof MessageTransfer &&
                                          ! m.isUnreliable();