activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tab...@apache.org
Subject svn commit: r979759 - in /activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport: ./ Failover/ Mock/ Tcp/
Date Tue, 27 Jul 2010 15:58:53 GMT
Author: tabish
Date: Tue Jul 27 15:58:52 2010
New Revision: 979759

URL: http://svn.apache.org/viewvc?rev=979759&view=rev
Log:
https://issues.apache.org/activemq/browse/AMQNET-266

* ITransport.cs:
* TransportFilter.cs:
* Tcp/TcpTransport.cs:
* Mock/MockTransport.cs:
* ICompositeTransport.cs:
* Failover/BackupTransport.cs:
* Failover/FailoverTransport.cs:
* Failover/FailoverTransportFactory.cs: 

Modified:
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/BackupTransport.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/Failover/FailoverTransportFactory.cs
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/ICompositeTransport.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/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

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/BackupTransport.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/BackupTransport.cs?rev=979759&r1=979758&r2=979759&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/BackupTransport.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/BackupTransport.cs
Tue Jul 27 15:58:52 2010
@@ -41,7 +41,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
             this.disposed = true;
             if(failoverTransport != null)
             {
-                this.failoverTransport.Reconnect();
+                this.failoverTransport.Reconnect(false);
             }
         }
 

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=979759&r1=979758&r2=979759&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
Tue Jul 27 15:58:52 2010
@@ -38,6 +38,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
         private bool disposed;
         private bool connected;
         private List<Uri> uris = new List<Uri>();
+		private List<Uri> updated = new List<Uri>();
         private CommandHandler commandHandler;
         private ExceptionHandler exceptionHandler;
         private InterruptedHandler interruptedHandler;
@@ -76,7 +77,9 @@ namespace Apache.NMS.ActiveMQ.Transport.
         private int maxCacheSize = 256;
         private volatile Exception failure;
         private readonly object mutex = new object();
-
+		private bool reconnectSupported = true;
+		private bool updateURIsSupported = true;
+		
         public FailoverTransport()
         {
             id = idCounter++;
@@ -303,7 +306,17 @@ namespace Apache.NMS.ActiveMQ.Transport.
         {
             get { return started; }
         }
-                   
+
+	    public bool IsReconnectSupported
+		{
+			get{ return this.reconnectSupported; }
+		}
+
+	    public bool IsUpdateURIsSupported
+		{
+			get{ return this.updateURIsSupported; }
+		}
+		
         /// <summary>
         /// </summary>
         /// <param name="command"></param>
@@ -399,7 +412,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
                 }
                 else
                 {
-                    Reconnect();
+                    Reconnect(false);
                 }
             }
         }
@@ -499,22 +512,13 @@ namespace Apache.NMS.ActiveMQ.Transport.
 
                 if(!initialized)
                 {
-                    if(command.IsBrokerInfo)
-                    {
-                        BrokerInfo info = (BrokerInfo) command;
-                        BrokerInfo[] peers = info.PeerBrokerInfos;
-                        if(peers != null)
-                        {
-                            for(int i = 0; i < peers.Length; i++)
-                            {
-                                String brokerString = peers[i].BrokerURL;
-                                Add(brokerString);
-                            }
-                        }
-
-                        initialized = true;
-                    }
+                    initialized = true;
                 }
+				
+				if(command.IsConnectionControl)
+				{
+					this.HandleConnectionControl(command as ConnectionControl);
+				}
             }
 
             this.Command(sender, command);
@@ -669,7 +673,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
             }
         }
 
-        public void Add(Uri[] u)
+        public void Add(bool rebalance, Uri[] u)
         {
             lock(uris)
             {
@@ -682,10 +686,10 @@ namespace Apache.NMS.ActiveMQ.Transport.
                 }
             }
 
-            Reconnect();
+            Reconnect(rebalance);
         }
 
-        public void Remove(Uri[] u)
+        public void Remove(bool rebalance, Uri[] u)
         {
             lock(uris)
             {
@@ -695,10 +699,10 @@ namespace Apache.NMS.ActiveMQ.Transport.
                 }
             }
 
-            Reconnect();
+            Reconnect(rebalance);
         }
 
