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

NMS Development

Browse Source
This commit is contained in:
markpollack
2008-07-23 22:54:02 +00:00
parent d52766af07
commit c467f49035
26 changed files with 1512 additions and 374 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/Spring/Spring.Core/Context/IConfigurableApplicationContext.cs
View File

@@ -57,7 +57,7 @@ namespace Spring.Context
/// <author>Mark Pollack (.NET)</author>
/// <seealso cref="System.IDisposable"/>
/// <seealso cref="Spring.Context.IApplicationContext"/>
public interface IConfigurableApplicationContext : IApplicationContext
public interface IConfigurableApplicationContext : IApplicationContext, ILifecycle
{
/// <summary>
/// Return the internal object factory of this application context.

73
src/Spring/Spring.Core/Context/ILifecycle.cs Normal file
View File

@@ -0,0 +1,73 @@
#region License
/*
* Copyright <20> 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
namespace Spring.Context
{
/// <summary>
/// Interface defining methods for start/stop lifecycle control.
/// The typical use case for this is to control asynchronous processing.
/// </summary>
/// <remarks>
/// <para>
/// Can be implemented by both components (typically a Spring object defined in
/// a spring <see cref="Spring.Objects.Factory.IObjectFactory"/> and containers
/// (typically a spring <see cref="IApplicationContext"/>. Containers will
/// propagate start/stop signals to all components that apply.
/// </para>
/// </remarks>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public interface ILifecycle
{
/// <summary>
/// Starts this component.
/// </summary>
/// <remarks>Should not throw an exception if the component is already running.
/// In the case of a container, this will propagate the start signal
/// to all components that apply.
/// </remarks>
void Start();
/// <summary>
/// Stops this component.
/// </summary>
/// <remarks>
/// Should not throw an exception if the component isn't started yet.
/// In the case of a container, this will propagate the stop signal
/// to all components that apply.
/// </remarks>
void Stop();
/// <summary>
/// Gets a value indicating whether this component is currently running.
/// </summary>
/// <remarks>
/// In the case of a container, this will return <code>true</code>
/// only if <i>all</i> components that apply are currently running.
/// </remarks>
/// <value>
/// <c>true</c> if this component is running; otherwise, <c>false</c>.
/// </value>
bool IsRunning
{
get;
}
}
}

111
src/Spring/Spring.Core/Context/Support/AbstractApplicationContext.cs
View File

@@ -844,8 +844,115 @@ namespace Spring.Context.Support
set { _parentApplicationContext = value; }
}
#endregion
#endregion
#region ILifecycle Members
/// <summary>
/// Starts this component.
/// </summary>
/// <remarks>Should not throw an exception if the component is already running.
/// In the case of a container, this will propagate the start signal
/// to all components that apply.
/// </remarks>
public void Start()
{
IDictionary lifecycleObjects = LifeCycleObjects;
foreach (DictionaryEntry dictionaryEntry in lifecycleObjects)
{
//TODO start dependencies of the lifecycle objects
ILifecycle obj = dictionaryEntry.Value as ILifecycle;
if (obj != null)
{
if (!obj.IsRunning)
{
obj.Start();
}
}
}
}
/// <summary>
/// Stops this component.
/// </summary>
/// <remarks>
/// Should not throw an exception if the component isn't started yet.
/// In the case of a container, this will propagate the stop signal
/// to all components that apply.
/// </remarks>
public void Stop()
{
IDictionary lifecycleObjects = LifeCycleObjects;
foreach (DictionaryEntry dictionaryEntry in lifecycleObjects)
{
//TODO stop dependencies of the lifecycle objects
ILifecycle obj = dictionaryEntry.Value as ILifecycle;
if (obj != null)
{
if (obj.IsRunning)
{
obj.Stop();
}
}
}
}
/// <summary>
/// Gets a value indicating whether this component is currently running.
/// </summary>
/// <value>
/// <c>true</c> if this component is running; otherwise, <c>false</c>.
/// </value>
/// <remarks>
/// In the case of a container, this will return <code>true</code>
/// only if <i>all</i> components that apply are currently running.
/// </remarks>
public bool IsRunning
{
get
{
IDictionary lifecycleObjects = LifeCycleObjects;
foreach (DictionaryEntry dictionaryEntry in lifecycleObjects)
{
ILifecycle obj = dictionaryEntry.Value as ILifecycle;
if (obj != null)
{
if (!obj.IsRunning)
{
return false;
}
}
}
return true;
}
}
/// <summary>
/// Gets a dictionary of all singleton beans that implement the
/// ILifecycle interface in this context.
/// </summary>
/// <value>A dictionary of ILifecycle objects with object name as key.</value>
private IDictionary LifeCycleObjects
{
get
{
IConfigurableListableObjectFactory objectFactory = ObjectFactory;
string[] objectNames = objectFactory.SingletonNames;
IDictionary lifeCycleObjects = new Hashtable();
foreach (string objectName in objectNames)
{
object obj = objectFactory.GetSingleton(objectName);
if (obj is ILifecycle)
{
lifeCycleObjects[objectName] = obj;
}
}
return lifeCycleObjects;
}
}
#endregion
#region IApplicationContext Members
/// <summary>

69
src/Spring/Spring.Core/Objects/Factory/Config/IConfigurableObjectFactory.cs
View File

@@ -49,7 +49,7 @@ namespace Spring.Objects.Factory.Config
/// </remarks>
/// <author>Juergen Hoeller</author>
/// <author>Rick Evans (.NET)</author>
public interface IConfigurableObjectFactory : IHierarchicalObjectFactory
public interface IConfigurableObjectFactory : IHierarchicalObjectFactory, ISingletonObjectRegistry
{
/// <summary>
/// Set the parent of this object factory.
@@ -146,27 +146,6 @@ namespace Spring.Objects.Factory.Config
/// </exception>
void RegisterAlias(string name, string theAlias);
/// <summary>
/// Register the given existing object as singleton in the object factory,
/// under the given object name.
/// </summary>
/// <remarks>
/// <p>
/// Typically invoked during factory configuration, but can also be
/// used for runtime registration of singletons. Therefore, a factory
/// implementation should synchronize singleton access; it will have
/// to do this anyway if it supports lazy initialization of singletons.
/// </p>
/// </remarks>
/// <param name="name">
/// The name of the object.
/// </param>
/// <param name="singleton">The existing object.</param>
/// <exception cref="Spring.Objects.ObjectsException">
/// If the singleton could not be registered.
/// </exception>
void RegisterSingleton(string name, object singleton);
/// <summary>
/// Register the given custom <see cref="System.ComponentModel.TypeConverter"/>
/// for all properties of the given <see cref="System.Type"/>.
@@ -184,51 +163,5 @@ namespace Spring.Objects.Factory.Config
/// </param>
void RegisterCustomConverter(Type requiredType, TypeConverter converter);
/// <summary>
/// Does this object factory contains a singleton instance with the
/// supplied <paramref name="name"/>?
/// </summary>
/// <remarks>
/// <p>
/// Only checks already instantiated singletons; does not return
/// <see langword="true"/> for singleton object definitions that have
/// not been instantiated yet.
/// </p>
/// <p>
/// The main purpose of this method is to check manually registered
/// singletons (<see cref="RegisterSingleton(string, object)"/>). This
/// method can also be used to check whether a singleton defined by an
/// object definition has already been created.
/// </p>
/// <p>
/// To check whether an object factory contains an object definition
/// with a given name, use the
/// <see cref="Spring.Objects.Factory.IListableObjectFactory.ContainsObjectDefinition(string)"/>
/// method. Calling both
/// <see cref="Spring.Objects.Factory.IListableObjectFactory.ContainsObjectDefinition(string)"/>
/// and <see cref="ContainsSingleton(string)"/> definitively answers
/// the question of whether a specific object factory contains a
/// singleton object with the given name.
/// </p>
/// <p>
/// Use the
/// <see cref="Spring.Objects.Factory.IObjectFactory.ContainsObject(string)"/>
/// method for general checks as to whether a factory knows about an
/// object with a given name (regrdless of whether the object in
/// question is a manually registed singleton instance or created by
/// an object definition)... this also has the happy bonus of also
/// checking any ancestor factories.
/// </p>
/// </remarks>
/// <param name="name">
/// The name of the (singleton) object to look for.
/// </param>
/// <returns>
/// <see langword="true"/> if this object factory contains a singleton
/// instance with the given <paramref name="name"/>.
/// </returns>
/// <seealso cref="Spring.Objects.Factory.IObjectFactory.ContainsObject(string)"/>
/// <seealso cref="Spring.Objects.Factory.IListableObjectFactory.ContainsObjectDefinition(string)"/>
bool ContainsSingleton(string name);
}
}

169
src/Spring/Spring.Core/Objects/Factory/Config/ISingletonObjectRegistry.cs Normal file
View File

@@ -0,0 +1,169 @@
#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
namespace Spring.Objects.Factory.Config
{
/// <summary>
/// Interface that defines a registry for shared object instances.
/// </summary>
/// <remarks>
/// Can be implemented by <see cref="Spring.Objects.Factory.IObjectFactory"/>
/// implementations in order to expose their singleton management facility
/// in a uniform manner.
/// <para>
/// The <see cref="IConfigurableObjectFactory"/> interface extends this interface.
/// </para>
/// </remarks>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public interface ISingletonObjectRegistry
{
/// <summary>
/// Registers the given existing object as singleton in the object registry,
/// under the given object name.
/// </summary>
/// <remarks>
/// <para>
/// The given instance is supposed to be fully initialized; the registry
/// will not perform any initialization callbacks (in particular, it won't
/// call IInitializingObject's <code>AfterPropertiesSet</code> method).
/// The given instance will not receive any destruction callbacks
/// (like IDisposable's <code>Dispose</code> method) either.
/// </para>
/// <para>
/// If running within a full IObjectFactory: Register an object definition
/// instead of an existing instance if your object is supposed to receive
/// initialization and/or destruction callbacks.
/// </para>
/// <para>
/// Typically invoked during registry configuration, but can also be used
/// for runtime registration of singletons. As a consequence, a registry
/// implementation should synchronize singleton access; it will have to do
/// this anyway if it supports a BeanFactory's lazy initialization of singletons.
/// </para>
/// </remarks>
/// <param name="objectName">Name of the object.</param>
/// <param name="singletonObject">The singleton object.</param>
/// <see cref="Spring.Objects.Factory.IInitializingObject.AfterPropertiesSet"/>
/// <see cref="Spring.Objects.Factory.Support.IObjectDefinitionRegistry.RegisterObjectDefinition"/>
void RegisterSingleton(string objectName, object singletonObject);
/// <summary>
/// Return the (raw) singleton object registered under the given name.
/// </summary>
/// <remarks>
/// <para>
/// Only checks already instantiated singletons; does not return an Object
/// for singleton object definitions which have not been instantiated yet.
/// </para>
/// <para>
/// The main purpose of this method is to access manually registered singletons
/// <see cref="RegisterSingleton"/>. Can also be used to access a singleton
/// defined by an object definition that already been created, in a raw fashion.
/// </para>
/// </remarks>
/// <param name="objectName">Name of the object to look for.</param>
/// <returns>the registered singleton object, or <code>null</code> if none found</returns>
/// <see cref="IConfigurableListableObjectFactory.GetObjectDefinition(string)"/>
object GetSingleton(string objectName);
/// <summary>
/// Check if this registry contains a singleton instance with the given name.
/// </summary>
/// <remarks>
/// <para>
/// Only checks already instantiated singletons; does not return <code>true</code>
/// for singleton bean definitions which have not been instantiated yet.
/// </para>
/// <para>
/// The main purpose of this method is to check manually registered singletons
/// <see cref="RegisterSingleton"/>. Can also be used to check whether a
/// singleton defined by an object definition has already been created.
/// </para>
/// <para>
/// To check whether an object factory contains an object definition with a given name,
/// use ListableBeanFactory's <code>ContainsObjectDefinition</code>. Calling both
/// <code>ContainsObjectDefinition</code> and <code>ContainsSingleton</code> answers
/// whether a specific object factory contains an own object with the given name.
/// </para>
/// <para>
/// Use IObjectFactory's <code>ContainsObject</code> for general checks whether the
/// factory knows about an object with a given name (whether manually registered singleton
/// instance or created by bean definition), also checking ancestor factories.
/// </para>
/// </remarks>
/// <param name="objectName">Name of the object to look for.</param>
/// <returns>
/// <c>true</c> if this bean factory contains a singleton instance with the given name; otherwise, <c>false</c>.
/// </returns>
/// <see cref="RegisterSingleton"/>
/// <see cref="Spring.Objects.Factory.IListableObjectFactory.ContainsObjectDefinition"/>
/// <see cref="Spring.Objects.Factory.IObjectFactory.ContainsObject"/>
bool ContainsSingleton(string objectName);
/// <summary>
/// Gets the names of singleton objects registered in this registry.
/// </summary>
/// <remarks>
/// <para>
/// Only checks already instantiated singletons; does not return names
/// for singleton bean definitions which have not been instantiated yet.
/// </para>
/// <para>
/// The main purpose of this method is to check manually registered singletons
/// <see cref="RegisterSingleton"/>. Can also be used to check which
/// singletons defined by an object definition have already been created.
/// </para>
/// </remarks>
/// <value>The list of names as String array (never <code>null</code>).</value>
/// <see cref="RegisterSingleton"/>
/// <see cref="Spring.Objects.Factory.Support.IObjectDefinitionRegistry.GetObjectDefinitionNames"/>
/// <see cref="Spring.Objects.Factory.IListableObjectFactory.GetObjectDefinitionNames"/>
string[] SingletonNames
{
get;
}
/// <summary>
/// Gets the number of singleton beans registered in this registry.
/// </summary>
/// <remarks>
/// <para>
/// Only checks already instantiated singletons; does not count
/// singleton object definitions which have not been instantiated yet.
/// </para>
/// <para>
/// The main purpose of this method is to check manually registered singletons
/// <see cref="RegisterSingleton"/>. Can also be used to count the number of
/// singletons defined by an object definition that have already been created.
/// </para>
/// </remarks>
/// <value>The number of singleton objects.</value>
/// <see cref="RegisterSingleton"/>
/// <see cref="Spring.Objects.Factory.Support.IObjectDefinitionRegistry.ObjectDefinitionCount"/>
/// <see cref="Spring.Objects.Factory.IListableObjectFactory.ObjectDefinitionCount"/>
int SingletonCount
{
get;
}
}
}

105
src/Spring/Spring.Core/Objects/Factory/Support/AbstractObjectFactory.cs
View File

@@ -445,6 +445,7 @@ namespace Spring.Objects.Factory.Support
lock (singletonCache)
{
singletonCache[name] = singleton;
registeredSingletons.Add(name);
}
}
@@ -1303,6 +1304,7 @@ namespace Spring.Objects.Factory.Support
{
object tempObject = singletonCache[name];
singletonCache.Remove(name);
registeredSingletons.Remove(name);
object singletonInstance = tempObject;
if (singletonInstance != null)
@@ -1416,6 +1418,12 @@ namespace Spring.Objects.Factory.Support
private bool caseSensitive;
private IDictionary aliasMap;
private IDictionary singletonCache;
/// <summary>
/// Set of registered singletons, containing the bean names in registration order
/// </summary>
private ISet registeredSingletons = new HashedSet();
private IDictionary singletonsInCreation;
#endregion
@@ -1693,18 +1701,7 @@ namespace Spring.Objects.Factory.Support
return GetObjectForInstance(name, instance);
}
/// <summary>
/// Tries to find a cached object for the specified name.
/// </summary>
/// <param name="objectName">Teh object name to look for.</param>
/// <returns>The cached object if found, <see langword="null"/> otherwise.</returns>
protected virtual object GetSingleton(string objectName)
{
lock (singletonCache)
{
return singletonCache[objectName];
}
}
/// <summary>
/// Creates a singleton instance for the specified object name and definition.
@@ -1796,8 +1793,6 @@ namespace Spring.Objects.Factory.Support
#endregion
#region IConfigurableObjectFactory Members
/// <summary>
/// Destroy all cached singletons in this factory.
/// </summary>
@@ -1943,7 +1938,7 @@ namespace Spring.Objects.Factory.Support
/// Register the given existing object as singleton in the object factory,
/// under the given object name.
/// </summary>
/// <seealso cref="Spring.Objects.Factory.Config.IConfigurableObjectFactory.RegisterSingleton"/>.
/// <seealso cref="Spring.Objects.Factory.Config.ISingletonObjectRegistry.RegisterSingleton"/>.
public void RegisterSingleton(string name, object singletonObject)
{
AssertUtils.ArgumentHasText(name, "name", "The singleton object cannot be registered under an empty name.");
@@ -1976,7 +1971,7 @@ namespace Spring.Objects.Factory.Support
/// Does this object factory contains a singleton instance with the
/// supplied <paramref name="name"/>?
/// </summary>
/// <seealso cref="Spring.Objects.Factory.Config.IConfigurableObjectFactory.ContainsSingleton(string)"/>
/// <seealso cref="Spring.Objects.Factory.Config.ISingletonObjectRegistry.ContainsSingleton(string)"/>
public bool ContainsSingleton(string name)
{
AssertUtils.ArgumentHasText(name, "name");
@@ -1986,6 +1981,84 @@ namespace Spring.Objects.Factory.Support
}
}
#region ISingletonObjectRegistry Members
/// <summary>
/// Gets the names of singleton objects registered in this registry.
/// </summary>
/// <value>The list of names as String array (never <code>null</code>).</value>
/// <remarks>
/// <para>
/// Only checks already instantiated singletons; does not return names
/// for singleton bean definitions which have not been instantiated yet.
/// </para>
/// <para>
/// The main purpose of this method is to check manually registered singletons
/// <see cref="RegisterSingleton"/>. Can also be used to check which
/// singletons defined by an object definition have already been created.
/// </para>
/// </remarks>
/// <see cref="RegisterSingleton"/>
/// <see cref="Spring.Objects.Factory.Support.IObjectDefinitionRegistry.GetObjectDefinitionNames"/>
/// <see cref="Spring.Objects.Factory.IListableObjectFactory.GetObjectDefinitionNames"/>
public string[] SingletonNames
{
get
{
lock (singletonCache)
{
return
StringUtils.DelimitedListToStringArray(
StringUtils.CollectionToDelimitedString(registeredSingletons, ","), ",");
}
}
}
/// <summary>
/// Gets the number of singleton beans registered in this registry.
/// </summary>
/// <value>The number of singleton objects.</value>
/// <remarks>
/// <para>
/// Only checks already instantiated singletons; does not count
/// singleton object definitions which have not been instantiated yet.
/// </para>
/// <para>
/// The main purpose of this method is to check manually registered singletons
/// <see cref="RegisterSingleton"/>. Can also be used to count the number of
/// singletons defined by an object definition that have already been created.
/// </para>
/// </remarks>
/// <see cref="RegisterSingleton"/>
/// <see cref="Spring.Objects.Factory.Support.IObjectDefinitionRegistry.ObjectDefinitionCount"/>
/// <see cref="Spring.Objects.Factory.IListableObjectFactory.ObjectDefinitionCount"/>
public int SingletonCount
{
get
{
lock (singletonCache)
{
return registeredSingletons.Count;
}
}
}
/// <summary>
/// Tries to find a cached object for the specified name.
/// </summary>
/// <param name="objectName">Teh object name to look for.</param>
/// <returns>The cached object if found, <see langword="null"/> otherwise.</returns>
public virtual object GetSingleton(string objectName)
{
lock (singletonCache)
{
return singletonCache[objectName];
}
}
#endregion
}
}

2
src/Spring/Spring.Core/Spring.Core.2005.csproj
View File

@@ -177,6 +177,7 @@
<Compile Include="Context\IHierarchicalMessageSource.cs">
<SubType>Code</SubType>
</Compile>
<Compile Include="Context\ILifecycle.cs" />
<Compile Include="Context\IMessageSource.cs">
<SubType>Code</SubType>
</Compile>
@@ -546,6 +547,7 @@
<Compile Include="Objects\Factory\Config\ConnectionStringsVariableSource.cs" />
<Compile Include="Objects\Factory\Config\IConfigurableFactoryObject.cs" />
<Compile Include="Objects\Factory\Config\InstantiationAwareObjectPostProcessorAdapter.cs" />
<Compile Include="Objects\Factory\Config\ISingletonObjectRegistry.cs" />
<Compile Include="Objects\Factory\Config\ObjectDefinitionHolder.cs" />
<Compile Include="Objects\Factory\Config\ObjectDefinitionVisitor.cs" />
<Compile Include="Objects\Factory\Config\ResourceHandlerConfigurer.cs" />

15
src/Spring/Spring.Messaging.Nms/Context/ILifecycle.cs
View File

@@ -1,15 +0,0 @@
namespace Spring.Context
{
interface ILifecycle
{
void Start();
void Stop();
bool IsRunning
{
get;
}
}
}

30
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsResourceHolder.cs
View File

@@ -47,14 +47,15 @@ namespace Spring.Messaging.Nms.Connection
#region Fields
private bool frozen;
private IConnectionFactory connectionFactory;
private bool frozen = false;
private IList connections = new LinkedList();
private IList sessions = new LinkedList();
private IDictionary sessionsPerIConnection = new Hashtable();
private IConnectionFactory connectionFactory;
#endregion
@@ -64,7 +65,30 @@ namespace Spring.Messaging.Nms.Connection
/// <summary> Create a new NmsResourceHolder that is open for resources to be added.</summary>
public NmsResourceHolder()
{
this.frozen = false;
}
/// <summary>
/// Initializes a new instance of the <see cref="NmsResourceHolder"/> class
/// at is open for resources to be added.
/// </summary>
/// <param name="connectionFactory">The connection factory that this
/// resource holder is associated with (may be <code>null</code>)
/// </param>
public NmsResourceHolder(IConnectionFactory connectionFactory)
{
this.connectionFactory = connectionFactory;
}
/// <summary>
/// Initializes a new instance of the <see cref="NmsResourceHolder"/> class for the
/// given Session.
/// </summary>
/// <param name="session">The session.</param>
public NmsResourceHolder(ISession session)
{
AddSession(session);
frozen = true;
}
/// <summary> Create a new NmsResourceHolder for the given NMS resources.</summary>

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

@@ -384,7 +384,21 @@ namespace Spring.Messaging.Nms.Connection
public string ClientId
{
get { return target.ClientId; }
set { target.ClientId = value; }
set
{
string currentClientId = target.ClientId;
if (currentClientId != null && currentClientId.Equals(value))
{
//ok
}
else
{
throw new ArgumentException(
"Setting of 'ClientID' property not supported on wrapper for shared Connection." +
"Set the 'ClientId' property on the SingleConnectionFactory instead.");
}
}
}
public void Dispose()

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

@@ -1,16 +1,46 @@
#region License
/*
* Copyright <20> 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 System.Collections;
using Spring.Context;
using Spring.Messaging.Nms.Connection;
using Common.Logging;
using Spring.Messaging.Nms.Support;
using Spring.Messaging.Nms.Support.IDestinations;
using Spring.Util;
using Apache.NMS;
namespace Spring.Messaging.Nms.Listener
{
/// <summary>
/// Abstract base class for message listener containers. Can either host
/// a standard NMS <see cref="IMessageListener"/> or a Spring-specific
/// <see cref="ISessionAwareMessageListener"/>
/// </summary>
public abstract class AbstractMessageListenerContainer : AbstractNmsListeningContainer
{
#region Logging
private readonly ILog logger = LogManager.GetLogger(typeof(AbstractMessageListenerContainer));
#endregion
#region Fields
private object destination;
private String messageSelector;
@@ -21,14 +51,13 @@ namespace Spring.Messaging.Nms.Listener
private string durableSubscriptionName;
private ExceptionListener exceptionListener;
private IExceptionListener exceptionListener;
private bool exposeListenerISession = true;
private bool acceptMessagesWhileStopping = false;
private IList pausedTasks = new Spring.Collections.LinkedList();
#endregion
#region Properties
@@ -73,6 +102,17 @@ namespace Spring.Messaging.Nms.Listener
}
/// <summary>
/// Gets or sets the message listener to register.
/// </summary>
///
/// <remarks>
/// <para>
/// This can be either a standard NMS <see cref="IMessageListener"/> object or a
/// Spring <see cref="ISessionAwareMessageListener"/> object.
/// </para>
/// </remarks>
/// <value>The message listener.</value>
public object MessageListener
{
set
@@ -113,7 +153,7 @@ namespace Spring.Messaging.Nms.Listener
}
public ExceptionListener ExceptionListener
public IExceptionListener ExceptionListener
{
get { return exceptionListener; }
set { exceptionListener = value; }
@@ -127,6 +167,35 @@ namespace Spring.Messaging.Nms.Listener
}
/// <summary>
/// Gets or sets a value indicating whether to accept messages while
/// the listener container is in the process of stopping.
/// </summary>
/// <remarks>
/// <para>
/// Return whether to accept received messages while the listener container
/// receive attempt. Switch this flag on to fully process such messages
/// even in the stopping phase, with the drawback that even newly sent
/// messages might still get processed (if coming in before all receive
/// timeouts have expired).
/// </para>
/// <para>
/// Aborting receive attempts for such incoming messages
/// might lead to the provider's retry count decreasing for the affected
/// messages. If you have a high number of concurrent consumers, make sure
/// that the number of retries is higher than the number of consumers,
/// to be on the safe side for all potential stopping scenarios.
/// </para>
/// </remarks>
/// <value>
/// <c>true</c> if accept messages while in the process of stopping; otherwise, <c>false</c>.
/// </value>
public bool AcceptMessagesWhileStopping
{
get { return acceptMessagesWhileStopping; }
set { acceptMessagesWhileStopping = value; }
}
public object LifecycleMonitor
{
get { return lifecycleMonitor; }
@@ -139,23 +208,69 @@ namespace Spring.Messaging.Nms.Listener
protected override void ValidateConfiguration()
{
if (this.destination == null)
{
throw new ArgumentException("Property 'destination' or 'DestinationName' is required");
}
if (SubscriptionDurable && !PubSubDomain)
{
throw new ArgumentException("A durable subscription requires a topic (pub-sub domain)");
}
}
#region Template methods for listeners
/// <summary>
/// Executes the specified listener,
/// committing or rolling back the transaction afterwards (if necessary).
/// </summary>
/// <param name="session">The session to operate on.</param>
/// <param name="message">The received message.</param>
/// <see cref="InvokeListener"/>
/// <see cref="CommitIfNecessary"/>
/// <see cref="RollbackOnExceptionIfNecessary"/>
/// <see cref="HandleListenerException"/>
public virtual void ExecuteListener(ISession session, IMessage message)
{
try
{
DoExecuteListener(session, message);
}
//UPGRADE_NOTE: Exception 'java.lang.Throwable' was converted to 'System.Exception' which has different behavior. "ms-help://MS.VSCC.v80/dv_commoner/local/redirect.htm?index='!DefaultContextWindowIndex'&keyword='jlca1100'"
catch (System.Exception ex)
catch (Exception ex)
{
HandleListenerException(ex);
}
}
/// <summary>
/// Executes the specified listener,
/// committing or rolling back the transaction afterwards (if necessary).
/// </summary>
/// <param name="session">The session to operate on.</param>
/// <param name="message">The received message.</param>
/// <exception cref="NMSException">If thrown by NMS API methods.</exception>
/// <see cref="InvokeListener"/>
/// <see cref="CommitIfNecessary"/>
/// <see cref="RollbackOnExceptionIfNecessary"/>
protected virtual void DoExecuteListener(ISession session, IMessage message)
{
if (!AcceptMessagesWhileStopping && !IsRunning)
{
#region Logging
if (logger.IsWarnEnabled)
{
logger.Warn("Rejecting received message because of the listener container " +
"having been stopped in the meantime: " + message);
}
#endregion
RollbackIfNecessary(session);
throw new MessageRejectedWhileStoppingException();
}
try
{
InvokeListener(session, message);
@@ -168,51 +283,35 @@ namespace Spring.Messaging.Nms.Listener
CommitIfNecessary(session, message);
}
private void CommitIfNecessary(ISession session, IMessage message)
/// <summary>
/// Invokes the specified listener: either as standard NMS IMessageListener
/// or (preferably) as Spring SessionAwareMessageListener.
/// </summary>
/// <param name="session">The session to operate on.</param>
/// <param name="message">The received message.</param>
/// <exception cref="NMSException">If thrown by NMS API methods.</exception>
/// <see cref="MessageListener"/>
protected virtual void InvokeListener(ISession session, IMessage message)
{
//TODO
//logger.Info("CommitIfNecessary not implemented");
}
private void RollbackOnExceptionIfNecessary(ISession session, Exception ex)
{
//TODO
//logger.Info("RollbackOnExceptionIfNecessary not implemented");
}
protected internal virtual void InvokeListener(ISession session, IMessage message)
{
if (MessageListener is ISessionAwareMessageListener)
object listener = MessageListener;
if (listener is ISessionAwareMessageListener)
{
DoInvokeListener((ISessionAwareMessageListener) MessageListener, session, message);
DoInvokeListener((ISessionAwareMessageListener) listener, session, message);
}
else if (listener is IMessageListener)
{
DoInvokeListener((IMessageListener) listener, message);
}
else if (listener != null)
{
throw new ArgumentException("Only IMessageListener and ISessionAwareMessageListener supported");
}
else
{
if (MessageListener is IMessageListener)
{
DoInvokeListener((IMessageListener) MessageListener, message);
}
else
{
throw new System.ArgumentException("Only IMessageListener and ISessionAwareMessageListener supported");
}
throw new InvalidOperationException("No message listener specified - see property MessageListener");
}
}
/// <summary>
/// Invoke the specified listener as standard JMS MessageListener.
/// </summary>
/// <remarks>Default implementation performs a plain invocation of the
/// <code>OnMessage</code> methods</remarks>
/// <param name="listener">The listener to invoke.</param>
/// <param name="message">The received message.</param>
/// <exception cref="NMSException">if thronw by the underlying NMS APIs</exception>
protected virtual void DoInvokeListener(IMessageListener listener, IMessage message)
{
listener.OnMessage(message);
}
/// <summary>
/// Invoke the specified listener as Spring SessionAwareMessageListener,
/// exposing a new NMS Session (potentially with its own transaction)
@@ -221,6 +320,9 @@ namespace Spring.Messaging.Nms.Listener
/// <param name="listener">The Spring ISessionAwareMessageListener to invoke.</param>
/// <param name="session">The session to operate on.</param>
/// <param name="message">The received message.</param>
/// <exception cref="NMSException">If thrown by NMS API methods.</exception>
/// <see cref="ISessionAwareMessageListener"/>
/// <see cref="ExposeListenerSession"/>
protected virtual void DoInvokeListener(ISessionAwareMessageListener listener, ISession session, IMessage message)
{
IConnection conToClose = null;
@@ -239,13 +341,15 @@ namespace Spring.Messaging.Nms.Listener
if (logger.IsDebugEnabled)
{
logger.Debug("Invoking listener with message of type [" + message.GetType() +
"] and session [" + sessionToUse + "]");
"] and session [" + sessionToUse + "]");
}
listener.OnMessage(message, sessionToUse);
// Clean up specially exposed Session, if any
if (sessionToUse != session)
{
if (sessionToUse.Transacted)
if (sessionToUse.Transacted && SessionTransacted)
{
// Transacted session created by this container -> commit.
NmsUtils.CommitIfNecessary(sessionToUse);
}
}
@@ -255,9 +359,100 @@ namespace Spring.Messaging.Nms.Listener
NmsUtils.CloseConnection(conToClose);
}
}
protected virtual void HandleListenerException(System.Exception ex)
/// <summary>
/// Invoke the specified listener as standard JMS MessageListener.
/// </summary>
/// <remarks>Default implementation performs a plain invocation of the
/// <code>OnMessage</code> methods</remarks>
/// <param name="listener">The listener to invoke.</param>
/// <param name="message">The received message.</param>
/// <exception cref="NMSException">if thrown by the NMS API methods</exception>
protected virtual void DoInvokeListener(IMessageListener listener, IMessage message)
{
listener.OnMessage(message);
}
/// <summary>
/// Perform a commit or message acknowledgement, as appropriate
/// </summary>
/// <param name="session">The session to commit.</param>
/// <param name="message">The message to acknowledge.</param>
/// <exception cref="NMSException">In case of commit failure</exception>
protected virtual void CommitIfNecessary(ISession session, IMessage message)
{
// Commit session or acknowledge message
if (session.Transacted)
{
if (SessionTransacted)
{
NmsUtils.CommitIfNecessary(session);
}
}
else if (ClientAcknowledge(session))
{
message.Acknowledge();
}
}
/// <summary>
/// Perform a rollback, if appropriate.
/// </summary>
/// <param name="session">The session to rollback.</param>
/// <exception cref="NMSException">In case of a rollback error</exception>
protected virtual void RollbackIfNecessary(ISession session)
{
if (session.Transacted && SessionTransacted)
{
// Transacted session created by this container -> rollback
NmsUtils.RollbackIfNecessary(session);
}
}
/// <summary>
/// Perform a rollback, handling rollback excepitons properly.
/// </summary>
/// <param name="session">The session to rollback.</param>
/// <param name="ex">The thrown application exception.</param>
/// <exception cref="NMSException">in case of a rollback error.</exception>
protected virtual void RollbackOnExceptionIfNecessary(ISession session, Exception ex)
{
try
{
if (session.Transacted && SessionTransacted)
{
// Transacted session created by this container -> rollback
if (logger.IsDebugEnabled)
{
logger.Debug("Initiating transaction rollback on application exception");
}
NmsUtils.RollbackIfNecessary(session);
}
} catch (NMSException ex2)
{
logger.Error("Application exception overriden by rollback exception", ex);
throw;
}
}
/// <summary>
/// Handle the given exception that arose during listener execution.
/// </summary>
/// <remarks>
/// The default implementation logs the exception at error level,
/// not propagating it to the JMS provider - assuming that all handling of
/// acknowledgement and/or transactions is done by this listener container.
/// This can be overridden in subclasses.
/// </remarks>
/// <param name="ex">The exceptin to handle</param>
protected virtual void HandleListenerException(Exception ex)
{
if (ex is MessageRejectedWhileStoppingException)
{
// Internal exception - has been handled before.
return;
}
if (ex is NMSException)
{
InvokeExceptionListener((NMSException)ex);
@@ -276,55 +471,22 @@ namespace Spring.Messaging.Nms.Listener
}
}
protected virtual void InvokeExceptionListener(NMSException ex)
/// <summary>
/// Invokes the registered exception listener, if any.
/// </summary>
/// <param name="ex">The exception that arose during NMS processing.</param>
/// <see cref="ExceptionListener"/>
protected virtual void InvokeExceptionListener(Exception ex)
{
ExceptionListener exceptionListener = ExceptionListener;
if (exceptionListener != null)
IExceptionListener exListener = ExceptionListener;
if (exListener != null)
{
exceptionListener(ex);
exListener.OnException(ex);
}
}
#endregion
public override void AfterPropertiesSet()
{
base.AfterPropertiesSet();
if (this.destination == null)
{
throw new System.ArgumentException("destination or destinationName is required");
}
if (this.messageListener == null)
{
throw new System.ArgumentException("messageListener is required");
}
if (SubscriptionDurable && !PubSubDomain)
{
throw new System.ArgumentException("A durable subscription requires a topic (pub-sub domain)");
}
Initialize();
}
#region Template methods to be implemented by subclasses
#endregion
protected virtual bool IsClientAcknowledge(ISession session)
{
return (session.AcknowledgementMode == AcknowledgementMode.ClientAcknowledge);
}
protected virtual void CheckMessageListener(System.Object messageListener)
{
AssertUtils.ArgumentNotNull(messageListener, "IMessage Listener can not be null");
@@ -334,4 +496,12 @@ namespace Spring.Messaging.Nms.Listener
}
}
}
/// <summary>
/// Internal exception class that indicates a rejected message on shutdown.
/// Used to trigger a rollback for an external transaction manager in that case.
/// </summary>
internal class MessageRejectedWhileStoppingException : ApplicationException
{
}
}

410
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractNmsListeningContainer.cs
View File

@@ -20,6 +20,7 @@
using System;
using Apache.NMS;
using Common.Logging;
using Spring.Context;
using Spring.Messaging.Nms.Connection;
using Spring.Messaging.Nms.Support;
@@ -46,6 +47,12 @@ namespace Spring.Messaging.Nms.Listener
/// <author>Mark Pollack</author>
public abstract class AbstractNmsListeningContainer : NmsDestinationAccessor, ILifecycle, IObjectNameAware, IDisposable
{
#region Logging
private readonly ILog logger = LogManager.GetLogger(typeof(AbstractNmsListeningContainer));
#endregion
#region Fields
private String clientId;
@@ -55,6 +62,8 @@ namespace Spring.Messaging.Nms.Listener
private string objectName;
private IConnection sharedConnection;
private bool sharedConnectionStarted = false;
protected object sharedConnectionMonitor = new object();
@@ -80,30 +89,52 @@ namespace Spring.Messaging.Nms.Listener
set { this.autoStartup = value; }
}
public string ObjectName
{
set { objectName = value; }
}
/// <summary>
/// Gets a value indicating whether this container is currently running,
/// that is, whether it has been started and not stopped yet.
/// </summary>
/// <value>
/// <c>true</c> if this container is running; otherwise, <c>false</c>.
/// </value>
public bool IsRunning
{
get
{
lock (lifecycleMonitor)
{
return running;
return (running && RunningAllowed);
}
}
}
/// <summary> Return whether a shared NMS IConnection should be maintained
/// by this listener container base class.
/// <summary>
/// Gets a value indicating whether this container's listeners are generally allowed to run.
/// </summary>
/// <seealso cref="AbstractMessageListenerContainer.SharedConnection">
/// </seealso>
protected abstract bool SharedConnectionEnabled { get; }
public void Dispose()
/// <remarks>
/// <para>
/// >This implementation always returns <code>true</code>; the default 'running'
/// state is purely determined by <see cref="Start"/>/<see cref="Stop"/>.
/// </para>
/// <para>
/// Subclasses may override this method to check against temporary
/// conditions that prevent listeners from actually running. In other words,
/// they may apply further restrictions to the 'running' state, returning
/// <code>false</code> if such a restriction prevents listeners from running.
/// </para>
/// </remarks>
/// <value><c>true</c> if running allowed; otherwise, <c>false</c>.</value>
protected virtual bool RunningAllowed
{
Shutdown();
get {
return true; }
}
virtual public bool Active
public virtual bool Active
{
get
{
@@ -115,15 +146,95 @@ namespace Spring.Messaging.Nms.Listener
}
public string ObjectName
/// <summary> Return whether a shared NMS IConnection should be maintained
/// by this listener container base class.
/// </summary>
/// <seealso cref="AbstractMessageListenerContainer.SharedConnection">
/// </seealso>
protected abstract bool SharedConnectionEnabled { get; }
/// <summary>
/// Gets the shared connection maintained by this container.
/// Available after initialization.
/// </summary>
/// <value>The shared connection (never null)</value>
/// <exception cref="InvalidOperationException">if this container does not maintain a
/// shared Connection, or if the Connection hasn't been initialized yet.
/// </exception>
/// <see cref="SharedConnectionEnabled"/>
protected IConnection SharedConnection
{
set { objectName = value; }
get
{
if (!SharedConnectionEnabled)
{
throw new InvalidOperationException("This listener container does not maintain a shared IConnection");
}
lock (this.sharedConnectionMonitor)
{
if (this.sharedConnection == null)
{
throw new SharedConnectionNotInitializedException("This listener container's shared Connection has not been initialized yet");
}
return this.sharedConnection;
}
}
}
public override void AfterPropertiesSet()
{
base.AfterPropertiesSet();
ValidateConfiguration();
Initialize();
}
/// <summary>
/// Validates the configuration of this container. The default implementation
/// is empty. To be overriden in subclasses.
/// </summary>
protected virtual void ValidateConfiguration()
{
}
public void Dispose()
{
Shutdown();
}
public void Start()
/// <summary>
/// Initializes this container. Creates a Connection, starts the Connection
/// (if the property <see cref="AutoStartup"/> hasn't been turned off), and calls
/// <see cref="DoInitialize"/>.
/// </summary>
/// <exception cref="NMSException">If startup failed</exception>
public virtual void Initialize()
{
DoStart();
try
{
lock (this.lifecycleMonitor)
{
this.active = true;
System.Threading.Monitor.PulseAll(this.lifecycleMonitor);
}
if (this.autoStartup)
{
DoStart();
}
DoInitialize();
}
catch (Exception)
{
lock (this.sharedConnectionMonitor)
{
ConnectionFactoryUtils.ReleaseConnection(sharedConnection, ConnectionFactory, autoStartup);
}
throw;
}
}
public virtual void Shutdown()
@@ -152,111 +263,136 @@ namespace Spring.Messaging.Nms.Listener
// Shut down the invokers
try
{
DestroyListener();
DoShutdown();
}
finally
{
lock (this.sharedConnectionMonitor)
{
NmsUtils.CloseConnection(this.sharedConnection, wasRunning);
ConnectionFactoryUtils.ReleaseConnection(this.sharedConnection, ConnectionFactory, false);
}
}
}
protected void DoStart()
/// <summary>
/// Starts this container.
/// </summary>
/// <exception cref="NMSException">if starting failed.</exception>
public void Start()
{
DoStart();
}
protected virtual void DoStart()
{
// Lazily establish a shared Connection, if necessary.
if (SharedConnectionEnabled)
{
EstablishSharedConnection();
}
lock (this.lifecycleMonitor)
{
running = true;
System.Threading.Monitor.PulseAll(this.lifecycleMonitor);
//TODO - PausedTasks
}
// Start the shared Connection, if any.
if (SharedConnectionEnabled)
{
StartSharedConnection();
}
}
protected virtual void StartSharedConnection()
/// <summary>
/// Stops this container.
/// </summary>
/// <exception cref="NMSException">if stopping failed.</exception>
public void Stop()
{
DoStop();
}
/// <summary>
/// Notify all invoker tasks and stop the shared Connection, if any.
/// </summary>
/// <exception cref="NMSException">if thrown by NMS API methods.</exception>
/// <see cref="StopSharedConnection"/>
protected virtual void DoStop()
{
lock (this.lifecycleMonitor)
{
this.running = false;
System.Threading.Monitor.PulseAll(this.lifecycleMonitor);
}
if (SharedConnectionEnabled)
{
StopSharedConnection();
}
}
/// <summary>
/// Register any invokers within this container.
/// Subclasses need to implement this method for their specific
/// invoker management process. A shared Connection, if any, will already have been
/// started at this point.
/// </summary>
protected abstract void DoInitialize();
/// <summary>
/// Close the registered invokers. Subclasses need to implement this method
/// for their specific invoker management process. A shared Connection, if any,
/// will automatically be closed afterwards.
/// </summary>
protected abstract void DoShutdown();
/// <summary>
/// Establishes a shared Connection for this container.
/// </summary>
/// <remarks>
/// <para>
/// The default implementation delegates to <see cref="CreateSharedConnection"/>
/// which does one immediate attempt and throws an exception if it fails.
/// Can be overridden to have a recovery process in place, retrying
/// until a Connection can be successfully established.
/// </para>
/// </remarks>
/// <exception cref="NMSException">If thrown by NMS API methods</exception>
protected virtual void EstablishSharedConnection()
{
lock (sharedConnectionMonitor)
{
if (sharedConnection != null)
if (sharedConnection == null)
{
try
{
sharedConnection.Start();
}
catch (Exception ex)
{
logger.Debug("Ignoring IConnection start exception - assuming already started", ex);
}
sharedConnection = CreateSharedConnection();
logger.Debug("Established shared NMS Connection");
}
}
}
public IConnection SharedConnection
{
get
{
if (!SharedConnectionEnabled)
{
throw new System.SystemException("This message listener container does not maintain a shared IConnection");
}
lock (this.sharedConnectionMonitor)
{
if (this.sharedConnection == null)
{
//TODO SharedConnectionNotInitializedException
throw new ApplicationException("This message listener container's shared IConnection has not been initialized yet");
}
return this.sharedConnection;
}
}
}
public virtual void Initialize()
{
try
{
lock (this.lifecycleMonitor)
{
this.active = true;
System.Threading.Monitor.PulseAll(this.lifecycleMonitor);
}
if (SharedConnectionEnabled)
{
EstablishSharedConnection();
}
if (this.autoStartup)
{
DoStart();
}
RegisterListener();
}
catch (Exception)
{
lock (this.sharedConnectionMonitor)
{
ConnectionFactoryUtils.ReleaseConnection(sharedConnection, ConnectionFactory, autoStartup);
}
throw;
}
}
protected virtual void EstablishSharedConnection()
{
RefreshSharedConnection();
}
/// <summary>
/// Refreshes the shared connection that this container holds.
/// </summary>
/// <remarks>
/// Called on startup and also after an infrastructure exception
/// that occurred during invoker setup and/or execution.
/// </remarks>
/// <exception cref="NMSException">If thrown by NMS API methods</exception>
protected void RefreshSharedConnection()
{
lock (sharedConnectionMonitor)
{
ConnectionFactoryUtils.ReleaseConnection(sharedConnection, ConnectionFactory, sharedConnectionStarted);
sharedConnection = CreateSharedConnection();
if (sharedConnectionStarted)
{
sharedConnection.Start();
}
}
/*
bool running = IsRunning;
lock (this.sharedConnectionMonitor)
{
@@ -273,9 +409,42 @@ namespace Spring.Messaging.Nms.Listener
throw;
}
this.sharedConnection = con;
}*/
}
/// <summary>
/// Creates the shared connection for this container.
/// </summary>
/// <remarks>
/// The default implementation creates a standard Connection
/// and prepares it through <see cref="PrepareSharedConnection"/>
/// </remarks>
/// <returns>the prepared Connection</returns>
/// <exception cref="NMSException">if the creation failed.</exception>
protected virtual IConnection CreateSharedConnection()
{
IConnection con = CreateConnection();
try
{
PrepareSharedConnection(con);
return con;
} catch (NMSException ex)
{
NmsUtils.CloseConnection(con);
throw;
}
}
/// <summary>
/// Prepares the given connection, which is about to be registered
/// as shared Connection for this container.
/// </summary>
/// <remarks>
/// The default implementation sets the specified client id, if any.
/// Subclasses can override this to apply further settings.
/// </remarks>
/// <param name="connection">The connection to prepare.</param>
/// <exception cref="NMSException">If the preparation efforts failed.</exception>
protected virtual void PrepareSharedConnection(IConnection connection)
{
if (ClientId != null)
@@ -284,33 +453,28 @@ namespace Spring.Messaging.Nms.Listener
}
}
/// <summary> Register the specified listener on the underlying NMS IConnection.
/// <p>Subclasses need to implement this method for their specific
/// listener management process.</p>
/// <summary>
/// Starts the shared connection.
/// </summary>
/// <throws> NMSException if registration failed </throws>
/// <seealso cref="IMessageListener">
/// </seealso>
/// <seealso cref="SharedConnection">
/// </seealso>
protected abstract void RegisterListener();
public void Stop()
/// <exception cref="NMSException">If thrown by NMS API methods</exception>
/// <see cref="Start"/>
protected virtual void StartSharedConnection()
{
DoStop();
}
protected virtual void DoStop()
{
lock (this.lifecycleMonitor)
lock (sharedConnectionMonitor)
{
this.running = false;
System.Threading.Monitor.PulseAll(this.lifecycleMonitor);
}
if (SharedConnectionEnabled)
{
StopSharedConnection();
if (sharedConnection != null)
{
try
{
sharedConnectionStarted = true;
sharedConnection.Start();
}
catch (Exception ex)
{
logger.Warn("Ignoring Connection start exception - assuming already started", ex);
}
}
}
}
@@ -322,24 +486,32 @@ namespace Spring.Messaging.Nms.Listener
{
try
{
this.sharedConnectionStarted = false;
this.sharedConnection.Stop();
}
catch (System.InvalidOperationException ex)
{
logger.Debug("Ignoring IConnection stop exception - assuming already stopped", ex);
logger.Warn("Ignoring Connection stop exception - assuming already stopped", ex);
}
}
}
}
}
/// <summary> Destroy the registered listener.
/// The NMS IConnection will automatically be closed <i>afterwards</i>
/// <p>Subclasses need to implement this method for their specific
/// listener management process.</p>
/// <summary>
/// Exception that indicates that the initial setup of this container's
/// shared Connection failed. This is indicating to invokers that they need
/// to establish the shared Connection themselves on first access.
/// </summary>
public class SharedConnectionNotInitializedException : ApplicationException
{
/// <summary>
/// Initializes a new instance of the <see cref="SharedConnectionNotInitializedException"/> class.
/// </summary>
/// <throws> NMSException if destruction failed </throws>
protected abstract void DestroyListener();
/// <param name="message">The message.</param>
public SharedConnectionNotInitializedException(string message) : base(message)
{
}
}
}

