From c467f4903527f1397362ed13458f3db4e3a2d91a Mon Sep 17 00:00:00 2001 From: markpollack Date: Wed, 23 Jul 2008 22:54:02 +0000 Subject: [PATCH] NMS Development --- .../IConfigurableApplicationContext.cs | 2 +- src/Spring/Spring.Core/Context/ILifecycle.cs | 73 ++++ .../Support/AbstractApplicationContext.cs | 111 ++++- .../Config/IConfigurableObjectFactory.cs | 69 +-- .../Config/ISingletonObjectRegistry.cs | 169 ++++++++ .../Factory/Support/AbstractObjectFactory.cs | 105 ++++- .../Spring.Core/Spring.Core.2005.csproj | 2 + .../Context/ILifecycle.cs | 15 - .../Nms/Connections/NmsResourceHolder.cs | 30 +- .../Connections/SingleConnectionFactory.cs | 16 +- .../AbstractMessageListenerContainer.cs | 360 +++++++++++---- .../Listener/AbstractNmsListeningContainer.cs | 410 +++++++++++++----- .../LocallyExposedNmsResourceHolder.cs | 43 ++ .../SimpleMessageListenerContainer.cs | 208 +++++++-- .../Messaging/Nms/NmsTemplate.cs | 17 +- .../Destinations/NmsDestinationAccessor.cs | 2 + .../Messaging/Nms/Support/NmsAccessor.cs | 12 +- .../Spring.Messaging.Nms.2005.csproj | 2 +- .../Factory/Support/WebObjectFactory.cs | 2 +- test/Spring/Spring.Core.Tests/CommonTypes.cs | 23 +- .../Integration/LoggingExceptionHandler.cs | 28 ++ .../Nms/Integration/SimpleMessageListener.cs | 41 ++ .../SimpleMessageListenerContainerTests.cs | 89 ++++ .../SimpleMessageListenerContainerTests.xml | 30 ++ .../Spring.Messaging.Nms.Tests.2005.csproj | 12 + .../Spring.Messaging.Nms.Tests.dll.config | 15 +- 26 files changed, 1512 insertions(+), 374 deletions(-) create mode 100644 src/Spring/Spring.Core/Context/ILifecycle.cs create mode 100644 src/Spring/Spring.Core/Objects/Factory/Config/ISingletonObjectRegistry.cs delete mode 100644 src/Spring/Spring.Messaging.Nms/Context/ILifecycle.cs create mode 100644 src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/LocallyExposedNmsResourceHolder.cs create mode 100644 test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/LoggingExceptionHandler.cs create mode 100644 test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListener.cs create mode 100644 test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListenerContainerTests.cs create mode 100644 test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListenerContainerTests.xml diff --git a/src/Spring/Spring.Core/Context/IConfigurableApplicationContext.cs b/src/Spring/Spring.Core/Context/IConfigurableApplicationContext.cs index a45ad274..6015a91e 100644 --- a/src/Spring/Spring.Core/Context/IConfigurableApplicationContext.cs +++ b/src/Spring/Spring.Core/Context/IConfigurableApplicationContext.cs @@ -57,7 +57,7 @@ namespace Spring.Context /// Mark Pollack (.NET) /// /// - public interface IConfigurableApplicationContext : IApplicationContext + public interface IConfigurableApplicationContext : IApplicationContext, ILifecycle { /// /// Return the internal object factory of this application context. diff --git a/src/Spring/Spring.Core/Context/ILifecycle.cs b/src/Spring/Spring.Core/Context/ILifecycle.cs new file mode 100644 index 00000000..12645e8f --- /dev/null +++ b/src/Spring/Spring.Core/Context/ILifecycle.cs @@ -0,0 +1,73 @@ +#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.Context +{ + /// + /// Interface defining methods for start/stop lifecycle control. + /// The typical use case for this is to control asynchronous processing. + /// + /// + /// + /// Can be implemented by both components (typically a Spring object defined in + /// a spring and containers + /// (typically a spring . Containers will + /// propagate start/stop signals to all components that apply. + /// + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public interface ILifecycle + { + /// + /// Starts this component. + /// + /// 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. + /// + void Start(); + + /// + /// Stops this component. + /// + /// + /// 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. + /// + void Stop(); + + /// + /// Gets a value indicating whether this component is currently running. + /// + /// + /// In the case of a container, this will return true + /// only if all components that apply are currently running. + /// + /// + /// true if this component is running; otherwise, false. + /// + bool IsRunning + { + get; + } + } +} diff --git a/src/Spring/Spring.Core/Context/Support/AbstractApplicationContext.cs b/src/Spring/Spring.Core/Context/Support/AbstractApplicationContext.cs index acf1052d..3c8a88c9 100644 --- a/src/Spring/Spring.Core/Context/Support/AbstractApplicationContext.cs +++ b/src/Spring/Spring.Core/Context/Support/AbstractApplicationContext.cs @@ -844,8 +844,115 @@ namespace Spring.Context.Support set { _parentApplicationContext = value; } } - #endregion - + #endregion + + #region ILifecycle Members + + /// + /// Starts this component. + /// + /// 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. + /// + 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(); + } + } + } + } + + /// + /// Stops this component. + /// + /// + /// 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. + /// + 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(); + } + } + } + } + + /// + /// Gets a value indicating whether this component is currently running. + /// + /// + /// true if this component is running; otherwise, false. + /// + /// + /// In the case of a container, this will return true + /// only if all components that apply are currently running. + /// + 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; + } + } + + /// + /// Gets a dictionary of all singleton beans that implement the + /// ILifecycle interface in this context. + /// + /// A dictionary of ILifecycle objects with object name as key. + 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 /// diff --git a/src/Spring/Spring.Core/Objects/Factory/Config/IConfigurableObjectFactory.cs b/src/Spring/Spring.Core/Objects/Factory/Config/IConfigurableObjectFactory.cs index f0fedd67..f9e01ea1 100644 --- a/src/Spring/Spring.Core/Objects/Factory/Config/IConfigurableObjectFactory.cs +++ b/src/Spring/Spring.Core/Objects/Factory/Config/IConfigurableObjectFactory.cs @@ -49,7 +49,7 @@ namespace Spring.Objects.Factory.Config /// /// Juergen Hoeller /// Rick Evans (.NET) - public interface IConfigurableObjectFactory : IHierarchicalObjectFactory + public interface IConfigurableObjectFactory : IHierarchicalObjectFactory, ISingletonObjectRegistry { /// /// Set the parent of this object factory. @@ -146,27 +146,6 @@ namespace Spring.Objects.Factory.Config /// void RegisterAlias(string name, string theAlias); - /// - /// Register the given existing object as singleton in the object factory, - /// under the given object name. - /// - /// - ///

