Spring/spring-net
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add reconnection logic to SimpleMessageListenerContainer.

Browse Source 
Test for SimpleMessageListenerContainer's ExceptionListener being invoked (SPRNET-542)
This commit is contained in:
markpollack
2008-08-11 21:20:03 +00:00
parent b867ebea34
commit a79d4769dc
12 changed files with 465 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Config/MessageListenerContainerObjectDefinitionParser.cs
View File

@@ -21,6 +21,7 @@
using System;
using System.Xml;
using Apache.NMS;
using Spring.Core.TypeConversion;
using Spring.Messaging.Nms.Listener;
using Spring.Messaging.Nms.Listener.Adapter;
using Spring.Objects.Factory.Config;
@@ -82,8 +83,12 @@ namespace Spring.Messaging.Nms.Config
private readonly string CONCURRENCY_ATTRIBUTE = "concurrency";
private readonly string CONNECTION_FACTORY_ATTRIBUTE = "connection-factory";
private readonly string RECOVERY_INTERVAL_ATTRIBUTE = "recovery-interval";
private readonly string MAX_RECOVERY_INTERVAL_ATTRIBUTE = "max-recovery-interval";
private readonly string CONNECTION_FACTORY_ATTRIBUTE = "connection-factory";
#endregion
#region IObjectDefinitionParser Members
@@ -231,6 +236,9 @@ namespace Spring.Messaging.Nms.Config
{
containerDef.AddPropertyValue("ConcurrentConsumers", concurrency[1]);
}
containerDef.AddPropertyValue("RecoveryInterval", ParseRecoveryInterval(containerElement, parserContext));
containerDef.AddPropertyValue("MaxRecoveryTime", ParseMaxRecoveryTime(containerElement, parserContext));
return containerDef;
}
@@ -346,7 +354,7 @@ namespace Spring.Messaging.Nms.Config
private int[] ParseConcurrency(XmlElement ele, ParserContext parserContext)
{
String concurrency = ele.GetAttribute(CONCURRENCY_ATTRIBUTE);
string concurrency = ele.GetAttribute(CONCURRENCY_ATTRIBUTE);
if (!StringUtils.HasText(concurrency))
{
return null;
@@ -363,5 +371,43 @@ namespace Spring.Messaging.Nms.Config
return null;
}
}
private TimeSpan ParseRecoveryInterval(XmlElement ele, ParserContext parserContext)
{
string recoveryInterval = ele.GetAttribute(RECOVERY_INTERVAL_ATTRIBUTE);
if (!StringUtils.HasText(recoveryInterval))
{
return SimpleMessageListenerContainer.DEFAULT_RECOVERY_INTERVAL;
}
try
{
TimeSpanConverter tsc = new TimeSpanConverter();
return (TimeSpan)tsc.ConvertFrom(recoveryInterval);
} catch (Exception ex)
{
parserContext.ReaderContext.ReportException(ele, RECOVERY_INTERVAL_ATTRIBUTE,
"Invalid recovery-interval value [" + recoveryInterval + "]", ex);
return SimpleMessageListenerContainer.DEFAULT_RECOVERY_INTERVAL;
}
}
private TimeSpan ParseMaxRecoveryTime(XmlElement ele, ParserContext parserContext)
{
string recoverTime = ele.GetAttribute(MAX_RECOVERY_INTERVAL_ATTRIBUTE);
if (!StringUtils.HasText(recoverTime))
{
return SimpleMessageListenerContainer.DEFAULT_MAX_RECOVERY_TIME;
}
try
{
TimeSpanConverter tsc = new TimeSpanConverter();
return (TimeSpan)tsc.ConvertFrom(recoverTime);
}
catch (Exception ex)
{
parserContext.ReaderContext.ReportException(ele, MAX_RECOVERY_INTERVAL_ATTRIBUTE,
"Invalid max-recovery-time value [" + recoverTime + "]", ex);
return SimpleMessageListenerContainer.DEFAULT_MAX_RECOVERY_TIME;
}
}
}
}

19
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Config/spring-nms-1.2.xsd
View File

@@ -114,6 +114,25 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="recovery-interval" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
The TimeSpan interval to sleep in between connection recovery attemtps. Default is 5 seconds.
]]>
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:attribute>
<xsd:attribute name="max-recovery-time" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
The TimeSpan interval in which connection recovery attempts will be made. Default is 10 minutes.
]]>
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