43
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/LocallyExposedNmsResourceHolder.cs Normal file
View File

@@ -0,0 +1,43 @@
#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;
using Spring.Messaging.Nms.Connection;
namespace Spring.Messaging.Nms.Listener
{
/// <summary>
/// NmsResourceHolder marker subclass that indicates local exposure,
/// i.e. that does not indicate an externally managed transaction.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class LocallyExposedNmsResourceHolder : NmsResourceHolder
{
/// <summary>
/// Initializes a new instance of the <see cref="LocallyExposedNmsResourceHolder"/> class.
/// </summary>
/// <param name="session">The session.</param>
public LocallyExposedNmsResourceHolder(ISession session) : base(session)
{
}
}
}

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

@@ -1,27 +1,61 @@
#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 Common.Logging;
using Spring.Collections;
using Spring.Messaging.Nms.Support;
using Apache.NMS;
using Spring.Transaction.Support;
using Spring.Util;
namespace Spring.Messaging.Nms.Listener
{
public class SimpleMessageListenerContainer : AbstractMessageListenerContainer
/// <summary>
/// Message listener container that uses the plain NMS client API's
/// <see cref="IMessageConsumer.Listener"/> method to create concurrent
/// MessageConsumers for the specified listeners.
/// </summary>
public class SimpleMessageListenerContainer : AbstractMessageListenerContainer, IExceptionListener
{
#region Logging
private readonly ILog logger = LogManager.GetLogger(typeof(SimpleMessageListenerContainer));
#endregion
#region fields
private bool pubSubNoLocal = false;
private int concurrentConsumers = 1;
//private TaskExecutor taskExecutor;
private ISet sessions;
private ISet consumers;
private object consumersMonitor = new object();
public int ConcurrentConsumers
{
get { return concurrentConsumers; }
set { concurrentConsumers = value; }
}
#endregion
#region Properties
public bool PubSubNoLocal
{
@@ -29,24 +63,123 @@ namespace Spring.Messaging.Nms.Listener
set { pubSubNoLocal = value; }
}
public int ConcurrentConsumers
{
set
{
AssertUtils.IsTrue(value > 0, "'ConcurrentConsumer' value must be at least 1 (one)");
concurrentConsumers = value;
}
}
/// <summary>
/// Always use a shared NMS connection
/// </summary>
protected override bool SharedConnectionEnabled
{
get { return true; }
}
protected override void RegisterListener()
#endregion
protected override void ValidateConfiguration()
{
this.sessions = new HashedSet();
this.consumers = new HashedSet();
for (int i = 0; i < this.concurrentConsumers; i++)
base.ValidateConfiguration();
if (SubscriptionDurable && concurrentConsumers !=1 )
{
ISession session = CreateSession(SharedConnection);
IMessageConsumer consumer = CreateListenerConsumer(session);
this.sessions.Add(session);
this.consumers.Add(consumer);
throw new ArgumentException("Only 1 concurrent consumer supported for durable subscription");
}
}
/// <summary>
/// Creates the specified number of concurrent consumers,
/// in the form of a JMS Session plus associated MessageConsumer
/// </summary>
/// <see cref="CreateListenerConsumer"/>
protected override void DoInitialize()
{
EstablishSharedConnection();
InitializeConsumers();
}
/// <summary>
/// Re-initializes this container's NMS message consumers,
/// if not initialized already.
/// </summary>
protected override void DoStart()
{
base.DoStart();
InitializeConsumers();
}
/// <summary>
/// Registers this listener container as NMS ExceptionListener on the shared connection.
/// </summary>
/// <param name="connection"></param>
protected override void PrepareSharedConnection(IConnection connection)
{
base.PrepareSharedConnection(connection);
connection.ExceptionListener += OnException;
}
public void OnException(Exception exception)
{
InvokeExceptionListener(exception);
// now try to recover the shared Connection and all consumers...
if (logger.IsInfoEnabled)
{
logger.Info("Trying to recover from NMS Connection exception: " + exception);
}
try
{
lock(consumersMonitor)
{
sessions = null;
consumers = null;
}
RefreshSharedConnection();
InitializeConsumers();
logger.Info("Successfully refreshed NMS Connection");
} catch (NMSException recoverEx)
{
logger.Debug("Failed to recover NMS Connection", recoverEx);
logger.Error("Encountered non-recoverable NMSException", exception);
}
}
/// <summary>
/// Initialize the Sessions and MessageConsumers for this container.
/// </summary>
/// <exception cref="NMSException">in case of setup failure.</exception>
protected virtual void InitializeConsumers()
{
// Register Sessions and MessageConsumers
lock (consumersMonitor)
{
if (this.consumers == null)
{
logger.Debug("InitializingConsumers **********");
this.sessions = new HashedSet();
this.consumers = new HashedSet();
IConnection con = SharedConnection;
for (int i = 0; i < this.concurrentConsumers; i++)
{
ISession session = CreateSession(SharedConnection);
IMessageConsumer consumer = CreateListenerConsumer(session);
this.sessions.Add(session);
this.consumers.Add(consumer);
}
}
}
}
/// <summary>
/// Creates a MessageConsumer for the given Session,
/// registering a MessageListener for the specified listener
/// </summary>
/// <param name="session">The session to work on.</param>
/// <returns>the IMessageConsumer"/></returns>
/// <exception cref="NMSException">if thrown by NMS methods</exception>
private IMessageConsumer CreateListenerConsumer(ISession session)
{
@@ -55,7 +188,7 @@ namespace Spring.Messaging.Nms.Listener
{
destination = ResolveDestinationName(session, DestinationName);
}
//TODO TaskExectuor abstraction would go here...
IMessageConsumer consumer = CreateConsumer(session, destination);
SimpleMessageListener listener = new SimpleMessageListener(this, session);
@@ -66,36 +199,22 @@ namespace Spring.Messaging.Nms.Listener
/// <summary>
/// Close the message consumers and sessions.
/// </summary>
protected override void DestroyListener()
/// <throws>NMSException if destruction failed </throws>
protected override void DoShutdown()
logger.Debug("Closing NMS IMessageConsumers");
{
logger.Debug("Closing NMS MessageConsumers");
foreach (IMessageConsumer messageConsumer in consumers)
{
NmsUtils.CloseMessageConsumer(messageConsumer);
logger.Debug("Closing NMS ISessions");
}
logger.Debug("Closing NMS Sessions");
foreach (ISession session in sessions)
{
NmsUtils.CloseSession(session);
}
consumers = null;
sessions = null;
}
/// <summary>
/// Afters the properties set.
/// </summary>
public override void AfterPropertiesSet()
{
if (this.concurrentConsumers <= 0)
{
throw new System.ArgumentException("concurrentConsumers value must be at least 1 (one)");
}
if (SubscriptionDurable && this.concurrentConsumers != 1)
{
throw new System.ArgumentException("Only 1 concurrent consumer supported for durable subscription");
}
base.AfterPropertiesSet();
}
protected IMessageConsumer CreateConsumer(ISession session, IDestination destination)
@@ -134,7 +253,22 @@ namespace Spring.Messaging.Nms.Listener
}
public void OnMessage(IMessage message)
container.ExecuteListener(session, message);
{
bool exposeResource = container.ExposeListenerSession;
if (exposeResource)
{
TransactionSynchronizationManager.BindResource(
container.ConnectionFactory, new LocallyExposedNmsResourceHolder(session));
}
try
{
container.ExecuteListener(session, message);
} finally
{
if (exposeResource)
{
TransactionSynchronizationManager.UnbindResource(container.ConnectionFactory);
}
}
}
}

17
src/Spring/Spring.Messaging.Nms/Messaging/Nms/NmsTemplate.cs
View File

@@ -19,6 +19,7 @@
#endregion
using System;
using Common.Logging;
using Spring.Messaging.Nms.Connection;
using Spring.Messaging.Nms.Support;
using Spring.Messaging.Nms.Support.Converter;
@@ -48,6 +49,12 @@ namespace Spring.Messaging.Nms
/// <author>Mark Pollack (.NET)</author>
public class NmsTemplate : NmsDestinationAccessor, INmsOperations
{
#region Logging
private readonly ILog logger = LogManager.GetLogger(typeof(NmsTemplate));
#endregion
#region Fields
public static readonly long DEFAULT_RECEIVE_TIMEOUT = -1;
@@ -936,16 +943,6 @@ namespace Spring.Messaging.Nms
}
}
/// <summary>
/// Returns whether the ISession is in client acknowledgement mode.
/// </summary>
/// <param name="session">The session.</param>
/// <returns>true ifin client ack mode, false otherwise</returns>
protected virtual bool ClientAcknowledge(ISession session)
{
return (session.AcknowledgementMode == AcknowledgementMode.ClientAcknowledge);
}
/// <summary> Receive a message synchronously from the default destination, but only
/// wait up to a specified time for delivery. Convert the message into an

