activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jgo...@apache.org
Subject svn commit: r1312026 - in /activemq/activemq-dotnet/Apache.NMS.ActiveMQ/branches/1.5.x/src: main/csharp/Connection.cs test/csharp/TempDestinationTest.cs
Date Tue, 10 Apr 2012 22:25:14 GMT
Author: jgomes
Date: Tue Apr 10 22:25:13 2012
New Revision: 1312026

URL: http://svn.apache.org/viewvc?rev=1312026&view=rev
Log:
Make of copy of the temporary destinations that are being cleaned up when closing a Connection
to avoid a race condition of modifying the list of temp destinations while it is being enumerated.
Found that the AdvisoryConsumer monitor was creating a race condition while populating the
internal temp destination list associated with a Connection so that the Connection member
reference was not being set on the temp destination, and subsequently the temp destination
would not be cleaned up when the Connection was closed.
Resolves AMQNET-378.

Modified:
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/branches/1.5.x/src/main/csharp/Connection.cs
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/branches/1.5.x/src/test/csharp/TempDestinationTest.cs

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/branches/1.5.x/src/main/csharp/Connection.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/branches/1.5.x/src/main/csharp/Connection.cs?rev=1312026&r1=1312025&r2=1312026&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/branches/1.5.x/src/main/csharp/Connection.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/branches/1.5.x/src/main/csharp/Connection.cs
Tue Apr 10 22:25:13 2012
@@ -46,8 +46,8 @@ namespace Apache.NMS.ActiveMQ
 		private bool sendAcksAsync = false;
 		private bool dispatchAsync = true;
 		private int producerWindowSize = 0;
-		private bool messagePrioritySupported=true;
-        private bool watchTopicAdviosires = true;
+		private bool messagePrioritySupported = true;
+		private bool watchTopicAdviosires = true;
 
 		private bool userSpecifiedClientID;
 		private readonly Uri brokerUri;
@@ -60,7 +60,7 @@ namespace Apache.NMS.ActiveMQ
 		private readonly IList sessions = ArrayList.Synchronized(new ArrayList());
 		private readonly IDictionary producers = Hashtable.Synchronized(new Hashtable());
 		private readonly IDictionary dispatchers = Hashtable.Synchronized(new Hashtable());
-        private readonly IDictionary tempDests = Hashtable.Synchronized(new Hashtable());
+		private readonly IDictionary tempDests = Hashtable.Synchronized(new Hashtable());
 		private readonly object myLock = new object();
 		private readonly Atomic<bool> connected = new Atomic<bool>(false);
 		private readonly Atomic<bool> closed = new Atomic<bool>(false);
@@ -77,11 +77,11 @@ namespace Apache.NMS.ActiveMQ
 		private PrefetchPolicy prefetchPolicy = new PrefetchPolicy();
 		private ICompressionPolicy compressionPolicy = new CompressionPolicy();
 		private readonly IdGenerator clientIdGenerator;
-        private int consumerIdCounter = 0;
+		private int consumerIdCounter = 0;
 		private volatile CountDownLatch transportInterruptionProcessingComplete;
 		private readonly MessageTransformation messageTransformation;
 		private readonly ThreadPoolExecutor executor = new ThreadPoolExecutor();
-        private AdvisoryConsumer advisoryConsumer = null;
+		private AdvisoryConsumer advisoryConsumer = null;
 
 		public Connection(Uri connectionUri, ITransport transport, IdGenerator clientIdGenerator)
 		{
@@ -324,11 +324,11 @@ namespace Apache.NMS.ActiveMQ
 			set { this.dispatchAsync = value; }
 		}
 
