diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractMessageListenerContainer.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractMessageListenerContainer.cs new file mode 100644 index 00000000..9af88517 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractMessageListenerContainer.cs @@ -0,0 +1,598 @@ +#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.Messaging.Ems.Core; +using Spring.Messaging.Ems.Support; +using Spring.Util; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Listener +{ + /// + /// Abstract base class for message listener containers. Can either host + /// a standard EMS MessageListener or a Spring-specific + /// + /// + public abstract class AbstractMessageListenerContainer : AbstractEmsListeningContainer + { + #region Logging + + private readonly ILog logger = LogManager.GetLogger(typeof(AbstractMessageListenerContainer)); + + #endregion + + #region Fields + + private object destination; + + private String messageSelector; + + private object messageListener; + + private bool subscriptionDurable = false; + + private string durableSubscriptionName; + + private IExceptionListener exceptionListener; + + private bool exposeListenerSession = true; + + private bool acceptMessagesWhileStopping = false; + + #endregion + + #region Properties + + /// + /// Gets or sets the destination to receive messages from. Will be null + /// if the configured destination is not an actual Destination type; + /// c.f. when the destination is a String. + /// + /// The destination. + public Destination Destination + { + get + { + return (this.destination is Destination ? (Destination) this.destination : null); + } + set + { + AssertUtils.ArgumentNotNull(value, "destination"); + destination = value; + if (destination is Topic && !(destination is Queue)) + { + PubSubDomain = true; + } + + } + } + + + /// + /// Gets or sets the name of the destination to receive messages from. + /// Will be null if the configured destination is not a + /// string type; c.f. when it is an actual Destination object. + /// + /// The name of the destination. + public string DestinationName + { + get + { + return (this.destination is string ? (string) this.destination : null); + + } + set + { + AssertUtils.ArgumentNotNull(value, "destinationName must not be null"); + this.destination = value; + } + } + + + /// + /// Gets or sets the message selector. + /// + /// The message selector expression (or null if none).. + public string MessageSelector + { + get { return messageSelector; } + set { messageSelector = value; } + } + + + /// + /// Gets or sets the message listener to register. + /// + /// + /// + /// + /// This can be either a standard EMS MessageListener object or a + /// Spring object. + /// + /// + /// The message listener. + public object MessageListener + { + set + { + CheckMessageListener(value); + if (durableSubscriptionName == null) + { + // Use message listener class name as default name for a durable subscription. + durableSubscriptionName = value.GetType().FullName; + } + messageListener = value; + } + get + { + return messageListener; + } + } + + + /// + /// Gets or sets a value indicating whether the subscription is durable. + /// + /// + /// Set whether to make the subscription durable. The durable subscription name + /// to be used can be specified through the "DurableSubscriptionName" property. + /// Default is "false". Set this to "true" to register a durable subscription, + /// typically in combination with a "DurableSubscriptionName" value (unless + /// your message listener class name is good enough as subscription name). + /// + /// Only makes sense when listening to a topic (pub-sub domain). + /// + /// true if the subscription is durable; otherwise, false. + public bool SubscriptionDurable + { + get { return subscriptionDurable; } + set { subscriptionDurable = value; } + } + + + /// + /// Gets or sets the name of the durable subscription to create. + /// + /// + /// To be applied in case of a topic (pub-sub domain) with subscription durability activated. + /// The durable subscription name needs to be unique within this client's + /// client id. Default is the class name of the specified message listener. + /// Note: Only 1 concurrent consumer (which is the default of this + /// message listener container) is allowed for each durable subscription. + /// + /// + /// The name of the durable subscription. + public string DurableSubscriptionName + { + get + { + return durableSubscriptionName; + } + set + { + AssertUtils.ArgumentNotNull(value, "durableSubscriptionName must not be null"); + durableSubscriptionName = value; + } + } + + + /// + /// Gets or sets the exception listener to notify in case of a EMSException thrown + /// by the registered message listener or the invocation infrastructure. + /// + /// The exception listener. + public IExceptionListener ExceptionListener + { + get { return exceptionListener; } + set { exceptionListener = value; } + } + + + /// + /// Gets or sets a value indicating whether to expose listener session to a registered + /// as well as to calls. + /// + /// + /// Default is "true", reusing the listener's Session. + /// Turn this off to expose a fresh Session fetched from the same + /// underlying Connection instead, which might be necessary + /// on some messaging providers. + /// Note that Sessions managed by an external transaction manager will + /// always get exposed to + /// calls. So in terms of EmsTemplate exposure, this setting only affects + /// locally transacted Sessions. + /// + /// + /// + /// true if expose listener session; otherwise, false. + /// + public bool ExposeListenerSession + { + get { return exposeListenerSession; } + set { exposeListenerSession = value; } + } + + + /// + /// 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; } + } + + + #endregion + + + + /// + /// Validate that the destination is not null and that if the subscription is durable, then we are not + /// using the Pub/Sub domain. + /// + 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(Session session, Message message) + { + try + { + DoExecuteListener(session, message); + } + 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 EMS API methods. + /// + /// + /// + protected virtual void DoExecuteListener(Session session, Message 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); + } + catch (Exception ex) + { + RollbackOnExceptionIfNecessary(session, ex); + throw; + } + CommitIfNecessary(session, message); + } + + /// + /// Invokes the specified listener: either as standard EMS MessageListener + /// or (preferably) as Spring ISessionAwareMessageListener. + /// + /// The session to operate on. + /// The received message. + /// If thrown by EMS API methods. + /// + protected virtual void InvokeListener(Session session, Message message) + { + object listener = MessageListener; + if (listener is ISessionAwareMessageListener) + { + DoInvokeListener((ISessionAwareMessageListener) listener, session, message); + } + + else if (listener is IMessageListener) + { + DoInvokeListener((IMessageListener)listener, message); + } + else if (listener != null) + { + throw new ArgumentException("Only MessageListener and ISessionAwareMessageListener supported"); + } + else + { + throw new InvalidOperationException("No message listener specified - see property MessageListener"); + } + } + + /// + /// Invoke the specified listener as Spring ISessionAwareMessageListener, + /// exposing a new EMS Session (potentially with its own transaction) + /// to the listener if demanded. + /// + /// The Spring ISessionAwareMessageListener to invoke. + /// The session to operate on. + /// The received message. + /// If thrown by EMS API methods. + /// + /// + protected virtual void DoInvokeListener(ISessionAwareMessageListener listener, Session session, Message message) + { + Connection conToClose = null; + Session sessionToClose = null; + try + { + Session sessionToUse = session; + if (!ExposeListenerSession) + { + //We need to expose a separate Session. + conToClose = CreateConnection(); + sessionToClose = CreateSession(conToClose); + sessionToUse = sessionToClose; + } + // Actually invoke the message listener + if (logger.IsDebugEnabled) + { + logger.Debug("Invoking listener with message of type [" + message.GetType() + + "] and session [" + sessionToUse + "]"); + } + listener.OnMessage(message, sessionToUse); + // Clean up specially exposed Session, if any + if (sessionToUse != session) + { + if (sessionToUse.Transacted && SessionTransacted) + { + // Transacted session created by this container -> commit. + EmsUtils.CommitIfNecessary(sessionToUse); + } + } + } finally + { + EmsUtils.CloseSession(sessionToClose); + EmsUtils.CloseConnection(conToClose); + } + } + + /// + /// 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 EMS API methods + protected virtual void DoInvokeListener(IMessageListener listener, Message 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(Session session, Message message) + { + // Commit session or acknowledge message + if (session.Transacted) + { + // Commit necessary - but avoid commit call is Session transaction is externally coordinated. + if (IsSessionLocallyTransacted(session)) + { + EmsUtils.CommitIfNecessary(session); + } + } + else if (IsClientAcknowledge(session)) + { + message.Acknowledge(); + } + } + + /// + /// Determines whether the given Session is locally transacted, that is, whether + /// its transaction is managed by this listener container's Session handling + /// and not by an external transaction coordinator. + /// + /// + /// The Session's own transacted flag will already have been checked + /// before. This method is about finding out whether the Session's transaction + /// is local or externally coordinated. + /// + /// The session to check. + /// + /// true if the is session locally transacted; otherwise, false. + /// + /// + protected virtual bool IsSessionLocallyTransacted(Session session) + { + return SessionTransacted; + } + + + /// + /// Perform a rollback, if appropriate. + /// + /// The session to rollback. + /// In case of a rollback error + protected virtual void RollbackIfNecessary(Session session) + { + if (session.Transacted && IsSessionLocallyTransacted(session)) + { + // Transacted session created by this container -> rollback + EmsUtils.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(Session session, Exception ex) + { + try + { + if (session.Transacted && IsSessionLocallyTransacted(session)) + { + // Transacted session created by this container -> rollback + if (logger.IsDebugEnabled) + { + logger.Debug("Initiating transaction rollback on application exception"); + } + EmsUtils.RollbackIfNecessary(session); + } + } catch (EMSException) + { + 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 EMSException) + { + InvokeExceptionListener((EMSException)ex); + } + if (Active) + { + // Regular case: failed while active. + // Log at error level. + logger.Error("Execution of EMS message listener failed", ex); + } + else + { + // Rare case: listener thread failed after container shutdown. + // Log at debug level, to avoid spamming the shutdown log. + logger.Debug("Listener exception after container shutdown", ex); + } + } + + /// + /// Invokes the registered exception listener, if any. + /// + /// The exception that arose during EMS processing. + /// + protected virtual void InvokeExceptionListener(EMSException ex) + { + IExceptionListener exListener = ExceptionListener; + if (exListener != null) + { + exListener.OnException(ex); + } + } + + #endregion + + /// + /// Checks the message listener, throwing an exception + /// if it does not correspond to a supported listener type. + /// By default, only a standard JMS MessageListener object or a + /// Spring object will be accepted. + /// + /// The message listener. + protected virtual void CheckMessageListener(object messageListener) + { + AssertUtils.ArgumentNotNull(messageListener, "Message Listener can not be null"); + if (!(messageListener is IMessageListener || messageListener is ISessionAwareMessageListener)) + { + throw new ArgumentException("messageListener needs to be of type [" + typeof(IMessageListener).FullName + "] or [" + typeof(ISessionAwareMessageListener).FullName + "]"); + } + } + } + + /// + /// 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.Ems/Messaging/Ems/Listener/AbstractemsListeningContainer.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractemsListeningContainer.cs new file mode 100644 index 00000000..9387fd36 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractemsListeningContainer.cs @@ -0,0 +1,546 @@ +#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.Context; +using Spring.Messaging.Ems.Connections; +using Spring.Messaging.Ems.Support; +using Spring.Messaging.Ems.Support.Destinations; +using Spring.Objects.Factory; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Listener +{ + /// + /// Common base class for all containers which need to implement listening + /// based on a Connection (either shared or freshly obtained for each attempt). + /// Inherits basic Connection and Session configuration handling from the + /// base class. + /// + /// + /// This class provides basic lifecycle management, in particular management + /// of a shared Connection. Subclasses are supposed to plug into this + /// lifecycle, implementing the as well as + /// + /// + /// + /// + /// + /// Mark Pollack + public abstract class AbstractEmsListeningContainer : EmsDestinationAccessor, ILifecycle, IObjectNameAware, IDisposable + { + #region Logging + + private readonly ILog logger = LogManager.GetLogger(typeof(AbstractEmsListeningContainer)); + + #endregion + + #region Fields + + private String clientId; + + private bool autoStartup = true; + + private string objectName; + + private Connection sharedConnection; + + private bool sharedConnectionStarted = false; + + /// + /// The monitor object to lock on when performing operations on the connection. + /// + protected object sharedConnectionMonitor = new object(); + + private volatile bool active = false; + + private bool running = false; + + /// + /// The monitor object to lock on when performing operations that update the lifecycle of the container. + /// + protected object lifecycleMonitor = new object(); + + #endregion + + /// + /// Gets or sets the client id for a shared Connection created and used by this container. + /// + /// + /// Note that client ids need to be unique among all active Connections + /// of the underlying JMS provider. Furthermore, a client id can only be + /// assigned if the original ConnectionFactory hasn't already assigned one. + /// + /// The client id. + public string ClientId + { + set { clientId = value; } + get { return clientId; } + } + + /// Set whether to automatically start the listener after initialization. + ///