2
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Support/Destinations/NmsDestinationAccessor.cs
View File

@@ -18,6 +18,8 @@
#endregion
using System;
using Spring.Objects.Factory;
using Spring.Util;
using Apache.NMS;

12
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Support/NmsAccessor.cs
View File

@@ -40,7 +40,7 @@ namespace Spring.Messaging.Nms.Support
{
#region Logging
protected readonly ILog logger = LogManager.GetLogger(typeof(NmsAccessor));
private readonly ILog logger = LogManager.GetLogger(typeof(NmsAccessor));
#endregion
@@ -146,5 +146,15 @@ namespace Spring.Messaging.Nms.Support
{
return con.CreateSession(SessionAcknowledgeMode);
}
/// <summary>
/// Returns whether the ISession is in client acknowledgement mode.
/// </summary>
/// <param name="session">The session.</param>
/// <returns>true ifin client ack mode, false otherwise</returns>
protected virtual bool ClientAcknowledge(ISession session)
{
return (session.AcknowledgementMode == AcknowledgementMode.ClientAcknowledge);
}
}
}

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

@@ -46,7 +46,6 @@
<Reference Include="System.Xml" />
</ItemGroup>
<ItemGroup>
<Compile Include="Context\ILifecycle.cs" />
<Compile Include="Messaging\Nms\Connections\CachedMessageProducer.cs" />
<Compile Include="Messaging\Nms\Connections\CachedSession.cs" />
<Compile Include="Messaging\Nms\Connections\CachingConnectionFactory.cs" />
@@ -70,6 +69,7 @@
<Compile Include="Messaging\Nms\Listener\Adapter\MessageListenerAdapter.cs" />
<Compile Include="Messaging\Nms\Listener\DefaultMessageListenerContainer.cs" />
<Compile Include="Messaging\Nms\Listener\ISessionAwareMessageListener.cs" />
<Compile Include="Messaging\Nms\Listener\LocallyExposedNmsResourceHolder.cs" />
<Compile Include="Messaging\Nms\Listener\SimpleMessageListenerContainer.cs" />
<Compile Include="Messaging\Nms\MessageCreatorDelegate.cs" />
<Compile Include="Messaging\Nms\NmsGatewaySupport.cs" />

