activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tab...@apache.org
Subject svn commit: r1054714 - in /activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp: OpenWire/ State/ Transport/ Transport/Failover/ Transport/Mock/ Transport/Tcp/
Date Mon, 03 Jan 2011 19:25:04 GMT
Author: tabish
Date: Mon Jan  3 19:25:03 2011
New Revision: 1054714

URL: http://svn.apache.org/viewvc?rev=1054714&view=rev
Log:
Fix for: https://issues.apache.org/jira/browse/AMQNET-293

Consumer not recovered using two step ConsumerInfo / ConsumerControl method if WireFormat
is less then openwire v6.  

Modified:
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/OpenWire/OpenWireFormat.cs
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/State/ConnectionStateTracker.cs
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/FailoverTransport.cs
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/ITransport.cs
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/IWireFormat.cs
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Mock/MockTransport.cs
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Tcp/TcpTransport.cs
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/TransportFilter.cs
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/WireFormatNegotiator.cs

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/OpenWire/OpenWireFormat.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/OpenWire/OpenWireFormat.cs?rev=1054714&r1=1054713&r2=1054714&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/OpenWire/OpenWireFormat.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/OpenWire/OpenWireFormat.cs
Mon Jan  3 19:25:03 2011
@@ -363,7 +363,7 @@ namespace Apache.NMS.ActiveMQ.OpenWire
             return null;
 		}
 
-		public void renegotiateWireFormat(WireFormatInfo info)
+		public void RenegotiateWireFormat(WireFormatInfo info)
 		{
 			if(info.Version < minimumVersion)
 			{

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/State/ConnectionStateTracker.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/State/ConnectionStateTracker.cs?rev=1054714&r1=1054713&r2=1054714&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/State/ConnectionStateTracker.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/State/ConnectionStateTracker.cs
Mon Jan  3 19:25:03 2011
@@ -29,7 +29,6 @@ namespace Apache.NMS.ActiveMQ.State
     /// </summary>
     public class ConnectionStateTracker : CommandVisitorAdapter
     {
-
         private static readonly Tracked TRACKED_RESPONSE_MARKER = new Tracked(null);
 
         protected Dictionary<ConnectionId, ConnectionState> connectionStates = new
Dictionary<ConnectionId, ConnectionState>();
@@ -81,7 +80,7 @@ namespace Apache.NMS.ActiveMQ.State
         /// </summary>
         /// <param name="command"></param>
         /// <returns>null if the command is not state tracked.</returns>
-        public Tracked track(Command command)
+        public Tracked Track(Command command)
         {
             try
             {
@@ -97,7 +96,7 @@ namespace Apache.NMS.ActiveMQ.State
             }
         }
 
-        public void trackBack(Command command)
+        public void TrackBack(Command command)
         {
             if(TrackMessages && command != null && command.IsMessage)
             {
@@ -190,7 +189,7 @@ namespace Apache.NMS.ActiveMQ.State
             {
                 ConsumerInfo infoToSend = consumerState.Info;
 
-                if(!connectionInterruptionProcessingComplete && infoToSend.PrefetchSize
> 0)
+                if(!connectionInterruptionProcessingComplete && infoToSend.PrefetchSize
> 0 && transport.WireFormat.Version > 5)
                 {
                     infoToSend = consumerState.Info.Clone() as ConsumerInfo;
                     connectionState.RecoveringPullConsumers.Add(infoToSend.ConsumerId, consumerState.Info);

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/FailoverTransport.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/FailoverTransport.cs?rev=1054714&r1=1054713&r2=1054714&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/FailoverTransport.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/FailoverTransport.cs
Mon Jan  3 19:25:03 2011
@@ -274,6 +274,20 @@ namespace Apache.NMS.ActiveMQ.Transport.
 			set { useExponentialBackOff = value; }
 		}
 
+        public IWireFormat WireFormat
+        {
+            get
+            {
+                ITransport transport = ConnectedTransport;
+                if(transport != null)
+                {
+                    return transport.WireFormat;
+                }
+
+                return null;
+            }
+        }
+
 		/// <summary>
 		/// Gets or sets a value indicating whether to asynchronously connect to sockets
 		/// </summary>
@@ -625,7 +639,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
 						// If it was a request and it was not being tracked by
 						// the state tracker, then hold it in the requestMap so
 						// that we can replay it later.
-						Tracked tracked = stateTracker.track(command);
+						Tracked tracked = stateTracker.Track(command);
 						lock(((ICollection) requestMap).SyncRoot)
 						{
 							if(tracked != null && tracked.WaitingForResponse)
@@ -642,7 +656,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
 						try
 						{
 							transport.Oneway(command);
-							stateTracker.trackBack(command);
+							stateTracker.TrackBack(command);
 						}
 						catch(Exception e)
 						{
@@ -1220,7 +1234,6 @@ namespace Apache.NMS.ActiveMQ.Transport.
 			}
 		}
 
-
 		public void UpdateURIs(bool rebalance, Uri[] updatedURIs)
 		{
 			if(IsUpdateURIsSupported)

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/ITransport.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/ITransport.cs?rev=1054714&r1=1054713&r2=1054714&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/ITransport.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/ITransport.cs
Mon Jan  3 19:25:03 2011
@@ -186,6 +186,15 @@ namespace Apache.NMS.ActiveMQ.Transport
 		/// </param>
 		void UpdateURIs(bool rebalance, Uri[] updatedURIs);
 
+        /// <summary>
+        /// Returns the IWireFormat object that this transport uses to marshal and
+        /// unmarshal Command objects.
+        /// </summary>
+        IWireFormat WireFormat
+        {
+            get;
+        }
+
 	}
 }
 

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/IWireFormat.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/IWireFormat.cs?rev=1054714&r1=1054713&r2=1054714&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/IWireFormat.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/IWireFormat.cs
Mon Jan  3 19:25:03 2011
@@ -34,9 +34,24 @@ namespace Apache.NMS.ActiveMQ.Transport
 		/// </summary>
         Object Unmarshal(BinaryReader dis);
 
-		ITransport Transport {
-			get; set;
+        /// <summary>
+        /// Gets the Transport that own this WireFormat instnace.
+        /// </summary>
+		ITransport Transport
+        {
+			get;
+            set;
 		}
+
+        /// <summary>
+        /// Gets the current version of the protocol that this WireFormat instance
+        /// supports
+        /// </summary>
+        int Version
+        {
+            get;
+        }
+        
     }
 }
 

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Mock/MockTransport.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Mock/MockTransport.cs?rev=1054714&r1=1054713&r2=1054714&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Mock/MockTransport.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Mock/MockTransport.cs
Mon Jan  3 19:25:03 2011
@@ -433,6 +433,11 @@ namespace Apache.NMS.ActiveMQ.Transport.
 			throw new IOException();
 		}
 
+        public IWireFormat WireFormat
+        {
+            get { return null; }
+        }
+
 		#endregion
 	}
 }

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Tcp/TcpTransport.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Tcp/TcpTransport.cs?rev=1054714&r1=1054713&r2=1054714&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Tcp/TcpTransport.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Tcp/TcpTransport.cs
Mon Jan  3 19:25:03 2011
@@ -129,7 +129,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
 					seenShutdown = true;
 				}
 