54
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/SingleConnectionFactory.cs
View File

@@ -35,11 +35,15 @@ namespace Spring.Messaging.Nms.Connections
/// details.
/// </summary>
/// <remarks>
/// <para>
/// You can either pass in a specific Connection directly or let this
/// factory lazily create a Connection via a given target ConnectionFactory.
/// <para>Useful in order to keep using the same Connection for multiple
/// <see cref="NmsTemplate"/> calls, without having a pooling ConnectionFactory
/// underneath. This may span any number of transactions, even concurrently executing transactions.
/// </para>
/// <para>
/// Useful for testing and in applications when you want to keep using the
/// same Connection for multiple <see cref="NmsTemplate"/>
/// calls, without having a pooling ConnectionFactory underneath. This may span
/// any number of transactions, even concurrently executing transactions.
/// </para>
/// <para>
/// Note that Spring's message listener containers support the use of
@@ -84,6 +88,7 @@ namespace Spring.Messaging.Nms.Connections
/// </summary>
private object connectionMonitor = new object();
#endregion
#region Constructors
@@ -180,7 +185,7 @@ namespace Spring.Messaging.Nms.Connections
/// </para>
/// </remarks>
/// <value>
/// <c>true</c> if [reconnect on exception]; otherwise, <c>false</c>.
/// <c>true</c> attempt to reconnect on exception during next access; otherwise, <c>false</c>.
/// </value>
public bool ReconnectOnException
{
@@ -275,16 +280,10 @@ namespace Spring.Messaging.Nms.Connections
IExceptionListener listenerToUse = ExceptionListener;
if (ReconnectOnException)
{
InternalChainedExceptionListener chained = new InternalChainedExceptionListener(this, listenerToUse);
con.ExceptionListener += chained.OnException;
}
else
{
if (ExceptionListener != null)
{
con.ExceptionListener += ExceptionListener.OnException;
}
//add reconnect exception handler first.
con.ExceptionListener += new ExceptionListener(this.OnException);
}
con.ExceptionListener += new ExceptionListener(listenerToUse.OnException);
}
}
@@ -389,25 +388,6 @@ namespace Spring.Messaging.Nms.Connections
}
}
internal class InternalChainedExceptionListener : ChainedExceptionListener, IExceptionListener
{
private IExceptionListener userListener;
public InternalChainedExceptionListener(IExceptionListener internalListener, IExceptionListener userListener)
{
AddListener(internalListener);
if (userListener != null)
{
AddListener(userListener);
this.userListener = userListener;
}
}
public IExceptionListener UserListener
{
get { return userListener; }
}
}
internal class CloseSupressingConnection : IConnection
{
private IConnection target;
@@ -432,7 +412,8 @@ namespace Spring.Messaging.Nms.Connections
else
{
throw new ArgumentException(
"Setting of 'ClientID' property not supported on wrapper for shared Connection." +
"Setting of 'ClientID' property not supported on wrapper for shared Connection since" +
"this is a shared connection that may serve any number of clients concurrently." +
"Set the 'ClientId' property on the SingleConnectionFactory instead.");
}
@@ -446,7 +427,7 @@ namespace Spring.Messaging.Nms.Connections
public void Stop()
{
//don't pass the call to the target.
//don't pass the call to the target as it would stop receiving for all clients sharing this connection.
}
public ISession CreateSession()
@@ -473,7 +454,10 @@ namespace Spring.Messaging.Nms.Connections
{
target.ExceptionListener += value;
}
remove { target.ExceptionListener -= value; }
remove
{
target.ExceptionListener -= value;
}
}

20
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractListenerContainer.cs
View File

