Author: tabish
Date: Tue Jul 27 21:06:24 2010
New Revision: 979873
URL: http://svn.apache.org/viewvc?rev=979873&view=rev
Log:
Fix for: https://issues.apache.org/activemq/browse/AMQNET-266
Modified:
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/Mock/MockTransport.cs
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Mock/MockTransportFactory.cs
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/Transport/failover/FailoverTransportTest.cs
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=979873&r1=979872&r2=979873&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 21:06:24 2010
@@ -740,6 +740,24 @@ namespace Apache.NMS.ActiveMQ.Transport.
"ActiveMQ Failover Worker: " +
this.GetHashCode().ToString());
}
+ if(rebalance)
+ {
+ ITransport transport =
connectedTransport.GetAndSet(null);
+ if(transport != null)
+ {
+ transport.Command = new
CommandHandler(disposedOnCommand);
+ transport.Exception = new
ExceptionHandler(disposedOnException);
+ try
+ {
+ transport.Stop();
+ }
+ catch(Exception ex)
+ {
+ ex.GetType(); // Ignore errors but this lets
us see the error during debugging
+ }
+ }
+ }
+
Tracer.Debug("Waking up reconnect task");
try
{
@@ -1188,7 +1206,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
{
if(copy.Remove(uri) == false)
{
- uriSet.Add(uri);
+ added.Add(uri);
}
}
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=979873&r1=979872&r2=979873&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 21:06:24 2010
@@ -34,7 +34,8 @@ namespace Apache.NMS.ActiveMQ.Transport.
public class MockTransport : ITransport
{
#region Properties
-
+
+ private string name;
private bool failOnSendMessage = false;
private int numSentMessagesBeforeFail = -1;
private int numSentMessages = 0;
@@ -263,6 +264,12 @@ namespace Apache.NMS.ActiveMQ.Transport.
#region Property Accessors
+ public string Name
+ {
+ get { return this.name; }
+ set { this.name = value; }
+ }
+
public CommandHandler Command
{
get { return commandHandler; }
Modified:
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Mock/MockTransportFactory.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Mock/MockTransportFactory.cs?rev=979873&r1=979872&r2=979873&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Mock/MockTransportFactory.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Mock/MockTransportFactory.cs
Tue Jul 27 21:06:24 2010
@@ -74,6 +74,13 @@ namespace Apache.NMS.ActiveMQ.Transport.
set { this.failOnCreate = value; }
}
+ private string name = null;
+ public string Name
+ {
+ get { return this.name; }
+ set { this.name = value; }
+ }
+
#endregion
public ITransport CreateTransport(Uri location)
@@ -108,6 +115,7 @@ namespace Apache.NMS.ActiveMQ.Transport.
transport.NumReceivedMessagesBeforeFail =
this.NumReceivedMessagesBeforeFail;
transport.FailOnSendMessage = this.FailOnSendMessage;
transport.NumSentMessagesBeforeFail =
this.NumSentMessagesBeforeFail;
+ transport.Name = this.Name;
return transport;
}
Modified:
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/Transport/failover/FailoverTransportTest.cs
URL:
http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/Transport/failover/FailoverTransportTest.cs?rev=979873&r1=979872&r2=979873&view=diff
==============================================================================
---
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/Transport/failover/FailoverTransportTest.cs
(original)
+++
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/Transport/failover/FailoverTransportTest.cs
Tue Jul 27 21:06:24 2010
@@ -543,5 +543,52 @@ namespace Apache.NMS.ActiveMQ.Test
transport.Oneway( command );
}
+ [Test]
+ public void TestFailoverTransportConnectionControlHandling()
+ {
+ Uri uri = new
Uri("failover:(mock://localhost:61613)?randomize=false");
+
+ string reconnectTo =
"mock://localhost:61616?transport.name=Reconnected";
+ string connectedBrokers =
"mock://localhost:61616?transport.name=Broker1," +
+
"mock://localhost:61617?transport.name=Broker2";
+
+ ConnectionControl cmd = new ConnectionControl();
+ cmd.FaultTolerant = true;
+ cmd.ReconnectTo = reconnectTo;
+ cmd.ConnectedBrokers = connectedBrokers;
+
+ FailoverTransportFactory factory = new FailoverTransportFactory();
+
+ ITransport transport = factory.CreateTransport( uri );
+ Assert.IsNotNull( transport );
+ transport.Command = new CommandHandler(OnCommand);
+ transport.Exception = new ExceptionHandler(OnException);
+
+ FailoverTransport failover = (FailoverTransport)
transport.Narrow(typeof(FailoverTransport));
+ Assert.IsNotNull(failover);
+ Assert.IsFalse(failover.Randomize);
+
+ transport.Start();
+
+ MockTransport mock = null;
+ while(mock == null ) {
+ mock = (MockTransport) transport.Narrow(typeof(MockTransport));
+ }
+
+ mock.InjectCommand(cmd);
+
+ failover.Remove(true, new Uri[] {new
Uri("mock://localhost:61613")});
+
+ Thread.Sleep(1000);
+
+ mock = null;
+
+ while(mock == null) {
+ mock = (MockTransport) transport.Narrow(typeof(MockTransport));
+ }
+
+ Assert.AreEqual("Reconnected", mock.Name);
+
+ }
}
}