-        public bool WatchTopicAdvisories
-        {
-            get { return this.watchTopicAdviosires; }
-            set { this.watchTopicAdviosires = value; }
-        }
+		public bool WatchTopicAdvisories
+		{
+			get { return this.watchTopicAdviosires; }
+			set { this.watchTopicAdviosires = value; }
+		}
 
 		public string ClientId
 		{
@@ -545,11 +545,11 @@ namespace Apache.NMS.ActiveMQ
 					Tracer.Info("Connection.Close(): Closing Connection Now.");
 					this.closing.Value = true;
 
-                    if(this.advisoryConsumer != null)
-                    {
-                        this.advisoryConsumer.Dispose();
-                        this.advisoryConsumer = null;
-                    }
+					if(this.advisoryConsumer != null)
+					{
+						this.advisoryConsumer.Dispose();
+						this.advisoryConsumer = null;
+					}
 
 					lock(sessions.SyncRoot)
 					{
@@ -560,10 +560,18 @@ namespace Apache.NMS.ActiveMQ
 					}
 					sessions.Clear();
 
-                    foreach(ActiveMQTempDestination dest in this.tempDests.Values)
-                    {
-                        dest.Delete();
-                    }
+					if(this.tempDests.Count > 0)
+					{
+						// Make a copy of the destinations to delete, because the act of deleting
+						// them will modify the collection.
+						ActiveMQTempDestination[] tempDestsToDelete = new ActiveMQTempDestination[this.tempDests.Count];
+
+						this.tempDests.Values.CopyTo(tempDestsToDelete, 0);
+						foreach(ActiveMQTempDestination dest in tempDestsToDelete)
+						{
+							dest.Delete();
+						}
+					}
 
 					// Connected is true only when we've successfully sent our ConnectionInfo
 					// to the broker, so if we haven't announced ourselves there's no need to
@@ -655,7 +663,7 @@ namespace Apache.NMS.ActiveMQ
 				Response response = transport.Request(command, requestTimeout);
 				if(response is ExceptionResponse)
 				{
-					ExceptionResponse exceptionResponse = (ExceptionResponse) response;
+					ExceptionResponse exceptionResponse = (ExceptionResponse)response;
 					BrokerError brokerError = exceptionResponse.Exception;
 					throw new BrokerException(brokerError);
 				}
@@ -752,12 +760,12 @@ namespace Apache.NMS.ActiveMQ
 										if(!(response is ExceptionResponse))
 										{
 											connected.Value = true;
-                                            if(this.watchTopicAdviosires)
-                                            {
-                                                ConsumerId id = new ConsumerId(
-                                                    new SessionId(info.ConnectionId, -1),
-                                                    Interlocked.Increment(ref this.consumerIdCounter));
-                                                this.advisoryConsumer = new AdvisoryConsumer(this,
id);
+											if(this.watchTopicAdviosires)
+											{
+												ConsumerId id = new ConsumerId(
+													new SessionId(info.ConnectionId, -1),
+													Interlocked.Increment(ref this.consumerIdCounter));
+												this.advisoryConsumer = new AdvisoryConsumer(this, id);
 											}
 										}
 									}
@@ -800,19 +808,19 @@ namespace Apache.NMS.ActiveMQ
 			if(command.IsMessageDispatch)
 			{
 				WaitForTransportInterruptionProcessingToComplete();
-				DispatchMessage((MessageDispatch) command);
+				DispatchMessage((MessageDispatch)command);
 			}
 			else if(command.IsKeepAliveInfo)
 			{
-				OnKeepAliveCommand(commandTransport, (KeepAliveInfo) command);
+				OnKeepAliveCommand(commandTransport, (KeepAliveInfo)command);
 			}
 			else if(command.IsWireFormatInfo)
 			{
-				this.brokerWireFormatInfo = (WireFormatInfo) command;
+				this.brokerWireFormatInfo = (WireFormatInfo)command;
 			}
 			else if(command.IsBrokerInfo)
 			{
-				this.brokerInfo = (BrokerInfo) command;
+				this.brokerInfo = (BrokerInfo)command;
 				this.brokerInfoReceived.countDown();
 			}
 			else if(command.IsShutdownInfo)
@@ -824,7 +832,7 @@ namespace Apache.NMS.ActiveMQ
 			}
 			else if(command.IsProducerAck)
 			{
-				ProducerAck ack = (ProducerAck) command as ProducerAck;
+				ProducerAck ack = (ProducerAck)command as ProducerAck;
 				if(ack.ProducerId != null)
 				{
 					MessageProducer producer = producers[ack.ProducerId] as MessageProducer;
@@ -843,7 +851,7 @@ namespace Apache.NMS.ActiveMQ
 			{
 				if(!closing.Value && !closed.Value)
 				{
-					ConnectionError connectionError = (ConnectionError) command;
+					ConnectionError connectionError = (ConnectionError)command;
 					BrokerError brokerError = connectionError.Exception;
 					string message = "Broker connection error.";
 					string cause = "";
@@ -873,7 +881,7 @@ namespace Apache.NMS.ActiveMQ
 			{
 				if(dispatchers.Contains(dispatch.ConsumerId))
 				{
-					IDispatcher dispatcher = (IDispatcher) dispatchers[dispatch.ConsumerId];
+					IDispatcher dispatcher = (IDispatcher)dispatchers[dispatch.ConsumerId];
 
 					// Can be null when a consumer has sent a MessagePull and there was
 					// no available message at the broker to dispatch or when signalled
@@ -926,7 +934,7 @@ namespace Apache.NMS.ActiveMQ
 					{
 						error = NMSExceptionSupport.Create(error);
 					}
-					NMSException e = (NMSException) error;
+					NMSException e = (NMSException)error;
 
 					// Called in another thread so that processing can continue
 					// here, ensures no lock contention.
@@ -1014,12 +1022,12 @@ namespace Apache.NMS.ActiveMQ
 		{
 			Tracer.Debug("Connection: Transport has been Interrupted.");
 
-            // Ensure that if there's an advisory consumer we don't add it to the
-            // set of consumers that need interruption processing.
+			// Ensure that if there's an advisory consumer we don't add it to the
+			// set of consumers that need interruption processing.
 			this.transportInterruptionProcessingComplete =
-                new CountDownLatch(dispatchers.Count - (this.advisoryConsumer != null ? 1
: 0));
+				new CountDownLatch(dispatchers.Count - (this.advisoryConsumer != null ? 1 : 0));
 
-            if(Tracer.IsDebugEnabled)
+			if(Tracer.IsDebugEnabled)
 			{
 				Tracer.Debug("transport interrupted, dispatchers: " + dispatchers.Count);
 			}
@@ -1120,8 +1128,8 @@ namespace Apache.NMS.ActiveMQ
 
 			this.SyncRequest(command);
 
+			destination = this.AddTempDestination(destination);
 			destination.Connection = this;
-            this.AddTempDestination(destination);
 
 			return destination;
 		}
@@ -1132,19 +1140,19 @@ namespace Apache.NMS.ActiveMQ
 
 		public void DeleteTemporaryDestination(IDestination destination)
 		{
-            CheckClosedOrFailed();
+			CheckClosedOrFailed();
 
-            ActiveMQTempDestination temp = destination as ActiveMQTempDestination;
+			ActiveMQTempDestination temp = destination as ActiveMQTempDestination;
 
-            foreach(Session session in this.sessions)
-            {
-                if(session.IsInUse(temp))
-                {
-                    throw new NMSException("A consumer is consuming from the temporary destination");
-                }
-            }
+			foreach(Session session in this.sessions)
+			{
+				if(session.IsInUse(temp))
+				{
+					throw new NMSException("A consumer is consuming from the temporary destination");
+				}
+			}
 
-            this.tempDests.Remove(destination as ActiveMQTempDestination);
+			this.tempDests.Remove(destination as ActiveMQTempDestination);
 			this.DeleteDestination(destination);
 		}
 
@@ -1153,7 +1161,7 @@ namespace Apache.NMS.ActiveMQ
 			DestinationInfo command = new DestinationInfo();
 			command.ConnectionId = this.ConnectionId;
 			command.OperationType = DestinationInfo.REMOVE_OPERATION_TYPE; // 1 is remove
-			command.Destination = (ActiveMQDestination) destination;
+			command.Destination = (ActiveMQDestination)destination;
 
 			this.Oneway(command);
 		}
@@ -1230,48 +1238,56 @@ namespace Apache.NMS.ActiveMQ
 			}
 		}
 
-        internal void AddTempDestination(ActiveMQTempDestination dest)
-        {
-            // .NET lacks a putIfAbsent operation for Maps.
-            lock(tempDests.SyncRoot)
-            {
-                if(!this.tempDests.Contains(dest))
-                {
-                    this.tempDests.Add(dest, dest);
-                }
-            }
-        }
-
-        internal void RemoveTempDestination(ActiveMQTempDestination dest)
-        {
-            this.tempDests.Remove(dest);
-        }
-
-        internal bool IsTempDestinationActive(ActiveMQTempDestination dest)
-        {
-            if(this.advisoryConsumer == null)
-            {
-                return true;
-            }
-
-            return this.tempDests.Contains(dest);
-        }
-
-        protected void CheckClosedOrFailed()
-        {
-            CheckClosed();
-            if (transportFailed.Value)
-            {
-                throw new ConnectionFailedException(firstFailureError.Message);
-            }
-        }
-
-        protected void CheckClosed()
-        {
-            if(closed.Value)
-            {
-                throw new ConnectionClosedException();
-            }
-        }
+		internal ActiveMQTempDestination AddTempDestination(ActiveMQTempDestination dest)
+		{
+			ActiveMQTempDestination addedDest = dest;
+
+			// .NET lacks a putIfAbsent operation for Maps.
+			lock(tempDests.SyncRoot)
+			{
+				if(!this.tempDests.Contains(dest))
+				{
+					this.tempDests.Add(dest, dest);
+				}
+				else
+				{
+					addedDest = this.tempDests[dest] as ActiveMQTempDestination;
+				}
+			}
+
+			return addedDest;
+		}
+
+		internal void RemoveTempDestination(ActiveMQTempDestination dest)
+		{
+			this.tempDests.Remove(dest);
+		}
+
+		internal bool IsTempDestinationActive(ActiveMQTempDestination dest)
+		{
+			if(this.advisoryConsumer == null)
+			{
+				return true;
+			}
+
+			return this.tempDests.Contains(dest);
+		}
+
+		protected void CheckClosedOrFailed()
+		{
+			CheckClosed();
+			if(transportFailed.Value)
+			{
+				throw new ConnectionFailedException(firstFailureError.Message);
+			}
+		}
+
+		protected void CheckClosed()
+		{
+			if(closed.Value)
+			{
+				throw new ConnectionClosedException();
+			}
+		}
 	}
 }

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/branches/1.5.x/src/test/csharp/TempDestinationTest.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/branches/1.5.x/src/test/csharp/TempDestinationTest.cs?rev=1312026&r1=1312025&r2=1312026&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/branches/1.5.x/src/test/csharp/TempDestinationTest.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/branches/1.5.x/src/test/csharp/TempDestinationTest.cs
Tue Apr 10 22:25:13 2012
@@ -236,7 +236,7 @@ namespace Apache.NMS.ActiveMQ.Test
         /// Make sure you cannot publish to a temp destination that does not exist anymore.
         /// </summary>
         [Test]
-        public void TestPublishFailsForDestoryedTempDestination()
+        public void TestPublishFailsForDestroyedTempDestination()
         {
 			Connection connection = GetNewConnection();
 			Connection tempConnection = GetNewConnection();
@@ -310,7 +310,7 @@ namespace Apache.NMS.ActiveMQ.Test
 
 				connections.Remove(producerConnection);
 				producerConnection.Close();
-				//Thread.Sleep(2000); // Wait a little bit to let the delete take effect.
+				Thread.Sleep(1000); // Wait a little bit to let the delete take effect.
 
 				// This message delivery NOT should work since the temp destination was removed by closing
the connection.
 				try
@@ -351,6 +351,25 @@ namespace Apache.NMS.ActiveMQ.Test
             }
         }
 
-    }
+		/// <summary>
+		/// Test clean up of multiple temp destinations
+		/// </summary>
+		[Test]
+		public void TestCloseConnectionWithTempQueues()
+		{
+			List<ITemporaryQueue> listTempQueues = new List<ITemporaryQueue>();
+			IConnection connection = CreateConnection();
+			ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+
+			connection.Start();
+
+			for(int index = 0; index < 25; index++)
+			{
+				listTempQueues.Add(session.CreateTemporaryQueue());
+			}
+
+			connection.Close();
+		}
+	}
 }
 



Mime
View raw message