- /// 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. - ///

- /// - /// - /// The name of the object. - /// - /// The existing object. - /// - /// If the singleton could not be registered. - /// - void RegisterSingleton(string name, object singleton); - /// /// Register the given custom /// for all properties of the given . @@ -184,51 +163,5 @@ namespace Spring.Objects.Factory.Config /// void RegisterCustomConverter(Type requiredType, TypeConverter converter); - /// - /// Does this object factory contains a singleton instance with the - /// supplied ? - /// - /// - ///

- /// Only checks already instantiated singletons; does not return - /// for singleton object definitions that have - /// not been instantiated yet. - ///

- ///

- /// The main purpose of this method is to check manually registered - /// singletons (). This - /// method can also be used to check whether a singleton defined by an - /// object definition has already been created. - ///

- ///

- /// To check whether an object factory contains an object definition - /// with a given name, use the - /// - /// method. Calling both - /// - /// and definitively answers - /// the question of whether a specific object factory contains a - /// singleton object with the given name. - ///

- ///

- /// Use the - /// - /// 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. - ///

- /// - /// - /// The name of the (singleton) object to look for. - /// - /// - /// if this object factory contains a singleton - /// instance with the given . - /// - /// - /// - bool ContainsSingleton(string name); } } diff --git a/src/Spring/Spring.Core/Objects/Factory/Config/ISingletonObjectRegistry.cs b/src/Spring/Spring.Core/Objects/Factory/Config/ISingletonObjectRegistry.cs new file mode 100644 index 00000000..4b218ef8 --- /dev/null +++ b/src/Spring/Spring.Core/Objects/Factory/Config/ISingletonObjectRegistry.cs @@ -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 +{ + /// + /// Interface that defines a registry for shared object instances. + /// + /// + /// Can be implemented by + /// implementations in order to expose their singleton management facility + /// in a uniform manner. + /// + /// The interface extends this interface. + /// + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public interface ISingletonObjectRegistry + { + /// + /// Registers the given existing object as singleton in the object registry, + /// under the given object name. + /// + /// + /// + /// 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 AfterPropertiesSet method). + /// The given instance will not receive any destruction callbacks + /// (like IDisposable's Dispose method) either. + /// + /// + /// 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. + /// + /// + /// 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. + /// + /// + /// Name of the object. + /// The singleton object. + /// + /// + void RegisterSingleton(string objectName, object singletonObject); + + + /// + /// Return the (raw) singleton object registered under the given name. + /// + /// + /// + /// Only checks already instantiated singletons; does not return an Object + /// for singleton object definitions which have not been instantiated yet. + /// + /// + /// The main purpose of this method is to access manually registered singletons + /// . Can also be used to access a singleton + /// defined by an object definition that already been created, in a raw fashion. + /// + /// + /// Name of the object to look for. + /// the registered singleton object, or null if none found + /// + object GetSingleton(string objectName); + + + /// + /// Check if this registry contains a singleton instance with the given name. + /// + /// + /// + /// Only checks already instantiated singletons; does not return true + /// for singleton bean definitions which have not been instantiated yet. + /// + /// + /// The main purpose of this method is to check manually registered singletons + /// . Can also be used to check whether a + /// singleton defined by an object definition has already been created. + /// + /// + /// To check whether an object factory contains an object definition with a given name, + /// use ListableBeanFactory's ContainsObjectDefinition. Calling both + /// ContainsObjectDefinition and ContainsSingleton answers + /// whether a specific object factory contains an own object with the given name. + /// + /// + /// Use IObjectFactory's ContainsObject 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. + /// + /// + /// Name of the object to look for. + /// + /// true if this bean factory contains a singleton instance with the given name; otherwise, false. + /// + /// + /// + /// + bool ContainsSingleton(string objectName); + + /// + /// Gets the names of singleton objects registered in this registry. + /// + /// + /// + /// Only checks already instantiated singletons; does not return names + /// for singleton bean definitions which have not been instantiated yet. + /// + /// + /// The main purpose of this method is to check manually registered singletons + /// . Can also be used to check which + /// singletons defined by an object definition have already been created. + /// + /// + /// The list of names as String array (never null). + /// + /// + /// + string[] SingletonNames + { + get; + } + + /// + /// Gets the number of singleton beans registered in this registry. + /// + /// + /// + /// Only checks already instantiated singletons; does not count + /// singleton object definitions which have not been instantiated yet. + /// + /// + /// The main purpose of this method is to check manually registered singletons + /// . Can also be used to count the number of + /// singletons defined by an object definition that have already been created. + /// + /// + /// The number of singleton objects. + /// + /// + /// + int SingletonCount + { + get; + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Core/Objects/Factory/Support/AbstractObjectFactory.cs b/src/Spring/Spring.Core/Objects/Factory/Support/AbstractObjectFactory.cs index 932e30d7..98cf1956 100644 --- a/src/Spring/Spring.Core/Objects/Factory/Support/AbstractObjectFactory.cs +++ b/src/Spring/Spring.Core/Objects/Factory/Support/AbstractObjectFactory.cs @@ -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; + + /// + /// Set of registered singletons, containing the bean names in registration order + /// + private ISet registeredSingletons = new HashedSet(); + private IDictionary singletonsInCreation; #endregion @@ -1693,18 +1701,7 @@ namespace Spring.Objects.Factory.Support return GetObjectForInstance(name, instance); } - /// - /// Tries to find a cached object for the specified name. - /// - /// Teh object name to look for. - /// The cached object if found, otherwise. - protected virtual object GetSingleton(string objectName) - { - lock (singletonCache) - { - return singletonCache[objectName]; - } - } + /// /// Creates a singleton instance for the specified object name and definition. @@ -1796,8 +1793,6 @@ namespace Spring.Objects.Factory.Support #endregion - #region IConfigurableObjectFactory Members - /// /// Destroy all cached singletons in this factory. /// @@ -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. /// - /// . + /// . 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 ? /// - /// + /// public bool ContainsSingleton(string name) { AssertUtils.ArgumentHasText(name, "name"); @@ -1986,6 +1981,84 @@ namespace Spring.Objects.Factory.Support } } + #region ISingletonObjectRegistry Members + + + /// + /// Gets the names of singleton objects registered in this registry. + /// + /// The list of names as String array (never null). + /// + /// + /// Only checks already instantiated singletons; does not return names + /// for singleton bean definitions which have not been instantiated yet. + /// + /// + /// The main purpose of this method is to check manually registered singletons + /// . Can also be used to check which + /// singletons defined by an object definition have already been created. + /// + /// + /// + /// + /// + public string[] SingletonNames + { + get + { + lock (singletonCache) + { + return + StringUtils.DelimitedListToStringArray( + StringUtils.CollectionToDelimitedString(registeredSingletons, ","), ","); + + } + } + } + + /// + /// Gets the number of singleton beans registered in this registry. + /// + /// The number of singleton objects. + /// + /// + /// Only checks already instantiated singletons; does not count + /// singleton object definitions which have not been instantiated yet. + /// + /// + /// The main purpose of this method is to check manually registered singletons + /// . Can also be used to count the number of + /// singletons defined by an object definition that have already been created. + /// + /// + /// + /// + /// + public int SingletonCount + { + get + { + lock (singletonCache) + { + return registeredSingletons.Count; + } + } + } + /// + /// Tries to find a cached object for the specified name. + /// + /// Teh object name to look for. + /// The cached object if found, otherwise. + public virtual object GetSingleton(string objectName) + { + lock (singletonCache) + { + return singletonCache[objectName]; + } + } + #endregion + + } } \ No newline at end of file diff --git a/src/Spring/Spring.Core/Spring.Core.2005.csproj b/src/Spring/Spring.Core/Spring.Core.2005.csproj index 17f5fc53..ec23a6b0 100644 --- a/src/Spring/Spring.Core/Spring.Core.2005.csproj +++ b/src/Spring/Spring.Core/Spring.Core.2005.csproj @@ -177,6 +177,7 @@ Code + Code @@ -546,6 +547,7 @@ + diff --git a/src/Spring/Spring.Messaging.Nms/Context/ILifecycle.cs b/src/Spring/Spring.Messaging.Nms/Context/ILifecycle.cs deleted file mode 100644 index a939cee0..00000000 --- a/src/Spring/Spring.Messaging.Nms/Context/ILifecycle.cs +++ /dev/null @@ -1,15 +0,0 @@ - -namespace Spring.Context -{ - interface ILifecycle - { - void Start(); - - void Stop(); - - bool IsRunning - { - get; - } - } -} diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsResourceHolder.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsResourceHolder.cs index fa0b9006..16ba704f 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsResourceHolder.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsResourceHolder.cs @@ -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 /// Create a new NmsResourceHolder that is open for resources to be added. public NmsResourceHolder() { - this.frozen = false; + } + + + /// + /// Initializes a new instance of the class + /// at is open for resources to be added. + /// + /// The connection factory that this + /// resource holder is associated with (may be null) + /// + public NmsResourceHolder(IConnectionFactory connectionFactory) + { + this.connectionFactory = connectionFactory; + } + + /// + /// Initializes a new instance of the class for the + /// given Session. + /// + /// The session. + public NmsResourceHolder(ISession session) + { + AddSession(session); + frozen = true; } /// Create a new NmsResourceHolder for the given NMS resources. diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/SingleConnectionFactory.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/SingleConnectionFactory.cs index 4b12520e..93f6ced9 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/SingleConnectionFactory.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/SingleConnectionFactory.cs @@ -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() diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractMessageListenerContainer.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractMessageListenerContainer.cs index ac2fb107..78dff4f6 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractMessageListenerContainer.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractMessageListenerContainer.cs @@ -1,16 +1,46 @@ +#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 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 { + /// + /// Abstract base class for message listener containers. Can either host + /// a standard NMS or a Spring-specific + /// + /// 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 } + /// + /// Gets or sets the message listener to register. + /// + /// + /// + /// + /// This can be either a standard NMS object or a + /// Spring object. + /// + /// + /// The message listener. 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 } + /// + /// Gets or sets a value indicating whether to accept messages while + /// the listener container is in the process of stopping. + /// + /// + /// + /// 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). + /// + /// + /// 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. + /// + /// + /// + /// true if accept messages while in the process of stopping; otherwise, false. + /// + 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 + + + + /// + /// Executes the specified listener, + /// committing or rolling back the transaction afterwards (if necessary). + /// + /// The session to operate on. + /// The received message. + /// + /// + /// + /// 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); } } - + /// + /// Executes the specified listener, + /// committing or rolling back the transaction afterwards (if necessary). + /// + /// The session to operate on. + /// The received message. + /// If thrown by NMS API methods. + /// + /// + /// 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) + /// + /// Invokes the specified listener: either as standard NMS IMessageListener + /// or (preferably) as Spring SessionAwareMessageListener. + /// + /// The session to operate on. + /// The received message. + /// If thrown by NMS API methods. + /// + 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"); } } - /// - /// Invoke the specified listener as standard JMS MessageListener. - /// - /// Default implementation performs a plain invocation of the - /// OnMessage methods - /// The listener to invoke. - /// The received message. - /// if thronw by the underlying NMS APIs - protected virtual void DoInvokeListener(IMessageListener listener, IMessage message) - { - listener.OnMessage(message); - } - /// /// 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 /// The Spring ISessionAwareMessageListener to invoke. /// The session to operate on. /// The received message. + /// If thrown by NMS API methods. + /// + /// 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) + + /// + /// Invoke the specified listener as standard JMS MessageListener. + /// + /// Default implementation performs a plain invocation of the + /// OnMessage methods + /// The listener to invoke. + /// The received message. + /// if thrown by the NMS API methods + protected virtual void DoInvokeListener(IMessageListener listener, IMessage message) { + listener.OnMessage(message); + } + + /// + /// Perform a commit or message acknowledgement, as appropriate + /// + /// The session to commit. + /// The message to acknowledge. + /// In case of commit failure + 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(); + } + } + + + /// + /// Perform a rollback, if appropriate. + /// + /// The session to rollback. + /// In case of a rollback error + protected virtual void RollbackIfNecessary(ISession session) + { + if (session.Transacted && SessionTransacted) + { + // Transacted session created by this container -> rollback + NmsUtils.RollbackIfNecessary(session); + } + } + /// + /// Perform a rollback, handling rollback excepitons properly. + /// + /// The session to rollback. + /// The thrown application exception. + /// in case of a rollback error. + 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; + } + } + + + /// + /// Handle the given exception that arose during listener execution. + /// + /// + /// 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. + /// + /// The exceptin to handle + 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) + /// + /// Invokes the registered exception listener, if any. + /// + /// The exception that arose during NMS processing. + /// + 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 } } } + + /// + /// Internal exception class that indicates a rejected message on shutdown. + /// Used to trigger a rollback for an external transaction manager in that case. + /// + internal class MessageRejectedWhileStoppingException : ApplicationException + { + } } diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractNmsListeningContainer.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractNmsListeningContainer.cs index a0ab6e28..92b05ac4 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractNmsListeningContainer.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractNmsListeningContainer.cs @@ -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 /// Mark Pollack 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; } + } + + /// + /// Gets a value indicating whether this container is currently running, + /// that is, whether it has been started and not stopped yet. + /// + /// + /// true if this container is running; otherwise, false. + /// public bool IsRunning { get { lock (lifecycleMonitor) { - return running; + return (running && RunningAllowed); } } } - /// Return whether a shared NMS IConnection should be maintained - /// by this listener container base class. + /// + /// Gets a value indicating whether this container's listeners are generally allowed to run. /// - /// - /// - protected abstract bool SharedConnectionEnabled { get; } - - public void Dispose() + /// + /// + /// >This implementation always returns true; the default 'running' + /// state is purely determined by /. + /// + /// + /// 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 + /// false if such a restriction prevents listeners from running. + /// + /// + /// true if running allowed; otherwise, false. + 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 + /// Return whether a shared NMS IConnection should be maintained + /// by this listener container base class. + /// + /// + /// + protected abstract bool SharedConnectionEnabled { get; } + + /// + /// Gets the shared connection maintained by this container. + /// Available after initialization. + /// + /// The shared connection (never null) + /// if this container does not maintain a + /// shared Connection, or if the Connection hasn't been initialized yet. + /// + /// + 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(); + } + + /// + /// Validates the configuration of this container. The default implementation + /// is empty. To be overriden in subclasses. + /// + protected virtual void ValidateConfiguration() + { + + } + + public void Dispose() + { + Shutdown(); } - public void Start() + /// + /// Initializes this container. Creates a Connection, starts the Connection + /// (if the property hasn't been turned off), and calls + /// . + /// + /// If startup failed + 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() + /// + /// Starts this container. + /// + /// if starting failed. + 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() + /// + /// Stops this container. + /// + /// if stopping failed. + public void Stop() + { + DoStop(); + } + + /// + /// Notify all invoker tasks and stop the shared Connection, if any. + /// + /// if thrown by NMS API methods. + /// + protected virtual void DoStop() + { + lock (this.lifecycleMonitor) + { + this.running = false; + System.Threading.Monitor.PulseAll(this.lifecycleMonitor); + } + + if (SharedConnectionEnabled) + { + StopSharedConnection(); + } + } + + /// + /// 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. + /// + protected abstract void DoInitialize(); + + + /// + /// 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. + /// + protected abstract void DoShutdown(); + + + /// + /// Establishes a shared Connection for this container. + /// + /// + /// + /// The default implementation delegates to + /// 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. + /// + /// + /// If thrown by NMS API methods + 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(); - } - - + /// + /// Refreshes the shared connection that this container holds. + /// + /// + /// Called on startup and also after an infrastructure exception + /// that occurred during invoker setup and/or execution. + /// + /// If thrown by NMS API methods 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; + }*/ + } + + /// + /// Creates the shared connection for this container. + /// + /// + /// The default implementation creates a standard Connection + /// and prepares it through + /// + /// the prepared Connection + /// if the creation failed. + protected virtual IConnection CreateSharedConnection() + { + IConnection con = CreateConnection(); + try + { + PrepareSharedConnection(con); + return con; + } catch (NMSException ex) + { + NmsUtils.CloseConnection(con); + throw; } } + /// + /// Prepares the given connection, which is about to be registered + /// as shared Connection for this container. + /// + /// + /// The default implementation sets the specified client id, if any. + /// Subclasses can override this to apply further settings. + /// + /// The connection to prepare. + /// If the preparation efforts failed. protected virtual void PrepareSharedConnection(IConnection connection) { if (ClientId != null) @@ -284,33 +453,28 @@ namespace Spring.Messaging.Nms.Listener } } - /// Register the specified listener on the underlying NMS IConnection. - ///