-        public void Add(String u)
+        public void Add(bool rebalance, String u)
         {
             try
             {
@@ -708,10 +712,9 @@ namespace Apache.NMS.ActiveMQ.Transport.
                     if(!uris.Contains(uri))
                     {
                         uris.Add(uri);
+						Reconnect(rebalance);
                     }
-                }
-
-                Reconnect();
+                }				
             }
             catch(Exception e)
             {
@@ -721,10 +724,10 @@ namespace Apache.NMS.ActiveMQ.Transport.
 
         public void Reconnect(Uri uri)
         {
-            Add(new Uri[] { uri });
+            Add(true, new Uri[] { uri });
         }
         
-        public void Reconnect()
+        public void Reconnect(bool rebalance)
         {
             lock(reconnectMutex)
             {
@@ -1161,7 +1164,118 @@ namespace Apache.NMS.ActiveMQ.Transport.
             }
         }
 
-        public void Dispose()
+		
+		public void UpdateURIs(bool rebalance, Uri[] updatedURIs)
+		{		
+	        if(IsUpdateURIsSupported)
+			{
+	            List<Uri> copy = new List<Uri>(this.updated);
+	            List<Uri> added = new List<Uri>();
+				
+	            if(updatedURIs != null && updatedURIs.Length > 0) 
+				{
+	                HashSet<Uri> uriSet = new HashSet<Uri>();
+	                for(int i = 0; i < updatedURIs.Length; i++) 
+					{
+	                    Uri uri = updatedURIs[i];
+	                    if(uri != null) 
+						{
+	                        uriSet.Add(uri);
+	                    }
+	                }
+					
+	                foreach(Uri uri in uriSet)
+					{
+	                    if(copy.Remove(uri) == false) 
+						{
+	                        uriSet.Add(uri);
+	                    }
+	                }
+					
+	                lock(reconnectMutex) 
+					{
+	                    this.updated.Clear();
+	                    this.updated.AddRange(added);
+	                    
+						foreach(Uri uri in copy) 
+						{
+	                        this.uris.Remove(uri);
+	                    }
+	                    
+						this.Add(rebalance, added.ToArray());
+	                }
+	            }
+	        }			
+		}
+
+	    public void HandleConnectionControl(ConnectionControl control) 
+		{
+	        string reconnectStr = control.ReconnectTo;
+	        
+			if(reconnectStr != null) 
+			{
+	            reconnectStr = reconnectStr.Trim();
+	            if(reconnectStr.Length > 0) 
+				{
+	                try 
+					{
+	                    Uri uri = new Uri(reconnectStr);
+	                    if(IsReconnectSupported) 
+						{
+	                        Reconnect(uri);
+	                        Tracer.Info("Reconnected to: " + uri.OriginalString);
+	                    }
+	                } 
+					catch(Exception e) 
+					{
+	                    Tracer.ErrorFormat("Failed to handle ConnectionControl reconnect to
{0}: {1}", reconnectStr, e);
+	                }
+	            }
+	        }
+			
+	        ProcessNewTransports(control.RebalanceConnection, control.ConnectedBrokers);
+	    }
+	
+	    private void ProcessNewTransports(bool rebalance, String newTransports) 
+		{
+	        if(newTransports != null) 
+			{
+	            newTransports = newTransports.Trim();
+				
+	            if(newTransports.Length > 0 && IsUpdateURIsSupported) 
+				{
+	                List<Uri> list = new List<Uri>();
+	                String[] tokens = newTransports.Split(new Char []{','});
+	                
+					foreach(String str in tokens)
+					{
+	                    try 
+						{
+	                        Uri uri = new Uri(str);
+	                        list.Add(uri);
+	                    } 
+						catch 
+						{
+	                        Tracer.Error("Failed to parse broker address: " + str);
+	                    }
+	                }
+					
+	                if(list.Count != 0)
+					{
+	                    try 
+						{
+	                        UpdateURIs(rebalance, list.ToArray());
+	                    } 
+						catch
+						{
+	                        Tracer.Error("Failed to update transport URI's from: " + newTransports);
+	                    }
+	                }
+	            }
+	        }
+	    }
+		
+		public void Dispose()
         {
             Dispose(true);
             GC.SuppressFinalize(this);

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/FailoverTransportFactory.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/FailoverTransportFactory.cs?rev=979759&r1=979758&r2=979759&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/FailoverTransportFactory.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Failover/FailoverTransportFactory.cs
Tue Jul 27 15:58:52 2010
@@ -54,7 +54,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
 		{
 			StringDictionary options = compositData.Parameters;
 			FailoverTransport transport = CreateTransport(options);
-			transport.Add(compositData.Components);
+			transport.Add(false, compositData.Components);
 			return transport;
 		}
 

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/ICompositeTransport.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/ICompositeTransport.cs?rev=979759&r1=979758&r2=979759&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/ICompositeTransport.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/ICompositeTransport.cs
Tue Jul 27 15:58:52 2010
@@ -21,8 +21,29 @@ namespace Apache.NMS.ActiveMQ.Transport
 {
 	public interface ICompositeTransport : ITransport
 	{
-		void Add(Uri[] uris);
-		void Remove(Uri[] uris);
+		/// <summary>
+		/// Adds a new set of Uris to the list of Uris that this Transport can connect to. 
+		/// </summary>
+		/// <param name="rebalance">
+		/// A <see cref="System.Boolean"/>
+		/// Should the current connection be broken and a new one created.
+		/// </param>
+		/// <param name="uris">
+		/// A <see cref="Uri[]"/>
+		/// </param>
+		void Add(bool rebalance, Uri[] uris);
+
+		/// <summary>
+		/// Remove the given Uris from this Transports list of known Uris. 
+		/// </summary>
+		/// <param name="rebalance">
+		/// A <see cref="System.Boolean"/>
+		/// Should the current connection be broken and a new one created.
+		/// </param>
+		/// <param name="uris">
+		/// A <see cref="Uri[]"/>
+		/// </param>
+		void Remove(bool rebalance, Uri[] uris);
 	}
 }
 

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=979759&r1=979758&r2=979759&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
Tue Jul 27 15:58:52 2010
@@ -156,6 +156,36 @@ namespace Apache.NMS.ActiveMQ.Transport
             get;
         }
         
+		/// <summary>
+		/// Returns true if this Transport supports reconnections.
+		/// </summary>
+	    bool IsReconnectSupported
+		{
+			get;
+		}
+	    
+		/// <summary>
+		/// Returns true if this Transport can accept updated lists of connection Uri's.
+		/// </summary>
+	    bool IsUpdateURIsSupported
+		{
+			get;
+		}
+		
+		/// <summary>
+		/// Updates the Uri's that this Transport is aware of and will use to
+		/// connect itself to.  If the rebalance option is true this method will
+		/// terminate any current connection and reconnect to another available
+		/// Uri.
+		/// </summary>
+		/// <param name="rebalance">
+		/// A <see cref="System.Boolean"/>
+		/// </param>
+		/// <param name="updatedURIs">
+		/// A <see cref="Uri[]"/>
+		/// </param>
+		void UpdateURIs(bool rebalance, Uri[] updatedURIs);
+		
 	}
 }
 

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=979759&r1=979758&r2=979759&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
Tue Jul 27 15:58:52 2010
@@ -372,6 +372,21 @@ namespace Apache.NMS.ActiveMQ.Transport.
             get{ return new Uri("mock://mock"); }
         }
         