2
src/Spring/Spring.Web/Objects/Factory/Support/WebObjectFactory.cs
View File

@@ -206,7 +206,7 @@ namespace Spring.Objects.Factory.Support
/// </remarks>
/// <param name="objectName">Object name to look for.</param>
/// <returns>Cached object if found, null otherwise.</returns>
protected override object GetSingleton(string objectName)
public override object GetSingleton(string objectName)
{
object instance = null;

23
test/Spring/Spring.Core.Tests/CommonTypes.cs
View File

@@ -42,7 +42,12 @@ namespace Spring
{
}
public void Dispose()
public object GetSingleton(string objectName)
{
throw new NotImplementedException();
}
public void Dispose()
{
_destroyCalls.Inc();
}
@@ -79,7 +84,21 @@ namespace Spring
throw new NotImplementedException();
}
public IObjectDefinition GetObjectDefinition(string objectName)
public string[] SingletonNames
{
get { throw new NotImplementedException(); }
}
#region ISingletonObjectRegistry Members
public int SingletonCount
{
get { throw new NotImplementedException(); }
}
#endregion
public IObjectDefinition GetObjectDefinition(string objectName)
{
return null;
}

28
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/LoggingExceptionHandler.cs Normal file
View File

@@ -0,0 +1,28 @@
using System;
using Spring.Messaging.Nms;
using Common.Logging;
namespace Spring.Messaging.Nms.Integration
{
/// <summary>
///
/// </summary>
public class LoggingExceptionHandler : IExceptionListener
{
#region Logging Definition
private static readonly ILog LOG = LogManager.GetLogger(typeof (LoggingExceptionHandler));
#endregion
#region IExceptionListener Members
public void OnException(Exception e)
{
LOG.Error("Exception processing message", e);
}
#endregion
}
}