Subclasses need to implement this method for their specific - /// listener management process.

+ + /// + /// Starts the shared connection. /// - /// NMSException if registration failed - /// - /// - /// - /// - protected abstract void RegisterListener(); - - public void Stop() + /// If thrown by NMS API methods + /// + 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); } } } } + } - - /// Destroy the registered listener. - /// The NMS IConnection will automatically be closed afterwards - ///

Subclasses need to implement this method for their specific - /// listener management process.

+ /// + /// 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. + /// + public class SharedConnectionNotInitializedException : ApplicationException + { + /// + /// Initializes a new instance of the class. /// - /// NMSException if destruction failed - protected abstract void DestroyListener(); + /// The message. + public SharedConnectionNotInitializedException(string message) : base(message) + { + } } } \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/LocallyExposedNmsResourceHolder.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/LocallyExposedNmsResourceHolder.cs new file mode 100644 index 00000000..16cb59fc --- /dev/null +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/LocallyExposedNmsResourceHolder.cs @@ -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 +{ + /// + /// NmsResourceHolder marker subclass that indicates local exposure, + /// i.e. that does not indicate an externally managed transaction. + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class LocallyExposedNmsResourceHolder : NmsResourceHolder + { + /// + /// Initializes a new instance of the class. + /// + /// The session. + public LocallyExposedNmsResourceHolder(ISession session) : base(session) + { + + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/SimpleMessageListenerContainer.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/SimpleMessageListenerContainer.cs index e3214354..50f10c07 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/SimpleMessageListenerContainer.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/SimpleMessageListenerContainer.cs @@ -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 + /// + /// Message listener container that uses the plain NMS client API's + /// method to create concurrent + /// MessageConsumers for the specified listeners. + /// + 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; + } + } + + /// + /// Always use a shared NMS connection + /// 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"); } } + /// + /// Creates the specified number of concurrent consumers, + /// in the form of a JMS Session plus associated MessageConsumer + /// + /// + protected override void DoInitialize() + { + EstablishSharedConnection(); + InitializeConsumers(); + } + + /// + /// Re-initializes this container's NMS message consumers, + /// if not initialized already. + /// + protected override void DoStart() + { + base.DoStart(); + InitializeConsumers(); + } + + /// + /// Registers this listener container as NMS ExceptionListener on the shared connection. + /// + /// + 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); + } + } + + /// + /// Initialize the Sessions and MessageConsumers for this container. + /// + /// in case of setup failure. + 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); + } + } + } + } + + /// + /// Creates a MessageConsumer for the given Session, + /// registering a MessageListener for the specified listener + /// + /// The session to work on. + /// the IMessageConsumer"/> + /// if thrown by NMS methods private IMessageConsumer CreateListenerConsumer(ISession session) { IDestination destination = Destination; @@ -55,7 +188,7 @@ namespace Spring.Messaging.Nms.Listener destination = ResolveDestinationName(session, DestinationName); } IMessageConsumer consumer = CreateConsumer(session, destination); - //TODO TaskExectuor abstraction would go here... + SimpleMessageListener listener = new SimpleMessageListener(this, session); consumer.Listener += new Apache.NMS.MessageListener(listener.OnMessage); @@ -66,36 +199,22 @@ namespace Spring.Messaging.Nms.Listener /// Close the message consumers and sessions. /// /// NMSException if destruction failed - protected override void DestroyListener() + 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; } - /// - /// Afters the properties set. - /// - 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); + } + } } } } diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/NmsTemplate.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/NmsTemplate.cs index 94c989b0..35c6e712 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/NmsTemplate.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/NmsTemplate.cs @@ -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 /// Mark Pollack (.NET) 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 } } - /// - /// Returns whether the ISession is in client acknowledgement mode. - /// - /// The session. - /// true ifin client ack mode, false otherwise - protected virtual bool ClientAcknowledge(ISession session) - { - return (session.AcknowledgementMode == AcknowledgementMode.ClientAcknowledge); - } - /// Receive a message synchronously from the default destination, but only /// wait up to a specified time for delivery. Convert the message into an diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Support/Destinations/NmsDestinationAccessor.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Support/Destinations/NmsDestinationAccessor.cs index 83d65c89..bf33d704 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Support/Destinations/NmsDestinationAccessor.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Support/Destinations/NmsDestinationAccessor.cs @@ -18,6 +18,8 @@ #endregion +using System; +using Spring.Objects.Factory; using Spring.Util; using Apache.NMS; diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Support/NmsAccessor.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Support/NmsAccessor.cs index fe17bd6c..dd0a17c3 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Support/NmsAccessor.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Support/NmsAccessor.cs @@ -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); } + + /// + /// Returns whether the ISession is in client acknowledgement mode. + /// + /// The session. + /// true ifin client ack mode, false otherwise + protected virtual bool ClientAcknowledge(ISession session) + { + return (session.AcknowledgementMode == AcknowledgementMode.ClientAcknowledge); + } } } diff --git a/src/Spring/Spring.Messaging.Nms/Spring.Messaging.Nms.2005.csproj b/src/Spring/Spring.Messaging.Nms/Spring.Messaging.Nms.2005.csproj index eac69da1..d181f9b1 100644 --- a/src/Spring/Spring.Messaging.Nms/Spring.Messaging.Nms.2005.csproj +++ b/src/Spring/Spring.Messaging.Nms/Spring.Messaging.Nms.2005.csproj @@ -46,7 +46,6 @@ - @@ -70,6 +69,7 @@ + diff --git a/src/Spring/Spring.Web/Objects/Factory/Support/WebObjectFactory.cs b/src/Spring/Spring.Web/Objects/Factory/Support/WebObjectFactory.cs index d3ca6dd5..f9c58e7e 100644 --- a/src/Spring/Spring.Web/Objects/Factory/Support/WebObjectFactory.cs +++ b/src/Spring/Spring.Web/Objects/Factory/Support/WebObjectFactory.cs @@ -206,7 +206,7 @@ namespace Spring.Objects.Factory.Support /// /// Object name to look for. /// Cached object if found, null otherwise. - protected override object GetSingleton(string objectName) + public override object GetSingleton(string objectName) { object instance = null; diff --git a/test/Spring/Spring.Core.Tests/CommonTypes.cs b/test/Spring/Spring.Core.Tests/CommonTypes.cs index 7f64d128..23150d5b 100644 --- a/test/Spring/Spring.Core.Tests/CommonTypes.cs +++ b/test/Spring/Spring.Core.Tests/CommonTypes.cs @@ -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; } diff --git a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/LoggingExceptionHandler.cs b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/LoggingExceptionHandler.cs new file mode 100644 index 00000000..49af545a --- /dev/null +++ b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/LoggingExceptionHandler.cs @@ -0,0 +1,28 @@ +using System; +using Spring.Messaging.Nms; +using Common.Logging; + +namespace Spring.Messaging.Nms.Integration +{ + /// + /// + /// + 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 + } +} \ No newline at end of file diff --git a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListener.cs b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListener.cs new file mode 100644 index 00000000..f75075e5 --- /dev/null +++ b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListener.cs @@ -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 + } +} \ No newline at end of file diff --git a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListenerContainerTests.cs b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListenerContainerTests.cs new file mode 100644 index 00000000..3c9c831d --- /dev/null +++ b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListenerContainerTests.cs @@ -0,0 +1,89 @@ +#region License + +/* + * Copyright © 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 +{ + /// + /// This class contains tests for + /// + /// Mark Pollack + /// $Id:$ + [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" }; } + } + + + } +} \ No newline at end of file diff --git a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListenerContainerTests.xml b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListenerContainerTests.xml new file mode 100644 index 00000000..8b0d49c3 --- /dev/null +++ b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Integration/SimpleMessageListenerContainerTests.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/test/Spring/Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Tests.2005.csproj b/test/Spring/Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Tests.2005.csproj index e9cea6ca..218bbade 100644 --- a/test/Spring/Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Tests.2005.csproj +++ b/test/Spring/Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Tests.2005.csproj @@ -37,6 +37,10 @@ False ..\..\..\lib\Net\2.0\Apache.NMS.dll + + False + ..\..\..\lib\Net\2.0\Apache.NMS.ActiveMQ.dll + False ..\..\..\lib\Net\2.0\Common.Logging.dll @@ -74,6 +78,10 @@ {AEB1578C-9018-4D49-B440-789F38DD2F29} Spring.Messaging.Nms.2005 + + {ED204A7B-832F-44C7-BFE3-504AEBE1BCC8} + Spring.Testing.NUnit.2005 + {44B16BAA-6DF8-447C-9D7F-3AD3D854D904} Spring.Core.Tests.2005 @@ -87,8 +95,12 @@ + + + + Always diff --git a/test/Spring/Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Tests.dll.config b/test/Spring/Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Tests.dll.config index 98ee661f..89bfe571 100644 --- a/test/Spring/Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Tests.dll.config +++ b/test/Spring/Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Tests.dll.config @@ -17,9 +17,20 @@ limitations under the License. - + +
+ - + + + + + + + + + +