-				Wireformat.Marshal(command, socketWriter);
+				WireFormat.Marshal(command, socketWriter);
 			}
 		}
 
@@ -276,7 +276,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
 
 				try
 				{
-					command = (Command) Wireformat.Unmarshal(socketReader);
+					command = (Command) WireFormat.Unmarshal(socketReader);
 				}
 				catch(Exception ex)
 				{
@@ -334,7 +334,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
 			set { this.resumedHandler = value; }
 		}
 
-		public IWireFormat Wireformat
+		public IWireFormat WireFormat
 		{
 			get { return wireformat; }
 			set { wireformat = value; }

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/TransportFilter.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/TransportFilter.cs?rev=1054714&r1=1054713&r2=1054714&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/TransportFilter.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/TransportFilter.cs
Mon Jan  3 19:25:03 2011
@@ -235,7 +235,11 @@ namespace Apache.NMS.ActiveMQ.Transport
 		{
 			next.UpdateURIs(rebalance, updatedURIs);
 		}
-		
+
+        public IWireFormat WireFormat
+        {
+            get { return next.WireFormat; }
+        }
     }
 }
 

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/WireFormatNegotiator.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/WireFormatNegotiator.cs?rev=1054714&r1=1054713&r2=1054714&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/WireFormatNegotiator.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/WireFormatNegotiator.cs
Mon Jan  3 19:25:03 2011
@@ -80,7 +80,7 @@ namespace Apache.NMS.ActiveMQ.Transport
                         throw new IOException("Remote wire format magic is invalid");
                     }
                     wireInfoSentDownLatch.await(negotiateTimeout);
-                    wireFormat.renegotiateWireFormat(info);
+                    wireFormat.RenegotiateWireFormat(info);
                 }
                 catch (Exception e)
                 {



Mime
View raw message