41
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListener.cs Normal file
View File

@@ -0,0 +1,41 @@
using Spring.Messaging.Nms;
using Apache.NMS;
using Common.Logging;
namespace Spring.Messaging.Nms.Integration
{
public class SimpleMessageListener : IMessageListener
{
#region Logging Definition
private static readonly ILog LOG = LogManager.GetLogger(typeof(SimpleMessageListener));
#endregion
private IMessage lastReceivedMessage;
private int messageCount;
public IMessage LastReceivedMessage
{
get { return lastReceivedMessage; }
}
public int MessageCount
{
get { return messageCount; }
}
#region IMessageListener Members
public void OnMessage(IMessage message)
{
lastReceivedMessage = message;
messageCount++;
LOG.Debug("Message listener count = " + messageCount);
}
#endregion
}
}

89
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListenerContainerTests.cs Normal file
View File

@@ -0,0 +1,89 @@
#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 System.Threading;
using NUnit.Framework;
using Spring.Messaging.Nms.Listener;
using Spring.Testing.NUnit;
#endregion
namespace Spring.Messaging.Nms.Integration
{
/// <summary>
/// This class contains tests for
/// </summary>
/// <author>Mark Pollack</author>
/// <version>$Id:$</version>
[TestFixture]
public class SimpleMessageListenerContainerTests : AbstractDependencyInjectionSpringContextTests
{
[Test]
[Explicit]
public void SendAndAsyncReceive()
{
SimpleMessageListenerContainer container =
(SimpleMessageListenerContainer) applicationContext["SimpleMessageListenerContainer"];
SimpleMessageListener listener = applicationContext["SimpleMessageListener"] as SimpleMessageListener;
Assert.IsNotNull(container);
Assert.IsNotNull(listener);
NmsTemplate nmsTemplate = (NmsTemplate) applicationContext["NmsTemplate"] as NmsTemplate;
Assert.IsNotNull(nmsTemplate);
Assert.AreEqual(0, listener.MessageCount);
nmsTemplate.ConvertAndSend("Hello World 1");
int waitInMillis = 2000;
Thread.Sleep(waitInMillis);
Assert.AreEqual(1,listener.MessageCount);
container.Stop();
Console.WriteLine("container stopped.");
nmsTemplate.ConvertAndSend("Hello World 2");
Thread.Sleep(waitInMillis);
Assert.AreEqual(1, listener.MessageCount);
container.Start();
Console.WriteLine("container started.");
Thread.Sleep(waitInMillis);
Assert.AreEqual(2, listener.MessageCount);
container.Shutdown();
Thread.Sleep(waitInMillis);
}
protected override string[] ConfigLocations
{
get { return new string[] { "assembly://Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Integration/SimpleMessageListenerContainerTests.xml" }; }
}
}
}