Default is "true"; set this to "false" to allow for manual startup.

+ /// + public virtual bool AutoStartup + { + set { this.autoStartup = value; } + } + + /// + /// Set the name of the object in the object factory that created this object. + /// + /// The name of the object in the factory. + /// + ///

+ /// Invoked after population of normal object properties but before an init + /// callback like 's + /// + /// method or a custom init-method. + ///

+ /// + 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 && RunningAllowed); + } + } + } + + /// + /// Gets a value indicating whether this container's listeners are generally allowed to run. + /// + /// + /// + /// >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 + { + get { + return true; } + } + + /// + /// Gets a value indicating whether this container is currently active, + /// that is, whether it has been set up but not shut down yet. + /// + /// true if active; otherwise, false. + public virtual bool Active + { + get + { + lock (this.lifecycleMonitor) + { + return this.active; + } + } + + } + + /// Return whether a shared EMS Connection 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 Connection SharedConnection + { + get + { + if (!SharedConnectionEnabled) + { + throw new InvalidOperationException("This listener container does not maintain a shared Connection"); + } + lock (this.sharedConnectionMonitor) + { + if (this.sharedConnection == null) + { + throw new SharedConnectionNotInitializedException("This listener container's shared Connection has not been initialized yet"); + } + return this.sharedConnection; + } + } + } + + /// + /// Call base class method, then and then + /// + 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() + { + + } + + /// + /// Calls when the application context destroys the container instance. + /// + public void Dispose() + { + Shutdown(); + } + + + /// + /// 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() + { + 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; + } + } + + /// + /// Stop the shared connection, call , and close this container. + /// + public virtual void Shutdown() + { + logger.Debug("Shutting down message listener container"); + bool wasRunning = false; + lock (this.lifecycleMonitor) + { + wasRunning = this.running; + this.running = false; + this.active = false; + System.Threading.Monitor.PulseAll(this.lifecycleMonitor); + } + + if (wasRunning && SharedConnectionEnabled) + { + try + { + StopSharedConnection(); + } catch (Exception ex) + { + logger.Debug("Could not stop EMS Connection on shutdown", ex); + } + } + + // Shut down the invokers + try + { + DoShutdown(); + } + finally + { + lock (this.sharedConnectionMonitor) + { + ConnectionFactoryUtils.ReleaseConnection(this.sharedConnection, ConnectionFactory, false); + } + } + } + + /// + /// Starts this container. + /// + /// if starting failed. + public void Start() + { + DoStart(); + } + + /// + /// Start the shared Connection, if any, and notify all invoker tasks. + /// + 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); + } + + // Start the shared Connection, if any. + if (SharedConnectionEnabled) + { + 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 EMS 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 EMS API methods + protected virtual void EstablishSharedConnection() + { + lock (sharedConnectionMonitor) + { + if (sharedConnection == null) + { + sharedConnection = CreateSharedConnection(); + logger.Debug("Established shared EMS Connection"); + } + } + } + + /// + /// 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 EMS API methods + protected void RefreshSharedConnection() + { + lock (sharedConnectionMonitor) + { + ConnectionFactoryUtils.ReleaseConnection(sharedConnection, ConnectionFactory, sharedConnectionStarted); + sharedConnection = CreateSharedConnection(); + if (sharedConnectionStarted) + { + sharedConnection.Start(); + } + } + } + + /// + /// 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 Connection CreateSharedConnection() + { + Connection con = CreateConnection(); + try + { + PrepareSharedConnection(con); + return con; + } catch (EMSException) + { + EmsUtils.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(Connection connection) + { + if (ClientId != null) + { + connection.ClientID = ClientId; + } + } + + + /// + /// Starts the shared connection. + /// + /// If thrown by EMS API methods + /// + protected virtual void StartSharedConnection() + { + lock (sharedConnectionMonitor) + { + if (sharedConnection != null) + { + try + { + sharedConnectionStarted = true; + sharedConnection.Start(); + } + catch (Exception ex) + { + logger.Warn("Ignoring Connection start exception - assuming already started", ex); + } + } + } + } + + /// + /// Stops the shared connection. + /// + /// if thrown by EMS API methods. + protected virtual void StopSharedConnection() + { + lock (this.sharedConnectionMonitor) + { + if (this.sharedConnection != null) + { + try + { + this.sharedConnectionStarted = false; + this.sharedConnection.Stop(); + } + catch (System.InvalidOperationException ex) + { + logger.Warn("Ignoring Connection stop exception - assuming already stopped", ex); + } + } + } + } + + } + + /// + /// 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 : EMSException + { + /// + /// Initializes a new instance of the class. + /// + /// The message. + public SharedConnectionNotInitializedException(string message) : base(message) + { + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/ListenerExecutionFailedException.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/ListenerExecutionFailedException.cs new file mode 100644 index 00000000..851e71c7 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/ListenerExecutionFailedException.cs @@ -0,0 +1,54 @@ +#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 TIBCO.EMS; + +namespace Spring.Messaging.Ems.Listener.Adapter +{ + /// + /// Exception to be thrown when the execution of a listener method failed. + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class ListenerExecutionFailedException : EMSException + { + + /// + /// Initializes a new instance of the class, with the specified message + /// + /// The message. + public ListenerExecutionFailedException(string message) : base(message) + { + } + + /// + /// Initializes a new instance of the class, with the specified message + /// and root cause exception + /// + /// The message. + /// The inner exception. + public ListenerExecutionFailedException(string message, Exception innerException) + : base(message) + { + LinkedException = innerException; + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/MessageListenerAdapter.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/MessageListenerAdapter.cs new file mode 100644 index 00000000..f895cec1 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/MessageListenerAdapter.cs @@ -0,0 +1,552 @@ +using System; +using System.Collections; +using System.Reflection; +using Common.Logging; +using Spring.Expressions; +using Spring.Messaging.Ems.Support; +using Spring.Messaging.Ems.Support.Converter; +using Spring.Messaging.Ems.Support.Destinations; +using Spring.Util; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Listener.Adapter +{ + /// + /// Message listener adapter that delegates the handling of messages to target + /// listener methods via reflection, with flexible message type conversion. + /// Allows listener methods to operate on message content types, completely + /// independent from the EMS API. + /// + /// + /// By default, the content of incoming messages gets extracted before + /// being passed into the target listener method, to let the target method + /// operate on message content types such as String or byte array instead of + /// the raw Message. Message type conversion is delegated to a Spring + /// . By default, a + /// will be used. (If you do not want such automatic message conversion taking + /// place, then be sure to set the property + /// to null.) + /// + /// If a target listener method returns a non-null object (typically of a + /// message content type such as String or byte array), it will get + /// wrapped in a EMS Message and sent to the response destination + /// (either the EMS "reply-to" destination or the + /// specified. + /// + /// + /// The sending of response messages is only available when + /// using the entry point (typically through a + /// Spring message listener container). Usage as standard EMS MessageListener + /// does not support the generation of response messages. + /// + /// Consult the reference documentation for examples of method signatures compliant with this + /// adapter class. + /// + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class MessageListenerAdapter : IMessageListener, ISessionAwareMessageListener + { + #region Logging + + private readonly ILog logger = LogManager.GetLogger(typeof (MessageListenerAdapter)); + + #endregion + + /// + /// The default handler method name. + /// + public static string ORIGINAL_DEFAULT_HANDLER_METHOD = "HandleMessage"; + + private object handlerObject; + + private string defaultHandlerMethod = ORIGINAL_DEFAULT_HANDLER_METHOD; + + private IExpression processingExpression; + + private object defaultResponseDestination; + + private DestinationResolver destinationResolver = new DynamicDestinationResolver(); + + private IMessageConverter messageConverter; + + /// + /// Initializes a new instance of the class with default settings. + /// + public MessageListenerAdapter() + { + InitDefaultStrategies(); + handlerObject = this; + processingExpression = Expression.Parse(defaultHandlerMethod + "(#convertedObject)"); + } + + /// + /// Initializes a new instance of the class for the given handler object + /// + /// The delegate object. + public MessageListenerAdapter(object handlerObject) + { + InitDefaultStrategies(); + this.handlerObject = handlerObject; + } + + /// + /// Gets or sets the handler object to delegate message listening to. + /// + /// + /// Specified listener methods have to be present on this target object. + /// If no explicit handler object has been specified, listener + /// methods are expected to present on this adapter instance, that is, + /// on a custom subclass of this adapter, defining listener methods. + /// + /// The handler object. + public object HandlerObject + { + get { return handlerObject; } + set { handlerObject = value; } + } + + /// + /// Gets or sets the default handler method to delegate to, + /// for the case where no specific listener method has been determined. + /// Out-of-the-box value is ("HandleMessage"}. + /// + /// The default handler method. + public string DefaultHandlerMethod + { + get { return defaultHandlerMethod; } + set + { + defaultHandlerMethod = value; + } + } + + + /// + /// Sets the default destination to send response messages to. This will be applied + /// in case of a request message that does not carry a "JMSReplyTo" field. + /// Response destinations are only relevant for listener methods that return + /// result objects, which will be wrapped in a response message and sent to a + /// response destination. + /// + /// Alternatively, specify a "DefaultResponseQueueName" or "DefaultResponseTopicName", + /// to be dynamically resolved via the DestinationResolver. + /// + /// + /// The default response destination. + public object DefaultResponseDestination + { + set { defaultResponseDestination = value; } + } + + /// + /// Sets the name of the default response queue to send response messages to. + /// This will be applied in case of a request message that does not carry a + /// "EMSReplyTo" field. + /// Alternatively, specify a JMS Destination object as "defaultResponseDestination". + /// + /// The name of the default response destination queue. + public string DefaultResponseQueueName + { + set { defaultResponseDestination = new DestinationNameHolder(value, false); } + } + + /// + /// Sets the name of the default response topic to send response messages to. + /// This will be applied in case of a request message that does not carry a + /// "ReplyTo" field. + /// Alternatively, specify a JMS Destination object as "defaultResponseDestination". + /// + /// The name of the default response destination topic. + public string DefaultResponseTopicName + { + set { defaultResponseDestination = new DestinationNameHolder(value, true); } + } + + + /// + /// Gets or sets the destination resolver that should be used to resolve response + /// destination names for this adapter. + /// The default resolver is a . + /// Specify another implementation, for other strategies, perhaps from a directory service. + /// + /// The destination resolver. + public DestinationResolver DestinationResolver + { + get { return destinationResolver; } + set + { + AssertUtils.ArgumentNotNull(value, "DestinationResolver must not be null"); + destinationResolver = value; + } + } + + /// + /// Gets or sets the message converter that will convert incoming JMS messages to + /// listener method arguments, and objects returned from listener + /// methods back to EMS messages. + /// + /// + /// The default converter is a {@link SimpleMessageConverter}, which is able + /// to handle BytesMessages}, TextMessages, MapMessages, and ObjectMessages. + /// + /// + /// The message converter. + public IMessageConverter MessageConverter + { + get { return messageConverter; } + set { messageConverter = value; } + } + + + /// + /// Standard JMS {@link MessageListener} entry point. + /// Delegates the message to the target listener method, with appropriate + /// conversion of the message arguments + /// + /// + /// + /// In case of an exception, the method will be invoked. + /// Note + /// Does not support sending response messages based on + /// result objects returned from listener methods. Use the + /// entry point (typically through a Spring + /// message listener container) for handling result objects as well. + /// + /// The incoming message. + public void OnMessage(Message message) + { + try + { + OnMessage(message, null); + } + catch (Exception e) + { + HandleListenerException(e); + } + } + + /// + /// Spring entry point. + /// + /// Delegates the message to the target listener method, with appropriate + /// conversion of the message argument. If the target method returns a + /// non-null object, wrap in a EMS message and send it back. + /// + /// + /// The incoming message. + /// The session to operate on. + public void OnMessage(Message message, Session session) + { + if (handlerObject != this) + { + if (typeof(ISessionAwareMessageListener).IsInstanceOfType(handlerObject)) + { + if (session != null) + { + ((ISessionAwareMessageListener)handlerObject).OnMessage(message, session); + return; + } + else if (!typeof(IMessageListener).IsInstanceOfType(handlerObject)) + { + throw new InvalidOperationException("MessageListenerAdapter cannot handle a " + + "ISessionAwareMessageListener delegate if it hasn't been invoked with a Session itself"); + } + } + if (typeof(IMessageListener).IsInstanceOfType(handlerObject)) + { + ((IMessageListener)handlerObject).OnMessage(message); + return; + } + } + + // Regular case: find a handler method reflectively. + object convertedMessage = ExtractMessage(message); + + + IDictionary vars = new Hashtable(); + vars["convertedObject"] = convertedMessage; + + //Need to parse each time since have overloaded methods and + //expression processor caches target of first invocation. + //TODO - check JIRA as I believe this has been fixed, otherwise, use regular reflection. -MLP + processingExpression = Expression.Parse(defaultHandlerMethod + "(#convertedObject)"); + + //Invoke message handler method and get result. + object result; + try + { + result = processingExpression.GetValue(handlerObject, vars); + } + catch (EMSException) + { + throw; + } + // Will only happen if dynamic method invocation falls back to standard reflection. + catch (TargetInvocationException ex) + { + Exception targetEx = ex.InnerException; + if (ObjectUtils.IsAssignable(typeof(EMSException), targetEx)) + { + throw ReflectionUtils.UnwrapTargetInvocationException(ex); + } + else + { + throw new ListenerExecutionFailedException("Listener method '" + defaultHandlerMethod + "' threw exception", targetEx); + } + } + catch (Exception ex) + { + throw new ListenerExecutionFailedException("Failed to invoke target method '" + defaultHandlerMethod + + "' with argument " + convertedMessage, ex); + } + + if (result != null) + { + HandleResult(result, message, session); + } + else + { + logger.Debug("No result object given - no result to handle"); + } + } + + /// + /// Initialize the default implementations for the adapter's strategies. + /// + protected virtual void InitDefaultStrategies() + { + MessageConverter = new SimpleMessageConverter(); + } + + /// + /// Handle the given exception that arose during listener execution. + /// The default implementation logs the exception at error level. + /// This method only applies when used as standard EMS MessageListener. + /// In case of the Spring mechanism, + /// exceptions get handled by the caller instead. + /// + /// + /// The exception to handle. + protected virtual void HandleListenerException(Exception ex) + { + logger.Error("Listener execution failed", ex); + } + + /// + /// Extract the message body from the given message. + /// + /// The message. + /// the content of the message, to be passed into the + /// listener method as argument + /// if thrown by EMS API methods + private object ExtractMessage(Message message) + { + IMessageConverter converter = MessageConverter; + if (converter != null) + { + return converter.FromMessage(message); + } + return message; + } + + /// + /// Gets the name of the listener method that is supposed to + /// handle the given message. + /// The default implementation simply returns the configured + /// default listener method, if any. + /// + /// The EMS request message. + /// The converted JMS request message, + /// to be passed into the listener method as argument. + /// the name of the listener method (never null) + /// if thrown by EMS API methods + protected virtual string GetHandlerMethodName(Message originalMessage, object extractedMessage) + { + return DefaultHandlerMethod; + } + + /// + /// Handles the given result object returned from the listener method, sending a response message back. + /// + /// The result object to handle (never null). + /// The original request message. + /// The session to operate on (may be null). + protected virtual void HandleResult(object result, Message request, Session session) + { + if (session != null) + { + if (logger.IsDebugEnabled) + { + logger.Debug("Listener method returned result [" + result + + "] - generating response message for it"); + } + Message response = BuildMessage(session, result); + PostProcessResponse(request, response); + Destination destination = GetResponseDestination(request, response, session); + SendResponse(session, destination, response); + } + else + { + if (logger.IsDebugEnabled) + { + logger.Debug("Listener method returned result [" + result + + "]: not generating response message for it because of no EMS Session given"); + } + } + } + + /// + /// Builds a JMS message to be sent as response based on the given result object. + /// + /// The JMS Session to operate on. + /// The content of the message, as returned from the listener method. + /// the JMS Message (never null) + /// If there was an error in message conversion + /// if thrown by EMS API methods + protected virtual Message BuildMessage(Session session, Object result) + { + IMessageConverter converter = MessageConverter; + if (converter != null) + { + return converter.ToMessage(result, session); + } + else + { + Message msg = result as Message; + if (msg == null) + { + throw new MessageConversionException( + "No IMessageConverter specified - cannot handle message [" + result + "]"); + } + return msg; + } + } + + /// + /// Post-process the given response message before it will be sent. The default implementation + /// sets the response's correlation id to the request message's correlation id. + /// + /// The original incoming message. + /// The outgoing JMS message about to be sent. + /// if thrown by EMS API methods + protected virtual void PostProcessResponse(Message request, Message response) + { + response.CorrelationID = request.CorrelationID; + } + + /// + /// Determine a response destination for the given message. + /// + /// + /// The default implementation first checks the JMS Reply-To + /// Destination of the supplied request; if that is not null + /// it is returned; if it is null, then the configured + /// default response destination} + /// is returned; if this too is null, then an + /// is thrown. + /// + /// + /// The original incoming message. + /// Tthe outgoing message about to be sent. + /// The session to operate on. + /// the response destination (never null) + /// if thrown by EMS API methods + /// if no destination can be determined. + protected virtual Destination GetResponseDestination(Message request, Message response, Session session) + { + Destination replyTo = request.ReplyTo; + if (replyTo == null) + { + replyTo = ResolveDefaultResponseDestination(session); + if (replyTo == null) + { + throw new InvalidDestinationException("Cannot determine response destination: " + + "Request message does not contain reply-to destination, and no default response destination set."); + } + } + return replyTo; + } + + /// + /// Resolves the default response destination into a Destination, using this + /// accessor's in case of a destination name. + /// + /// The session to operate on. + /// The located destination + protected virtual Destination ResolveDefaultResponseDestination(Session session) + { + Destination dest = defaultResponseDestination as Destination; + if (dest != null) + { + return dest; + } + + DestinationNameHolder destNameHolder = defaultResponseDestination as DestinationNameHolder; + if (destNameHolder != null) + { + return DestinationResolver.ResolveDestinationName(session, destNameHolder.Name, destNameHolder.IsTopic); + } + + return null; + } + + /// + /// Sends the given response message to the given destination. + /// + /// The session to operate on. + /// The destination to send to. + /// The outgoing message about to be sent. + protected virtual void SendResponse(Session session, Destination destination, Message response) + { + MessageProducer producer = session.CreateProducer(destination); + try + { + PostProcessProducer(producer, response); + producer.Send(response); + } + finally + { + EmsUtils.CloseMessageProducer(producer); + } + } + + /// + /// Post-process the given message producer before using it to send the response. + /// The default implementation is empty. + /// + /// The producer that will be used to send the message. + /// The outgoing message about to be sent. + protected virtual void PostProcessProducer(MessageProducer producer, Message response) + { + + } + } + + /// + /// Internal class combining a destination name and its target destination type (queue or topic). + /// + internal class DestinationNameHolder + { + private readonly string name; + + private readonly bool isTopic; + + public DestinationNameHolder(string name, bool isTopic) + { + this.name = name; + this.isTopic = isTopic; + } + + + public string Name + { + get { return name; } + } + + public bool IsTopic + { + get { return isTopic; } + } + } +} diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/ISessionAwareMessageListener.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/ISessionAwareMessageListener.cs new file mode 100644 index 00000000..c59f311f --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/ISessionAwareMessageListener.cs @@ -0,0 +1,51 @@ +#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 TIBCO.EMS; + +namespace Spring.Messaging.Ems.Listener +{ + /// + /// Variant of the standard EMS MessageListener interface, + /// offering not only the received Message but also the underlying + /// Session object. The latter can be used to send reply messages, + /// without the need to access an external Connection/Session, + /// i.e. without the need to access the underlying ConnectionFactory. + /// + /// + /// Supported by Spring's + /// as direct alternative to the standard MessageListener interface. + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public interface ISessionAwareMessageListener + { + /// Callback for processing a received EMS message. + /// Implementors are supposed to process the given Message, + /// typically sending reply messages through the given Session. + /// + /// the received EMS message + /// + /// the underlying EMS Session + /// + /// EMSException if thrown by EMS methods + void OnMessage(Message message, Session session); + } +} diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/LocallyExposedEmsResourceHolder.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/LocallyExposedEmsResourceHolder.cs new file mode 100644 index 00000000..4a13e426 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/LocallyExposedEmsResourceHolder.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 Spring.Messaging.Ems.Connections; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Listener +{ + /// + /// EmsResourceHolder marker subclass that indicates local exposure, + /// i.e. that does not indicate an externally managed transaction. + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class LocallyExposedEmsResourceHolder : EmsResourceHolder + { + /// + /// Initializes a new instance of the class. + /// + /// The session. + public LocallyExposedEmsResourceHolder(Session session) : base(session) + { + + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/SimpleMessageListenerContainer.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/SimpleMessageListenerContainer.cs new file mode 100644 index 00000000..19450da2 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/SimpleMessageListenerContainer.cs @@ -0,0 +1,398 @@ +#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.Threading; +using Common.Logging; +using Spring.Collections; +using Spring.Messaging.Ems.Support; +using Spring.Transaction.Support; +using Spring.Util; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Listener +{ + /// + /// Message listener container that uses the plain EMS client API's + /// MessageConsumer.Listener 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 + + /// + /// The default recovery time interval between connection reconnection attempts + /// + public static TimeSpan DEFAULT_RECOVERY_INTERVAL = new TimeSpan(0, 0, 0, 5, 0); + + /// + /// The total time connection recovery will be attempted. + /// + public static TimeSpan DEFAULT_MAX_RECOVERY_TIME = new TimeSpan(0, 0, 10, 0, 0); + + private bool pubSubNoLocal = false; + + private int concurrentConsumers = 1; + + private ISet sessions; + + private ISet consumers; + + private object consumersMonitor = new object(); + + private TimeSpan recoveryInterval = DEFAULT_RECOVERY_INTERVAL; + + private TimeSpan maxRecoveryTime = DEFAULT_MAX_RECOVERY_TIME; + + #endregion + + #region Properties + + /// + /// Gets or sets a value indicating whether to inhibit the delivery of messages published by its own connection. + /// Default is "false". + /// + /// true if should inhibit the delivery of messages published by its own connection; otherwise, false. + public bool PubSubNoLocal + { + get { return pubSubNoLocal; } + set { pubSubNoLocal = value; } + } + + /// + /// Specify the number of concurrent consumers to create. Default is 1. + /// + /// + /// Raising the number of concurrent consumers is recommendable in order + /// to scale the consumption of messages coming in from a queue. However, + /// note that any ordering guarantees are lost once multiple consumers are + /// registered. In general, stick with 1 consumer for low-volume queues. + /// Do not raise the number of concurrent consumers for a topic. + /// This would lead to concurrent consumption of the same message, + /// which is hardly ever desirable. + /// + /// + /// The concurrent consumers. + public int ConcurrentConsumers + { + set + { + AssertUtils.IsTrue(value > 0, "'ConcurrentConsumer' value must be at least 1 (one)"); + concurrentConsumers = value; + } + } + + /// + /// Sets the time interval between connection recovery attempts. The default is 5 seconds. + /// + /// The recovery interval. + public TimeSpan RecoveryInterval + { + set { recoveryInterval = value; } + } + + + /// + /// Sets the max recovery time to try reconnection attempts. The default is 10 minutes. + /// + /// The max recovery time. + public TimeSpan MaxRecoveryTime + { + set { maxRecoveryTime = value; } + } + + /// + /// Always use a shared EMS connection + /// + protected override bool SharedConnectionEnabled + { + get { return true; } + } + + #endregion + + /// + /// Call base class for valdation and then check that if the subscription is durable that the number of + /// concurrent consumers is equal to one. + /// + protected override void ValidateConfiguration() + { + base.ValidateConfiguration(); + if (SubscriptionDurable && concurrentConsumers !=1 ) + { + 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 EMS message consumers, + /// if not initialized already. + /// + protected override void DoStart() + { + base.DoStart(); + InitializeConsumers(); + } + + /// + /// Registers this listener container as EMS ExceptionListener on the shared connection. + /// + /// + protected override void PrepareSharedConnection(Connection connection) + { + base.PrepareSharedConnection(connection); + connection.ExceptionListener = this; + } + + + /// + /// implementation, invoked by the EMS provider in + /// case of connection failures. Re-initializes this listener container's + /// shared connection and its sessions and consumers. + /// + /// The reported connection exception. + public void OnException(EMSException exception) + { + // First invoke the user-specific ExceptionListener, if any. + InvokeExceptionListener(exception); + // now try to recover the shared Connection and all consumers... + if (logger.IsInfoEnabled) + { + logger.Info("Trying to recover from EMS Connection exception: " + exception); + } + try + { + lock (consumersMonitor) + { + sessions = null; + consumers = null; + } + RefreshConnectionUntilSuccessful(); + InitializeConsumers(); + logger.Info("Successfully refreshed EMS Connection"); + } + catch (RecoveryTimeExceededException) + { + throw; + } catch (EMSException recoverEx) + { + logger.Debug("Failed to recover EMS Connection", recoverEx); + logger.Error("Encountered non-recoverable EMSException", exception); + } + } + + /// + /// Refresh the underlying Connection, not returning before an attempt has been + /// successful. Called in case of a shared Connection as well as without shared + /// Connection, so either needs to operate on the shared Connection or on a + /// temporary Connection that just gets established for validation purposes. + /// + /// + /// The default implementation retries until it successfully established a + /// Connection, for as long as this message listener container is active. + /// Applies the specified recovery interval between retries. + /// + protected virtual void RefreshConnectionUntilSuccessful() + { + TimeSpan totalTryTime = new TimeSpan(); + while (IsRunning) + { + try + { + RefreshSharedConnection(); + break; + } + catch (Exception ex) + { + if (logger.IsInfoEnabled) + { + logger.Info("Could not refresh Connection - retrying in " + recoveryInterval, ex); + } + } + + if (totalTryTime > maxRecoveryTime) + { + logger.Info("Could not refresh Connection after " + totalTryTime + ". Stopping reconnection attempts."); + throw new RecoveryTimeExceededException("Could not recover after " + totalTryTime); + } + + DateTime startTime = DateTime.Now; + SleepInBetweenRecoveryAttempts(); + TimeSpan sleepTimeSpan = DateTime.Now - startTime; + totalTryTime += sleepTimeSpan; + } + } + + /// + /// The amount of time to sleep in between recovery attempts. + /// + protected virtual void SleepInBetweenRecoveryAttempts() + { + Thread.Sleep(recoveryInterval); + } + + + /// + /// 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(); + Connection con = SharedConnection; + for (int i = 0; i < this.concurrentConsumers; i++) + { + Session session = CreateSession(SharedConnection); + MessageConsumer 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 MessageConsumer"/> + /// if thrown by EMS methods + private MessageConsumer CreateListenerConsumer(Session session) + { + Destination destination = Destination; + if (destination == null) + { + destination = ResolveDestinationName(session, DestinationName); + } + MessageConsumer consumer = CreateConsumer(session, destination); + + consumer.MessageListener = new SimpleMessageListener(this, session); + return consumer; + } + + /// + /// Close the message consumers and sessions. + /// + /// EMSException if destruction failed + protected override void DoShutdown() + { + logger.Debug("Closing EMS MessageConsumers"); + foreach (MessageConsumer messageConsumer in consumers) + { + EmsUtils.CloseMessageConsumer(messageConsumer); + } + logger.Debug("Closing EMS Sessions"); + foreach (Session session in sessions) + { + EmsUtils.CloseSession(session); + } + consumers = null; + sessions = null; + } + + + /// + /// Creates a MessageConsumer for the given Session and Destination. + /// + /// The session to create a MessageConsumer for. + /// The destination to create a MessageConsumer for. + /// The new MessageConsumer + protected MessageConsumer CreateConsumer(Session session, Destination destination) + { + // Only pass in the NoLocal flag in case of a Topic: + // Some EMS providers, such as WebSphere MQ 6.0, throw IllegalStateException + // in case of the NoLocal flag being specified for a Queue. + if (PubSubDomain) + { + if (SubscriptionDurable && destination is Topic) + { + return session.CreateDurableSubscriber( + (Topic) destination, DurableSubscriptionName, MessageSelector, PubSubNoLocal); + } + else + { + return session.CreateConsumer(destination, MessageSelector, PubSubNoLocal); + } + } + else + { + return session.CreateConsumer(destination, MessageSelector); + } + } + } + + internal class SimpleMessageListener : IMessageListener + { + private SimpleMessageListenerContainer container; + private Session session; + + public SimpleMessageListener(SimpleMessageListenerContainer container, Session session) + { + this.container = container; + this.session = session; + } + + public void OnMessage(Message message) + { + bool exposeResource = container.ExposeListenerSession; + if (exposeResource) + { + TransactionSynchronizationManager.BindResource( + container.ConnectionFactory, new LocallyExposedEmsResourceHolder(session)); + } + try + { + container.ExecuteListener(session, message); + } finally + { + if (exposeResource) + { + TransactionSynchronizationManager.UnbindResource(container.ConnectionFactory); + } + } + } + } +}