+	    public bool IsReconnectSupported
+		{
+			get{ return false; }
+		}
+	    
+	    public bool IsUpdateURIsSupported
+		{
+			get{ return false; }
+		}
+		
+		public void UpdateURIs(bool rebalance, Uri[] updatedURIs)
+		{
+			throw new IOException();
+		}
+		
         #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=979759&r1=979758&r2=979759&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
Tue Jul 27 15:58:52 2010
@@ -365,6 +365,20 @@ namespace Apache.NMS.ActiveMQ.Transport.
 			return null;
 		}
 
+	    public bool IsReconnectSupported
+		{
+			get{ return false; }
+		}
+	    
+	    public bool IsUpdateURIsSupported
+		{
+			get{ return false; }
+		}
+		
+		public void UpdateURIs(bool rebalance, Uri[] updatedURIs)
+		{
+			throw new IOException();
+		}		
 	}
 }
 

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=979759&r1=979758&r2=979759&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
Tue Jul 27 15:58:52 2010
@@ -217,6 +217,22 @@ namespace Apache.NMS.ActiveMQ.Transport
         {
             get{ return next.RemoteAddress; }
         }
+		
+	    public bool IsReconnectSupported
+		{
+			get{ return next.IsReconnectSupported; }
+		}
+	    
+	    public bool IsUpdateURIsSupported
+		{
+			get{ return next.IsUpdateURIsSupported; }
+		}
+		
+		public void UpdateURIs(bool rebalance, Uri[] updatedURIs)
+		{
+			next.UpdateURIs(rebalance, updatedURIs);
+		}
+		
     }
 }
 



Mime
View raw message