30
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListenerContainerTests.xml Normal file
View File

@@ -0,0 +1,30 @@
<?xml version="1.0" encoding="utf-8" ?>
<objects xmlns="http://www.springframework.net">
<object name="ConnectionFactory" type="Apache.NMS.ActiveMQ.ConnectionFactory, Apache.NMS.ActiveMQ">
<property name="BrokerUri" value="tcp://localhost:61616"/>
</object>
<!--
<object name="SingleConnectionFactory" type="Spring.Messaging.Nms.Connection.SingleConnectionFactory, Spring.Messaging.Nms">
<property name="TargetConnectionFactory" ref="ConnectionFactory"/>
</object>
-->
<object name="SimpleMessageListenerContainer" type="Spring.Messaging.Nms.Listener.SimpleMessageListenerContainer, Spring.Messaging.Nms">
<property name="ConnectionFactory" ref="ConnectionFactory"/>
<property name="DestinationName" value="test.queue"/>
<property name="MessageListener" ref="SimpleMessageListener"/>
<property name="ExceptionListener">
<object type="Spring.Messaging.Nms.Integration.LoggingExceptionHandler, Spring.Messaging.Nms.Tests"/>
</property>
</object>
<object id="SimpleMessageListener" type="Spring.Messaging.Nms.Integration.SimpleMessageListener, Spring.Messaging.Nms.Tests"/>
<object name="NmsTemplate" type="Spring.Messaging.Nms.NmsTemplate, Spring.Messaging.Nms">
<property name="ConnectionFactory" ref="ConnectionFactory"/>
<property name="DefaultDestinationName" value="test.queue"/>
</object>
</objects>

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