@@ -435,24 +435,6 @@ namespace Spring.Messaging.Nms.Listener
sharedConnection.Start();
}
}
/*
bool running = IsRunning;
lock (this.sharedConnectionMonitor)
{
MessagingUtils.CloseConnection(this.sharedConnection, running);
IConnection con = CreateConnection();
try
{
PrepareSharedConnection(con);
}
catch (Exception)
{
MessagingUtils.CloseConnection(con);
throw;
}
this.sharedConnection = con;
}*/
}
/// <summary>
@@ -471,7 +453,7 @@ namespace Spring.Messaging.Nms.Listener
{
PrepareSharedConnection(con);
return con;
} catch (NMSException)
} catch (Exception)
{
NmsUtils.CloseConnection(con);
throw;

5
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractMessageListenerContainer.cs
View File

@@ -19,6 +19,8 @@
#endregion
using System;
using System.Collections;
using System.IO;
using Common.Logging;
using Spring.Messaging.Nms.Core;
using Spring.Messaging.Nms.Support;
@@ -261,7 +263,6 @@ namespace Spring.Messaging.Nms.Listener
set { acceptMessagesWhileStopping = value; }
}
#endregion
@@ -538,7 +539,7 @@ namespace Spring.Messaging.Nms.Listener
}
if (ex is NMSException)
{
InvokeExceptionListener((NMSException)ex);
InvokeExceptionListener(ex);
}
if (Active)
{

53
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/RecoveryTimeExceededException.cs Normal file
View File

@@ -0,0 +1,53 @@
#region License
/*
* Copyright 2002-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#endregion
using System;
using Apache.NMS;
namespace Spring.Messaging.Nms.Listener
{
/// <summary>
/// Exception thrown when the maximum connection recovery time has been exceeded.
/// </summary>
/// <author>Mark Pollack</author>
public class RecoveryTimeExceededException : NMSException
{
/// <summary>
/// Initializes a new instance of the <see cref="RecoveryTimeExceededException"/> class, with the specified message
/// </summary>
/// <param name="message">The message.</param>
public RecoveryTimeExceededException(string message)
: base(message)
{
}
/// <summary>
/// Initializes a new instance of the <see cref="RecoveryTimeExceededException"/> class, with the specified message
/// and root cause exception
/// </summary>
/// <param name="message">The message.</param>
/// <param name="innerException">The inner exception.</param>
public RecoveryTimeExceededException(string message, Exception innerException)
: base(message, innerException)
{
}
}
}

101
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/SimpleMessageListenerContainer.cs
View File

@@ -19,6 +19,7 @@
#endregion
using System;
using System.Threading;
using Common.Logging;
using Spring.Collections;
using Spring.Messaging.Nms.Core;
@@ -44,6 +45,16 @@ namespace Spring.Messaging.Nms.Listener
#region fields
/// <summary>
/// The default recovery time interval between connection reconnection attempts
/// </summary>
public static TimeSpan DEFAULT_RECOVERY_INTERVAL = new TimeSpan(0,0,0,5,0);
/// <summary>
/// The total time connection recovery will be attempted.
/// </summary>
public static TimeSpan DEFAULT_MAX_RECOVERY_TIME = new TimeSpan(0, 0, 10, 0, 0);
private bool pubSubNoLocal = false;
private int concurrentConsumers = 1;
@@ -54,6 +65,11 @@ namespace Spring.Messaging.Nms.Listener
private object consumersMonitor = new object();
private TimeSpan recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
private TimeSpan maxRecoveryTime = DEFAULT_MAX_RECOVERY_TIME;
#endregion
#region Properties
@@ -92,6 +108,26 @@ namespace Spring.Messaging.Nms.Listener
}
}
/// <summary>
/// Sets the time interval between connection recovery attempts. The default is 5 seconds.
/// </summary>
/// <value>The recovery interval.</value>
public TimeSpan RecoveryInterval
{
set { recoveryInterval = value; }
}
/// <summary>
/// Sets the max recovery time to try reconnection attempts. The default is 10 minutes.
/// </summary>
/// <value>The max recovery time.</value>
public TimeSpan MaxRecoveryTime
{
set { maxRecoveryTime = value; }
}
/// <summary>
/// Always use a shared NMS connection
/// </summary>
@@ -143,7 +179,7 @@ namespace Spring.Messaging.Nms.Listener
protected override void PrepareSharedConnection(IConnection connection)
{
base.PrepareSharedConnection(connection);
connection.ExceptionListener += OnException;
connection.ExceptionListener += new ExceptionListener(OnException);
}
@@ -156,7 +192,7 @@ namespace Spring.Messaging.Nms.Listener
public void OnException(Exception exception)
{
// First invoke the user-specific ExceptionListener, if any.
InvokeExceptionListener(exception);
//InvokeExceptionListener(exception);
// now try to recover the shared Connection and all consumers...
if (logger.IsInfoEnabled)
{
@@ -164,21 +200,74 @@ namespace Spring.Messaging.Nms.Listener
}
try
{
lock(consumersMonitor)
lock (consumersMonitor)
{
sessions = null;
consumers = null;
}
RefreshSharedConnection();
RefreshConnectionUntilSuccessful();
InitializeConsumers();
logger.Info("Successfully refreshed NMS Connection");
} catch (NMSException recoverEx)
} catch (RecoveryTimeExceededException)
{
throw;
} catch (Exception recoverEx)
{
logger.Debug("Failed to recover NMS Connection", recoverEx);
logger.Error("Encountered non-recoverable NMSException", exception);
logger.Error("Encountered non-recoverable Exception", exception);
throw;
}
}
/// <summary>
/// Refresh the underlying Connection, not returning before an attempt has been
/// successful. Called in case of a shared Connection as well as without shared
/// Connection, so either needs to operate on the shared Connection or on a
/// temporary Connection that just gets established for validation purposes.
/// </summary>
/// <remarks>
/// The default implementation retries until it successfully established a
/// Connection, for as long as this message listener container is active.
/// Applies the specified recovery interval between retries.
/// </remarks>
protected virtual void RefreshConnectionUntilSuccessful()
{
TimeSpan totalTryTime = new TimeSpan();
while (IsRunning)
{
try
{
RefreshSharedConnection();
break;
} catch (Exception ex)
{
if (logger.IsInfoEnabled)
{
logger.Info("Could not refresh Connection - retrying in " + recoveryInterval, ex);
}
}
if (totalTryTime > maxRecoveryTime)
{
logger.Info("Could not refresh Connection after " + totalTryTime + ". Stopping reconnection attempts.");
throw new RecoveryTimeExceededException("Could not recover after " + totalTryTime);
}
DateTime startTime = DateTime.Now;
SleepInBetweenRecoveryAttempts();
TimeSpan sleepTimeSpan = DateTime.Now - startTime;
totalTryTime += sleepTimeSpan;
}
}
/// <summary>
/// The amount of time to sleep in between recovery attempts.
/// </summary>
protected virtual void SleepInBetweenRecoveryAttempts()
{
Thread.Sleep(recoveryInterval);
}
/// <summary>
/// Initialize the Sessions and MessageConsumers for this container.
/// </summary>

1
src/Spring/Spring.Messaging.Nms/Spring.Messaging.Nms.2005.csproj
View File

@@ -79,6 +79,7 @@
<Compile Include="Messaging\Nms\Listener\Adapter\MessageListenerAdapter.cs" />
<Compile Include="Messaging\Nms\Listener\ISessionAwareMessageListener.cs" />
<Compile Include="Messaging\Nms\Listener\LocallyExposedNmsResourceHolder.cs" />
<Compile Include="Messaging\Nms\Listener\RecoveryTimeExceededException.cs" />
<Compile Include="Messaging\Nms\Listener\SimpleMessageListenerContainer.cs" />
<Compile Include="Messaging\Nms\Support\Converter\IMessageConverter.cs" />
<Compile Include="Messaging\Nms\Support\Converter\MessageConversionException.cs" />

159
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Core/SimpleMessageListenerContainerTests.cs Normal file
View File

@@ -0,0 +1,159 @@
#region License
/*
* Copyright <20> 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#endregion
#region Imports
using System;
using Apache.NMS;
using NUnit.Framework;
using Rhino.Mocks;
using Spring.Messaging.Nms.Listener;
#endregion
namespace Spring.Messaging.Nms.Core
{
/// <summary>
/// This class contains tests for
/// </summary>
/// <author>Mark Pollack</author>
/// <version>$Id:$</version>
[TestFixture]
public class SimpleMessageListenerContainerTests
{
private static string DESTINATION_NAME = "foo";
private static StubQueue QUEUE_DESTINATION = new StubQueue();
private static string EXCEPTION_MESSAGE = "This.Is.It";
private SimpleMessageListenerContainer container;
private MockRepository mocks;
[SetUp]
public void Setup()
{
mocks = new MockRepository();
container = new SimpleMessageListenerContainer();
}
[Test]
public void RegisteredExceptionListenerIsInvokedOnException()
{
SimpleMessageConsumer messageConsumer = new SimpleMessageConsumer();
ISession session = (ISession) mocks.CreateMock(typeof (ISession));
Expect.Call(session.GetQueue(DESTINATION_NAME)).Return(QUEUE_DESTINATION);
Expect.Call(session.CreateConsumer(QUEUE_DESTINATION, null)).Return(messageConsumer);
// an exception is thrown, so the rollback logic is being applied here...
Expect.Call(session.Transacted).Return(false);
IConnection connection = (IConnection)mocks.CreateMock(typeof(IConnection));
connection.ExceptionListener += container.OnException;
Expect.Call(connection.CreateSession(container.SessionAcknowledgeMode)).Return(session);
connection.Start();
IConnectionFactory connectionFactory = (IConnectionFactory) mocks.CreateMock(typeof (IConnectionFactory));
Expect.Call(connectionFactory.CreateConnection()).Return(connection);
NMSException theException = new NMSException(EXCEPTION_MESSAGE);
IExceptionListener exceptionListener = (IExceptionListener) mocks.CreateMock(typeof (IExceptionListener));
exceptionListener.OnException(theException);
IMessage message = (IMessage) mocks.CreateMock(typeof (IMessage));
mocks.ReplayAll();
container.ConnectionFactory = connectionFactory;
container.DestinationName = DESTINATION_NAME;
container.MessageListener = new BadSessionAwareMessageListener(theException);
container.ExceptionListener = exceptionListener;
container.AfterPropertiesSet();
// manually trigger an Exception with the above bad MessageListener...
messageConsumer.SendMessage(message);
mocks.VerifyAll();
}
}
internal class BadSessionAwareMessageListener : ISessionAwareMessageListener
{
private NMSException exception;
public BadSessionAwareMessageListener(NMSException exception)
{
this.exception = exception;
}
public void OnMessage(IMessage message, ISession session)
{
throw exception;
}
}
internal class SimpleMessageConsumer : IMessageConsumer
{
public event MessageListener Listener;
public void SendMessage(IMessage message)
{
Listener(message);
}
public IMessage Receive()
{
throw new NotImplementedException();
}
public IMessage Receive(TimeSpan timeout)
{
throw new NotImplementedException();
}
public IMessage ReceiveNoWait()
{
throw new NotImplementedException();
}
public void Close()
{
throw new NotImplementedException();
}
public void Dispose()
{
throw new NotImplementedException();
}
}
}

2
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListenerContainerTests.cs
View File

@@ -32,7 +32,7 @@ using Spring.Testing.NUnit;
namespace Spring.Messaging.Nms.Integration
{
/// <summary>
/// This class contains tests for
/// This class contains integration tests for the SimpleMessageListenerContainer
/// </summary>
/// <author>Mark Pollack</author>
/// <version>$Id:$</version>

64
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/StubQueue.cs Normal file
View File

@@ -0,0 +1,64 @@
#region License
/*
* Copyright 2002-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#endregion
using Apache.NMS;
namespace Spring.Messaging.Nms
{
/// <summary>
///
/// </summary>
/// <remarks>
///
/// </remarks>
/// <author>Mark Pollack</author>
public class StubQueue : IQueue
{
public static string DEFAULT_QUEUE_NAME = "banjo";
private string queueName = DEFAULT_QUEUE_NAME;
public string QueueName
{
get { return queueName; }
}
public DestinationType DestinationType
{
get { return DestinationType.Queue; }
}
public bool IsTopic
{
get { return false; }
}
public bool IsQueue
{
get { return true; }
}
public bool IsTemporary
{
get { return false; }
}
}
}

2
test/Spring/Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Tests.2005.csproj
View File

@@ -98,10 +98,12 @@
<Compile Include="Messaging\Nms\Connections\TestMessageProducer.cs" />
<Compile Include="Messaging\Nms\Connections\TestSession.cs" />
<Compile Include="Messaging\Nms\Core\MessageTemplateTests.cs" />
<Compile Include="Messaging\Nms\Core\SimpleMessageListenerContainerTests.cs" />
<Compile Include="Messaging\Nms\Integration\LoggingExceptionHandler.cs" />
<Compile Include="Messaging\Nms\Integration\SimpleMessageListener.cs" />
<Compile Include="Messaging\Nms\Integration\SimpleMessageListenerContainerTests.cs" />
<Compile Include="Messaging\Nms\Connections\TestConnectionFactory.cs" />
<Compile Include="Messaging\Nms\StubQueue.cs" />
</ItemGroup>
<ItemGroup>
<EmbeddedResource Include="Messaging\Nms\Integration\SimpleMessageListenerContainerTests.xml" />