fix for: https://issues.apache.org/jira/browse/AMQNET-326
diff --git a/src/main/csharp/MessageConsumer.cs b/src/main/csharp/MessageConsumer.cs
index 406ad52..a03ba70 100755
--- a/src/main/csharp/MessageConsumer.cs
+++ b/src/main/csharp/MessageConsumer.cs
@@ -42,7 +42,7 @@
private readonly MessageDispatchChannel unconsumedMessages;
private readonly LinkedList<MessageDispatch> dispatchedMessages = new LinkedList<MessageDispatch>();
private readonly ConsumerInfo info;
- private Session session;
+ private readonly Session session;
private MessageAck pendingAck = null;
@@ -62,9 +62,6 @@
private Exception failureError;
- private const int DEFAULT_REDELIVERY_DELAY = 0;
- private const int DEFAULT_MAX_REDELIVERIES = 5;
-
private event MessageListener listener;
private IRedeliveryPolicy redeliveryPolicy;
@@ -1227,12 +1224,16 @@
public void AfterCommit()
{
- this.consumer.DoClose();
+ Tracer.DebugFormat("ConsumerCloseSynchronization - AfterCommit Called for Consumer {0}.",
+ this.consumer.ConsumerId);
+ this.consumer.DoClose();
}
public void AfterRollback()
{
- this.consumer.DoClose();
+ Tracer.DebugFormat("ConsumerCloseSynchronization - AfterRollback Called for Consumer {0}.",
+ this.consumer.ConsumerId);
+ this.consumer.DoClose();
}
}
diff --git a/src/main/csharp/Session.cs b/src/main/csharp/Session.cs
index daedbbd..c929cac 100755
--- a/src/main/csharp/Session.cs
+++ b/src/main/csharp/Session.cs
@@ -41,7 +41,7 @@
private readonly SessionExecutor executor;
private readonly TransactionContext transactionContext;
- private Connection connection;
+ private readonly Connection connection;
private bool dispatchAsync;
private bool exclusive;
@@ -293,31 +293,29 @@
public void Close()
{
- lock(myLock)
+ if(this.closed)
{
- if(this.closed)
- {
- return;
- }
+ return;
+ }
- try
+ try
+ {
+ if(transactionContext.InNetTransaction)
{
- if(transactionContext.InNetTransaction)
- {
- this.transactionContext.AddSynchronization(new SessionCloseSynchronization(this));
- }
- else
- {
- Tracer.InfoFormat("Closing The Session with Id {0}", this.info.SessionId.ToString());
- DoClose();
- Tracer.InfoFormat("Closed The Session with Id {0}", this.info.SessionId.ToString());
- }
+ this.transactionContext.AddSynchronization(new SessionCloseSynchronization(this));
+ this.transactionContext.DtcWaitHandle.WaitOne();
}
- catch(Exception ex)
+ else
{
- Tracer.ErrorFormat("Error during session close: {0}", ex);
+ Tracer.InfoFormat("Closing The Session with Id {0}", this.info.SessionId);
+ DoClose();
+ Tracer.InfoFormat("Closed The Session with Id {0}", this.info.SessionId);
}
}
+ catch(Exception ex)
+ {
+ Tracer.ErrorFormat("Error during session close: {0}", ex);
+ }
}
internal void DoClose()
@@ -331,9 +329,16 @@
internal void Shutdown()
{
+ Tracer.InfoFormat("Executing Shutdown on Session with Id {0}", this.info.SessionId);
+
+ if(this.closed)
+ {
+ return;
+ }
+
lock(myLock)
{
- if(this.closed)
+ if(this.closed || this.closing)
{
return;
}
@@ -366,8 +371,8 @@
}
producers.Clear();
- // If in a transaction roll it back
- if(this.IsTransacted && this.transactionContext.InLocalTransaction)
+ // If in a local transaction we just roll back at this point.
+ if (this.IsTransacted && this.transactionContext.InLocalTransaction)
{
try
{
@@ -954,12 +959,14 @@
public void AfterCommit()
{
- this.session.DoClose();
+ Tracer.Debug("SessionCloseSynchronization AfterCommit called for Session: " + session.SessionId);
+ session.DoClose();
}
public void AfterRollback()
{
- this.session.DoClose();
+ Tracer.Debug("SessionCloseSynchronization AfterRollback called for Session: " + session.SessionId);
+ session.DoClose();
}
}
diff --git a/src/main/csharp/TransactionContext.cs b/src/main/csharp/TransactionContext.cs
index 72d4aca..194d1b8 100644
--- a/src/main/csharp/TransactionContext.cs
+++ b/src/main/csharp/TransactionContext.cs
@@ -220,41 +220,51 @@
throw new TransactionInProgressException("A Transaction is already in Progress");
}
- Guid rmId = ResourceManagerGuid;
+ dtcControlEvent.Reset();
- // Enlist this object in the transaction.
- this.currentEnlistment =
- transaction.EnlistDurable(rmId, this, EnlistmentOptions.None);
-
- Tracer.Debug("Enlisted in Durable Transaction with RM Id: " + rmId);
-
- TransactionInformation txInfo = transaction.TransactionInformation;
-
- XATransactionId xaId = new XATransactionId();
- this.transactionId = xaId;
-
- if(txInfo.DistributedIdentifier != Guid.Empty)
+ try
{
- xaId.GlobalTransactionId = txInfo.DistributedIdentifier.ToByteArray();
- xaId.BranchQualifier = Encoding.UTF8.GetBytes(Guid.NewGuid().ToString());
+ Guid rmId = ResourceManagerGuid;
+
+ // Enlist this object in the transaction.
+ this.currentEnlistment =
+ transaction.EnlistDurable(rmId, this, EnlistmentOptions.None);
+
+ Tracer.Debug("Enlisted in Durable Transaction with RM Id: " + rmId);
+
+ TransactionInformation txInfo = transaction.TransactionInformation;
+
+ XATransactionId xaId = new XATransactionId();
+ this.transactionId = xaId;
+
+ if (txInfo.DistributedIdentifier != Guid.Empty)
+ {
+ xaId.GlobalTransactionId = txInfo.DistributedIdentifier.ToByteArray();
+ xaId.BranchQualifier = Encoding.UTF8.GetBytes(Guid.NewGuid().ToString());
+ }
+ else
+ {
+ xaId.GlobalTransactionId = Encoding.UTF8.GetBytes(txInfo.LocalIdentifier);
+ xaId.BranchQualifier = Encoding.UTF8.GetBytes(Guid.NewGuid().ToString());
+ }
+
+ // Now notify the broker that a new XA'ish transaction has started.
+ TransactionInfo info = new TransactionInfo();
+ info.ConnectionId = this.connection.ConnectionId;
+ info.TransactionId = this.transactionId;
+ info.Type = (int) TransactionType.Begin;
+
+ this.session.Connection.Oneway(info);
+
+ if (Tracer.IsDebugEnabled)
+ {
+ Tracer.Debug("Began XA'ish Transaction:" + xaId.GlobalTransactionId.ToString());
+ }
}
- else
+ catch(Exception)
{
- xaId.GlobalTransactionId = Encoding.UTF8.GetBytes(txInfo.LocalIdentifier);
- xaId.BranchQualifier = Encoding.UTF8.GetBytes(Guid.NewGuid().ToString());
- }
-
- // Now notify the broker that a new XA'ish transaction has started.
- TransactionInfo info = new TransactionInfo();
- info.ConnectionId = this.connection.ConnectionId;
- info.TransactionId = this.transactionId;
- info.Type = (int) TransactionType.Begin;
-
- this.session.Connection.Oneway(info);
-
- if(Tracer.IsDebugEnabled)
- {
- Tracer.Debug("Began XA'ish Transaction:" + xaId.GlobalTransactionId.ToString());
+ dtcControlEvent.Set();
+ throw;
}
}
@@ -262,8 +272,6 @@
{
try
{
- dtcControlEvent.Reset();
-
Tracer.Debug("Prepare notification received for TX id: " + this.transactionId);
BeforeEnd();
@@ -302,6 +310,9 @@
// Done so commit won't be called.
AfterCommit();
+
+ // A Read-Only TX is considered closed at this point, DTC won't call us again.
+ this.dtcControlEvent.Set();
}
else
{
@@ -329,19 +340,14 @@
this.currentEnlistment = null;
this.transactionId = null;
- }
- finally
- {
this.dtcControlEvent.Set();
- }
+ }
}
public void Commit(Enlistment enlistment)
{
try
{
- dtcControlEvent.Reset();
-
Tracer.Debug("Commit notification received for TX id: " + this.transactionId);
if (this.transactionId != null)
@@ -398,8 +404,6 @@
{
try
{
- dtcControlEvent.Reset();
-
Tracer.Debug("Single Phase Commit notification received for TX id: " + this.transactionId);
if (this.transactionId != null)
@@ -450,9 +454,7 @@
public void Rollback(Enlistment enlistment)
{
try
- {
- dtcControlEvent.Reset();
-
+ {
Tracer.Debug("Rollback notification received for TX id: " + this.transactionId);
if (this.transactionId != null)
@@ -515,8 +517,6 @@
{
try
{
- dtcControlEvent.Reset();
-
Tracer.Debug("In Doubt notification received for TX id: " + this.transactionId);
BeforeEnd();
diff --git a/src/test/csharp/DtcConsumerTransactionsTest.cs b/src/test/csharp/DtcConsumerTransactionsTest.cs
index 75292c2..9250fec 100644
--- a/src/test/csharp/DtcConsumerTransactionsTest.cs
+++ b/src/test/csharp/DtcConsumerTransactionsTest.cs
@@ -16,6 +16,7 @@
*/
using System;
+using System.Collections;
using System.Data.SqlClient;
using System.IO;
using System.Threading;
@@ -32,6 +33,68 @@
class DtcConsumerTransactionsTest : DtcTransactionsTestSupport
{
[Test]
+ public void TestRedelivered()
+ {
+ // enqueue several messages
+ PurgeDatabase();
+ PurgeAndFillQueue();
+
+ // receive just one
+ INetTxConnectionFactory factory = new NetTxConnectionFactory(ReplaceEnvVar(connectionURI));
+ using (INetTxConnection connection = factory.CreateNetTxConnection())
+ {
+ connection.Start();
+
+ using (INetTxSession session = connection.CreateNetTxSession())
+ {
+ IQueue queue = session.GetQueue(testQueueName);
+
+ // read message from queue and insert into db table
+ using (IMessageConsumer consumer = session.CreateConsumer(queue))
+ {
+ using (TransactionScope scoped = new TransactionScope(TransactionScopeOption.RequiresNew))
+ using (SqlConnection sqlConnection = new SqlConnection(sqlConnectionString))
+ using (SqlCommand sqlInsertCommand = new SqlCommand())
+ {
+ sqlConnection.Open();
+ sqlInsertCommand.Connection = sqlConnection;
+
+ ITextMessage message = consumer.Receive(TimeSpan.FromMilliseconds(10000)) as ITextMessage;
+ sqlInsertCommand.CommandText =
+ string.Format("INSERT INTO {0} VALUES ({1})", testTable, Convert.ToInt32(message.Text));
+ sqlInsertCommand.ExecuteNonQuery();
+
+ scoped.Complete();
+ }
+ }
+
+ session.Close();
+ }
+ }
+
+ // check that others message have status redelivered = false
+ IConnectionFactory checkFactory = new ConnectionFactory(ReplaceEnvVar(connectionURI));
+
+ using (IConnection connection = checkFactory.CreateConnection())
+ {
+ connection.Start();
+
+ using (ISession session = connection.CreateSession())
+ using (IQueueBrowser browser = session.CreateBrowser(session.GetQueue(testQueueName)))
+ {
+ IEnumerator enumerator = browser.GetEnumerator();
+
+ while (enumerator.MoveNext())
+ {
+ IMessage msg = enumerator.Current as IMessage;
+ Assert.IsNotNull(msg, "message is not in the queue!");
+ Assert.IsFalse(msg.NMSRedelivered, "message is redelivered!");
+ }
+ }
+ }
+ }
+
+ [Test]
public void TestRecoveryAfterCommitFailsBeforeSent()
{
// Test initialize - Fills in queue with data to send and clears the DB.
diff --git a/src/test/csharp/NetTxTransactionTest.cs b/src/test/csharp/NetTxTransactionTest.cs
index b60f3a7..75d2b73 100644
--- a/src/test/csharp/NetTxTransactionTest.cs
+++ b/src/test/csharp/NetTxTransactionTest.cs
@@ -21,8 +21,6 @@
using NUnit.Framework;
using Apache.NMS.Test;
-using Apache.NMS.Util;
-using Apache.NMS.ActiveMQ;
namespace Apache.NMS.ActiveMQ.Test
{
@@ -236,10 +234,10 @@
producer.Send(session.CreateTextMessage("Hello World"));
}
- session.Close();
-
scoped.Complete();
}
+
+ session.Close();
}
}
@@ -255,10 +253,10 @@
Assert.IsNotNull(msg, "Message was null for index: " + i);
}
- session.Close();
-
scoped.Complete();
}
+
+ session.Close();
}
}