@@ -37,6 +37,10 @@
<SpecificVersion>False</SpecificVersion>
<HintPath>..\..\..\lib\Net\2.0\Apache.NMS.dll</HintPath>
</Reference>
<Reference Include="Apache.NMS.ActiveMQ, Version=1.0.0.0, Culture=neutral, PublicKeyToken=65e474d141e25e07, processorArchitecture=MSIL">
<SpecificVersion>False</SpecificVersion>
<HintPath>..\..\..\lib\Net\2.0\Apache.NMS.ActiveMQ.dll</HintPath>
</Reference>
<Reference Include="Common.Logging, Version=1.2.0.0, Culture=neutral, PublicKeyToken=af08829b84f0328e">
<SpecificVersion>False</SpecificVersion>
<HintPath>..\..\..\lib\Net\2.0\Common.Logging.dll</HintPath>
@@ -74,6 +78,10 @@
<Project>{AEB1578C-9018-4D49-B440-789F38DD2F29}</Project>
<Name>Spring.Messaging.Nms.2005</Name>
</ProjectReference>
<ProjectReference Include="..\..\..\src\Spring\Spring.Testing.NUnit\Spring.Testing.NUnit.2005.csproj">
<Project>{ED204A7B-832F-44C7-BFE3-504AEBE1BCC8}</Project>
<Name>Spring.Testing.NUnit.2005</Name>
</ProjectReference>
<ProjectReference Include="..\Spring.Core.Tests\Spring.Core.Tests.2005.csproj">
<Project>{44B16BAA-6DF8-447C-9D7F-3AD3D854D904}</Project>
<Name>Spring.Core.Tests.2005</Name>
@@ -87,8 +95,12 @@
<Compile Include="Messaging\Nms\Connections\TestExceptionListener.cs" />
<Compile Include="Messaging\Nms\Connections\TestMessageProducer.cs" />
<Compile Include="Messaging\Nms\Connections\TestSession.cs" />
<Compile Include="Messaging\Nms\Integration\LoggingExceptionHandler.cs" />
<Compile Include="Messaging\Nms\Integration\SimpleMessageListener.cs" />
<Compile Include="Messaging\Nms\Integration\SimpleMessageListenerContainerTests.cs" />
</ItemGroup>
<ItemGroup>
<EmbeddedResource Include="Messaging\Nms\Integration\SimpleMessageListenerContainerTests.xml" />
<Content Include="Spring.Messaging.Nms.Tests.dll.config">
<CopyToOutputDirectory>Always</CopyToOutputDirectory>
</Content>

15
test/Spring/Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Tests.dll.config
View File

@@ -17,9 +17,20 @@ limitations under the License.
<configuration>
<configSections>
<sectionGroup name="common">
<section name="logging" type="Common.Logging.ConfigurationSectionHandler, Common.Logging" />
</sectionGroup>
</configSections>
<common>
<logging>
<factoryAdapter type="Common.Logging.Simple.ConsoleOutLoggerFactoryAdapter, Common.Logging">
<arg key="level" value="DEBUG" />
<arg key="showLogName" value="true" />
<arg key="showDataTime" value="true" />
<arg key="dateTimeFormat" value="yyyy/MM/dd HH:mm:ss:fff" />
</factoryAdapter>
</logging>
</common>
</configuration>