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

TIBCO EMS Integration - SPRNET-982

Browse Source 
visualsvn didn't see some file...added them now.
This commit is contained in:
markpollack
2008-08-12 23:17:41 +00:00
parent 927f1ed209
commit d7f9a3446a
7 changed files with 2242 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

598
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractMessageListenerContainer.cs Normal file
View File

@@ -0,0 +1,598 @@
#region License
/*
* Copyright <20> 2002-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#endregion
using System;
using Common.Logging;
using Spring.Messaging.Ems.Core;
using Spring.Messaging.Ems.Support;
using Spring.Util;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Listener
{
/// <summary>
/// Abstract base class for message listener containers. Can either host
/// a standard EMS MessageListener or a Spring-specific
/// <see cref="ISessionAwareMessageListener"/>
/// </summary>
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
/// <summary>
/// Gets or sets the destination to receive messages from. Will be <code>null</code>
/// if the configured destination is not an actual Destination type;
/// c.f. <see cref="DestinationName"/> when the destination is a String.
/// </summary>
/// <value>The destination.</value>
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;
}
}
}
/// <summary>
/// Gets or sets the name of the destination to receive messages from.
/// Will be <code>null</code> if the configured destination is not a
/// string type; c.f. <see cref="Destination"/> when it is an actual Destination object.
/// </summary>
/// <value>The name of the destination.</value>
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;
}
}
/// <summary>
/// Gets or sets the message selector.
/// </summary>
/// <value>The message selector expression (or <code>null</code> if none)..</value>
public string MessageSelector
{
get { return messageSelector; }
set { messageSelector = value; }
}
/// <summary>
/// Gets or sets the message listener to register.
/// </summary>
///
/// <remarks>
/// <para>
/// This can be either a standard EMS MessageListener object or a
/// Spring <see cref="ISessionAwareMessageListener"/> object.
/// </para>
/// </remarks>
/// <value>The message listener.</value>
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;
}
}
/// <summary>
/// Gets or sets a value indicating whether the subscription is durable.
/// </summary>
/// <remarks>
/// Set whether to make the subscription durable. The durable subscription name
/// to be used can be specified through the "DurableSubscriptionName" property.
/// <para>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).
/// </para>
/// <para>Only makes sense when listening to a topic (pub-sub domain).</para>
/// </remarks>
/// <value><c>true</c> if the subscription is durable; otherwise, <c>false</c>.</value>
public bool SubscriptionDurable
{
get { return subscriptionDurable; }
set { subscriptionDurable = value; }
}
/// <summary>
/// Gets or sets the name of the durable subscription to create.
/// </summary>
/// <remarks>
/// 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.
/// <para>Note: Only 1 concurrent consumer (which is the default of this
/// message listener container) is allowed for each durable subscription.
/// </para>
/// </remarks>
/// <value>The name of the durable subscription.</value>
public string DurableSubscriptionName
{
get
{
return durableSubscriptionName;
}
set
{
AssertUtils.ArgumentNotNull(value, "durableSubscriptionName must not be null");
durableSubscriptionName = value;
}
}
/// <summary>
/// Gets or sets the exception listener to notify in case of a EMSException thrown
/// by the registered message listener or the invocation infrastructure.
/// </summary>
/// <value>The exception listener.</value>
public IExceptionListener ExceptionListener
{
get { return exceptionListener; }
set { exceptionListener = value; }
}
/// <summary>
/// Gets or sets a value indicating whether to expose listener session to a registered
/// <see cref="ISessionAwareMessageListener"/> as well as to <see cref="EmsTemplate"/> calls.
/// </summary>
/// <remarks>
/// 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.
/// <para>Note that Sessions managed by an external transaction manager will
/// always get exposed to <see cref="EmsTemplate"/>
/// calls. So in terms of EmsTemplate exposure, this setting only affects
/// locally transacted Sessions.
/// </para>
/// </remarks>
/// <value>
/// <c>true</c> if expose listener session; otherwise, <c>false</c>.
/// </value>
public bool ExposeListenerSession
{
get { return exposeListenerSession; }
set { exposeListenerSession = value; }
}
/// <summary>
/// Gets or sets a value indicating whether to accept messages while
/// the listener container is in the process of stopping.
/// </summary>
/// <remarks>
/// <para>
/// Return whether to accept received messages while the listener container
/// receive attempt. Switch this flag on to fully process such messages
/// even in the stopping phase, with the drawback that even newly sent
/// messages might still get processed (if coming in before all receive
/// timeouts have expired).
/// </para>
/// <para>
/// Aborting receive attempts for such incoming messages
/// might lead to the provider's retry count decreasing for the affected
/// messages. If you have a high number of concurrent consumers, make sure
/// that the number of retries is higher than the number of consumers,
/// to be on the safe side for all potential stopping scenarios.
/// </para>
/// </remarks>
/// <value>
/// <c>true</c> if accept messages while in the process of stopping; otherwise, <c>false</c>.
/// </value>
public bool AcceptMessagesWhileStopping
{
get { return acceptMessagesWhileStopping; }
set { acceptMessagesWhileStopping = value; }
}
#endregion
/// <summary>
/// Validate that the destination is not null and that if the subscription is durable, then we are not
/// using the Pub/Sub domain.
/// </summary>
protected override void ValidateConfiguration()
{
if (this.destination == null)
{
throw new ArgumentException("Property 'destination' or 'DestinationName' is required");
}
if (SubscriptionDurable && !PubSubDomain)
{
throw new ArgumentException("A durable subscription requires a topic (pub-sub domain)");
}
}
#region Template methods for listeners
/// <summary>
/// Executes the specified listener,
/// committing or rolling back the transaction afterwards (if necessary).
/// </summary>
/// <param name="session">The session to operate on.</param>
/// <param name="message">The received message.</param>
/// <see cref="InvokeListener"/>
/// <see cref="CommitIfNecessary"/>
/// <see cref="RollbackOnExceptionIfNecessary"/>
/// <see cref="HandleListenerException"/>
public virtual void ExecuteListener(Session session, Message message)
{
try
{
DoExecuteListener(session, message);
}
catch (Exception ex)
{
HandleListenerException(ex);
}
}
/// <summary>
/// Executes the specified listener,
/// committing or rolling back the transaction afterwards (if necessary).
/// </summary>
/// <param name="session">The session to operate on.</param>
/// <param name="message">The received message.</param>
/// <exception cref="EMSException">If thrown by EMS API methods.</exception>
/// <see cref="InvokeListener"/>
/// <see cref="CommitIfNecessary"/>
/// <see cref="RollbackOnExceptionIfNecessary"/>
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);
}
/// <summary>
/// Invokes the specified listener: either as standard EMS MessageListener
/// or (preferably) as Spring ISessionAwareMessageListener.
/// </summary>
/// <param name="session">The session to operate on.</param>
/// <param name="message">The received message.</param>
/// <exception cref="EMSException">If thrown by EMS API methods.</exception>
/// <see cref="MessageListener"/>
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");
}
}
/// <summary>
/// Invoke the specified listener as Spring ISessionAwareMessageListener,
/// exposing a new EMS Session (potentially with its own transaction)
/// to the listener if demanded.
/// </summary>
/// <param name="listener">The Spring ISessionAwareMessageListener to invoke.</param>
/// <param name="session">The session to operate on.</param>
/// <param name="message">The received message.</param>
/// <exception cref="EMSException">If thrown by EMS API methods.</exception>
/// <see cref="ISessionAwareMessageListener"/>
/// <see cref="ExposeListenerSession"/>
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);
}
}
/// <summary>
/// Invoke the specified listener as standard JMS MessageListener.
/// </summary>
/// <remarks>Default implementation performs a plain invocation of the
/// <code>OnMessage</code> methods</remarks>
/// <param name="listener">The listener to invoke.</param>
/// <param name="message">The received message.</param>
/// <exception cref="EMSException">if thrown by the EMS API methods</exception>
protected virtual void DoInvokeListener(IMessageListener listener, Message message)
{
listener.OnMessage(message);
}
/// <summary>
/// Perform a commit or message acknowledgement, as appropriate
/// </summary>
/// <param name="session">The session to commit.</param>
/// <param name="message">The message to acknowledge.</param>
/// <exception cref="EMSException">In case of commit failure</exception>
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();
}
}
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
/// <param name="session">The session to check.</param>
/// <returns>
/// <c>true</c> if the is session locally transacted; otherwise, <c>false</c>.
/// </returns>
/// <see cref="EmsAccessor.SessionTransacted"/>
protected virtual bool IsSessionLocallyTransacted(Session session)
{
return SessionTransacted;
}
/// <summary>
/// Perform a rollback, if appropriate.
/// </summary>
/// <param name="session">The session to rollback.</param>
/// <exception cref="EMSException">In case of a rollback error</exception>
protected virtual void RollbackIfNecessary(Session session)
{
if (session.Transacted && IsSessionLocallyTransacted(session))
{
// Transacted session created by this container -> rollback
EmsUtils.RollbackIfNecessary(session);
}
}
/// <summary>
/// Perform a rollback, handling rollback excepitons properly.
/// </summary>
/// <param name="session">The session to rollback.</param>
/// <param name="ex">The thrown application exception.</param>
/// <exception cref="EMSException">in case of a rollback error.</exception>
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;
}
}
/// <summary>
/// Handle the given exception that arose during listener execution.
/// </summary>
/// <remarks>
/// The default implementation logs the exception at error level,
/// not propagating it to the JMS provider - assuming that all handling of
/// acknowledgement and/or transactions is done by this listener container.
/// This can be overridden in subclasses.
/// </remarks>
/// <param name="ex">The exceptin to handle</param>
protected virtual void HandleListenerException(Exception ex)
{
if (ex is MessageRejectedWhileStoppingException)
{
// Internal exception - has been handled before.
return;
}
if (ex is 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);
}
}
/// <summary>
/// Invokes the registered exception listener, if any.
/// </summary>
/// <param name="ex">The exception that arose during EMS processing.</param>
/// <see cref="ExceptionListener"/>
protected virtual void InvokeExceptionListener(EMSException ex)
{
IExceptionListener exListener = ExceptionListener;
if (exListener != null)
{
exListener.OnException(ex);
}
}
#endregion
/// <summary>
/// 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 <see cref="ISessionAwareMessageListener"/> object will be accepted.
/// </summary>
/// <param name="messageListener">The message listener.</param>
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 + "]");
}
}
}
/// <summary>
/// Internal exception class that indicates a rejected message on shutdown.
/// Used to trigger a rollback for an external transaction manager in that case.
/// </summary>
internal class MessageRejectedWhileStoppingException : ApplicationException
{
}
}

546
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractemsListeningContainer.cs Normal file
View File

@@ -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
{
/// <summary>
/// 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
/// <see cref="EmsAccessor"/> base class.
/// </summary>
/// <para>
/// This class provides basic lifecycle management, in particular management
/// of a shared Connection. Subclasses are supposed to plug into this
/// lifecycle, implementing the <see cref="SharedConnectionEnabled"/> as well as
///
/// </para>
/// <remarks>
///
/// </remarks>
/// <author>Mark Pollack</author>
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;
/// <summary>
/// The monitor object to lock on when performing operations on the connection.
/// </summary>
protected object sharedConnectionMonitor = new object();
private volatile bool active = false;
private bool running = false;
/// <summary>
/// The monitor object to lock on when performing operations that update the lifecycle of the container.
/// </summary>
protected object lifecycleMonitor = new object();
#endregion
/// <summary>
/// Gets or sets the client id for a shared Connection created and used by this container.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
/// <value>The client id.</value>
public string ClientId
{
set { clientId = value; }
get { return clientId; }
}
/// <summary> Set whether to automatically start the listener after initialization.
/// <p>Default is "true"; set this to "false" to allow for manual startup.</p>
/// </summary>
public virtual bool AutoStartup
{
set { this.autoStartup = value; }
}
/// <summary>
/// Set the name of the object in the object factory that created this object.
/// </summary>
/// <value>The name of the object in the factory.</value>
/// <remarks>
/// <p>
/// Invoked after population of normal object properties but before an init
/// callback like <see cref="Spring.Objects.Factory.IInitializingObject"/>'s
/// <see cref="Spring.Objects.Factory.IInitializingObject.AfterPropertiesSet"/>
/// method or a custom init-method.
/// </p>
/// </remarks>
public string ObjectName
{
set { objectName = value; }
}
/// <summary>
/// Gets a value indicating whether this container is currently running,
/// that is, whether it has been started and not stopped yet.
/// </summary>
/// <value>
/// <c>true</c> if this container is running; otherwise, <c>false</c>.
/// </value>
public bool IsRunning
{
get
{
lock (lifecycleMonitor)
{
return (running && RunningAllowed);
}
}
}
/// <summary>
/// Gets a value indicating whether this container's listeners are generally allowed to run.
/// </summary>
/// <remarks>
/// <para>
/// >This implementation always returns <code>true</code>; the default 'running'
/// state is purely determined by <see cref="Start"/>/<see cref="Stop"/>.
/// </para>
/// <para>
/// Subclasses may override this method to check against temporary
/// conditions that prevent listeners from actually running. In other words,
/// they may apply further restrictions to the 'running' state, returning
/// <code>false</code> if such a restriction prevents listeners from running.
/// </para>
/// </remarks>
/// <value><c>true</c> if running allowed; otherwise, <c>false</c>.</value>
protected virtual bool RunningAllowed
{
get {
return true; }
}
/// <summary>
/// Gets a value indicating whether this container is currently active,
/// that is, whether it has been set up but not shut down yet.
/// </summary>
/// <value><c>true</c> if active; otherwise, <c>false</c>.</value>
public virtual bool Active
{
get
{
lock (this.lifecycleMonitor)
{
return this.active;
}
}
}
/// <summary> Return whether a shared EMS Connection should be maintained
/// by this listener container base class.
/// </summary>
/// <seealso cref="SharedConnection"/>
protected abstract bool SharedConnectionEnabled { get; }
/// <summary>
/// Gets the shared connection maintained by this container.
/// Available after initialization.
/// </summary>
/// <value>The shared connection (never null)</value>
/// <exception cref="InvalidOperationException">if this container does not maintain a
/// shared Connection, or if the Connection hasn't been initialized yet.
/// </exception>
/// <see cref="SharedConnectionEnabled"/>
protected 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;
}
}
}
/// <summary>
/// Call base class method, then <see cref="ValidateConfiguration"/> and then <see cref="Initialize"/>
/// </summary>
public override void AfterPropertiesSet()
{
base.AfterPropertiesSet();
ValidateConfiguration();
Initialize();
}
/// <summary>
/// Validates the configuration of this container. The default implementation
/// is empty. To be overriden in subclasses.
/// </summary>
protected virtual void ValidateConfiguration()
{
}
/// <summary>
/// Calls <see cref="Shutdown"/> when the application context destroys the container instance.
/// </summary>
public void Dispose()
{
Shutdown();
}
/// <summary>
/// Initializes this container. Creates a Connection, starts the Connection
/// (if the property <see cref="AutoStartup"/> hasn't been turned off), and calls
/// <see cref="DoInitialize"/>.
/// </summary>
/// <exception cref="EMSException">If startup failed</exception>
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;
}
}
/// <summary>
/// Stop the shared connection, call <see cref="DoShutdown"/>, and close this container.
/// </summary>
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);
}
}
}
/// <summary>
/// Starts this container.
/// </summary>
/// <exception cref="EMSException">if starting failed.</exception>
public void Start()
{
DoStart();
}
/// <summary>
/// Start the shared Connection, if any, and notify all invoker tasks.
/// </summary>
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();
}
}
/// <summary>
/// Stops this container.
/// </summary>
/// <exception cref="EMSException">if stopping failed.</exception>
public void Stop()
{
DoStop();
}
/// <summary>
/// Notify all invoker tasks and stop the shared Connection, if any.
/// </summary>
/// <exception cref="EMSException">if thrown by EMS API methods.</exception>
/// <see cref="StopSharedConnection"/>
protected virtual void DoStop()
{
lock (this.lifecycleMonitor)
{
this.running = false;
System.Threading.Monitor.PulseAll(this.lifecycleMonitor);
}
if (SharedConnectionEnabled)
{
StopSharedConnection();
}
}
/// <summary>
/// Register any invokers within this container.
/// Subclasses need to implement this method for their specific
/// invoker management process. A shared Connection, if any, will already have been
/// started at this point.
/// </summary>
protected abstract void DoInitialize();
/// <summary>
/// Close the registered invokers. Subclasses need to implement this method
/// for their specific invoker management process. A shared Connection, if any,
/// will automatically be closed afterwards.
/// </summary>
protected abstract void DoShutdown();
/// <summary>
/// Establishes a shared Connection for this container.
/// </summary>
/// <remarks>
/// <para>
/// The default implementation delegates to <see cref="CreateSharedConnection"/>
/// which does one immediate attempt and throws an exception if it fails.
/// Can be overridden to have a recovery process in place, retrying
/// until a Connection can be successfully established.
/// </para>
/// </remarks>
/// <exception cref="EMSException">If thrown by EMS API methods</exception>
protected virtual void EstablishSharedConnection()
{
lock (sharedConnectionMonitor)
{
if (sharedConnection == null)
{
sharedConnection = CreateSharedConnection();
logger.Debug("Established shared EMS Connection");
}
}
}
/// <summary>
/// Refreshes the shared connection that this container holds.
/// </summary>
/// <remarks>
/// Called on startup and also after an infrastructure exception
/// that occurred during invoker setup and/or execution.
/// </remarks>
/// <exception cref="EMSException">If thrown by EMS API methods</exception>
protected void RefreshSharedConnection()
{
lock (sharedConnectionMonitor)
{
ConnectionFactoryUtils.ReleaseConnection(sharedConnection, ConnectionFactory, sharedConnectionStarted);
sharedConnection = CreateSharedConnection();
if (sharedConnectionStarted)
{
sharedConnection.Start();
}
}
}
/// <summary>
/// Creates the shared connection for this container.
/// </summary>
/// <remarks>
/// The default implementation creates a standard Connection
/// and prepares it through <see cref="PrepareSharedConnection"/>
/// </remarks>
/// <returns>the prepared Connection</returns>
/// <exception cref="EMSException">if the creation failed.</exception>
protected virtual Connection CreateSharedConnection()
{
Connection con = CreateConnection();
try
{
PrepareSharedConnection(con);
return con;
} catch (EMSException)
{
EmsUtils.CloseConnection(con);
throw;
}
}
/// <summary>
/// Prepares the given connection, which is about to be registered
/// as shared Connection for this container.
/// </summary>
/// <remarks>
/// The default implementation sets the specified client id, if any.
/// Subclasses can override this to apply further settings.
/// </remarks>
/// <param name="connection">The connection to prepare.</param>
/// <exception cref="EMSException">If the preparation efforts failed.</exception>
protected virtual void PrepareSharedConnection(Connection connection)
{
if (ClientId != null)
{
connection.ClientID = ClientId;
}
}
/// <summary>
/// Starts the shared connection.
/// </summary>
/// <exception cref="EMSException">If thrown by EMS API methods</exception>
/// <see cref="Start"/>
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);
}
}
}
}
/// <summary>
/// Stops the shared connection.
/// </summary>
/// <exception cref="EMSException">if thrown by EMS API methods.</exception>
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);
}
}
}
}
}
/// <summary>
/// Exception that indicates that the initial setup of this container's
/// shared Connection failed. This is indicating to invokers that they need
/// to establish the shared Connection themselves on first access.
/// </summary>
public class SharedConnectionNotInitializedException : EMSException
{
/// <summary>
/// Initializes a new instance of the <see cref="SharedConnectionNotInitializedException"/> class.
/// </summary>
/// <param name="message">The message.</param>
public SharedConnectionNotInitializedException(string message) : base(message)
{
}
}
}

54
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/ListenerExecutionFailedException.cs Normal file
View File

@@ -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
{
/// <summary>
/// Exception to be thrown when the execution of a listener method failed.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class ListenerExecutionFailedException : EMSException
{
/// <summary>
/// Initializes a new instance of the <see cref="ListenerExecutionFailedException"/> class, with the specified message
/// </summary>
/// <param name="message">The message.</param>
public ListenerExecutionFailedException(string message) : base(message)
{
}
/// <summary>
/// Initializes a new instance of the <see cref="ListenerExecutionFailedException"/> class, with the specified message
/// and root cause exception
/// </summary>
/// <param name="message">The message.</param>
/// <param name="innerException">The inner exception.</param>
public ListenerExecutionFailedException(string message, Exception innerException)
: base(message)
{
LinkedException = innerException;
}
}
}

552
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/MessageListenerAdapter.cs Normal file
View File

@@ -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
{
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// <para>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
/// <see cref="IMessageConverter"/>. By default, a <see cref="SimpleMessageConverter"/>
/// will be used. (If you do not want such automatic message conversion taking
/// place, then be sure to set the <see cref="MessageConverter"/> property
/// to <code>null</code>.)
/// </para>
/// <para>If a target listener method returns a non-null object (typically of a
/// message content type such as <code>String</code> or byte array), it will get
/// wrapped in a EMS <code>Message</code> and sent to the response destination
/// (either the EMS "reply-to" destination or the <see cref="defaultResponseDestination"/>
/// specified.
/// </para>
/// <para>
/// The sending of response messages is only available when
/// using the <see cref="ISessionAwareMessageListener"/> entry point (typically through a
/// Spring message listener container). Usage as standard EMS MessageListener
/// does <i>not</i> support the generation of response messages.
/// </para>
/// <para>Consult the reference documentation for examples of method signatures compliant with this
/// adapter class.
/// </para>
/// </remarks>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class MessageListenerAdapter : IMessageListener, ISessionAwareMessageListener
{
#region Logging
private readonly ILog logger = LogManager.GetLogger(typeof (MessageListenerAdapter));
#endregion
/// <summary>
/// The default handler method name.
/// </summary>
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;
/// <summary>
/// Initializes a new instance of the <see cref="MessageListenerAdapter"/> class with default settings.
/// </summary>
public MessageListenerAdapter()
{
InitDefaultStrategies();
handlerObject = this;
processingExpression = Expression.Parse(defaultHandlerMethod + "(#convertedObject)");
}
/// <summary>
/// Initializes a new instance of the <see cref="MessageListenerAdapter"/> class for the given handler object
/// </summary>
/// <param name="handlerObject">The delegate object.</param>
public MessageListenerAdapter(object handlerObject)
{
InitDefaultStrategies();
this.handlerObject = handlerObject;
}
/// <summary>
/// Gets or sets the handler object to delegate message listening to.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
/// <value>The handler object.</value>
public object HandlerObject
{
get { return handlerObject; }
set { handlerObject = value; }
}
/// <summary>
/// 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 <see cref="ORIGINAL_DEFAULT_HANDLER_METHOD"/> ("HandleMessage"}.
/// </summary>
/// <value>The default handler method.</value>
public string DefaultHandlerMethod
{
get { return defaultHandlerMethod; }
set
{
defaultHandlerMethod = value;
}
}
/// <summary>
/// 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.
/// <para>
/// Alternatively, specify a "DefaultResponseQueueName" or "DefaultResponseTopicName",
/// to be dynamically resolved via the DestinationResolver.
/// </para>
/// </summary>
/// <value>The default response destination.</value>
public object DefaultResponseDestination
{
set { defaultResponseDestination = value; }
}
/// <summary>
/// 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.
/// <para>Alternatively, specify a JMS Destination object as "defaultResponseDestination".</para>
/// </summary>
/// <value>The name of the default response destination queue.</value>
public string DefaultResponseQueueName
{
set { defaultResponseDestination = new DestinationNameHolder(value, false); }
}
/// <summary>
/// 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.
/// <para>Alternatively, specify a JMS Destination object as "defaultResponseDestination".</para>
/// </summary>
/// <value>The name of the default response destination topic.</value>
public string DefaultResponseTopicName
{
set { defaultResponseDestination = new DestinationNameHolder(value, true); }
}
/// <summary>
/// Gets or sets the destination resolver that should be used to resolve response
/// destination names for this adapter.
/// <para>The default resolver is a <see cref="DynamicDestinationResolver"/>.
/// Specify another implementation, for other strategies, perhaps from a directory service.</para>
/// </summary>
/// <value>The destination resolver.</value>
public DestinationResolver DestinationResolver
{
get { return destinationResolver; }
set
{
AssertUtils.ArgumentNotNull(value, "DestinationResolver must not be null");
destinationResolver = value;
}
}
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// <para>The default converter is a {@link SimpleMessageConverter}, which is able
/// to handle BytesMessages}, TextMessages, MapMessages, and ObjectMessages.
/// </para>
/// </remarks>
/// <value>The message converter.</value>
public IMessageConverter MessageConverter
{
get { return messageConverter; }
set { messageConverter = value; }
}
/// <summary>
/// Standard JMS {@link MessageListener} entry point.
/// <para>Delegates the message to the target listener method, with appropriate
/// conversion of the message arguments
/// </para>
/// </summary>
/// <remarks>
/// In case of an exception, the <see cref="HandleListenerException"/> method will be invoked.
/// <b>Note</b>
/// Does not support sending response messages based on
/// result objects returned from listener methods. Use the
/// <see cref="ISessionAwareMessageListener"/> entry point (typically through a Spring
/// message listener container) for handling result objects as well.
/// </remarks>
/// <param name="message">The incoming message.</param>
public void OnMessage(Message message)
{
try
{
OnMessage(message, null);
}
catch (Exception e)
{
HandleListenerException(e);
}
}
/// <summary>
/// Spring <see cref="ISessionAwareMessageListener"/> entry point.
/// <para>
/// 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.
/// </para>
/// </summary>
/// <param name="message">The incoming message.</param>
/// <param name="session">The session to operate on.</param>
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");
}
}
/// <summary>
/// Initialize the default implementations for the adapter's strategies.
/// </summary>
protected virtual void InitDefaultStrategies()
{
MessageConverter = new SimpleMessageConverter();
}
/// <summary>
/// Handle the given exception that arose during listener execution.
/// The default implementation logs the exception at error level.
/// <para>This method only applies when used as standard EMS MessageListener.
/// In case of the Spring <see cref="ISessionAwareMessageListener"/> mechanism,
/// exceptions get handled by the caller instead.
/// </para>
/// </summary>
/// <param name="ex">The exception to handle.</param>
protected virtual void HandleListenerException(Exception ex)
{
logger.Error("Listener execution failed", ex);
}
/// <summary>
/// Extract the message body from the given message.
/// </summary>
/// <param name="message">The message.</param>
/// <returns>the content of the message, to be passed into the
/// listener method as argument</returns>
/// <exception cref="EMSException">if thrown by EMS API methods</exception>
private object ExtractMessage(Message message)
{
IMessageConverter converter = MessageConverter;
if (converter != null)
{
return converter.FromMessage(message);
}
return message;
}
/// <summary>
/// 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.
/// </summary>
/// <param name="originalMessage">The EMS request message.</param>
/// <param name="extractedMessage">The converted JMS request message,
/// to be passed into the listener method as argument.</param>
/// <returns>the name of the listener method (never <code>null</code>)</returns>
/// <exception cref="EMSException">if thrown by EMS API methods</exception>
protected virtual string GetHandlerMethodName(Message originalMessage, object extractedMessage)
{
return DefaultHandlerMethod;
}
/// <summary>
/// Handles the given result object returned from the listener method, sending a response message back.
/// </summary>
/// <param name="result">The result object to handle (never <code>null</code>).</param>
/// <param name="request">The original request message.</param>
/// <param name="session">The session to operate on (may be <code>null</code>).</param>
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");
}
}
}
/// <summary>
/// Builds a JMS message to be sent as response based on the given result object.
/// </summary>
/// <param name="session">The JMS Session to operate on.</param>
/// <param name="result">The content of the message, as returned from the listener method.</param>
/// <returns>the JMS <code>Message</code> (never <code>null</code>)</returns>
/// <exception cref="MessageConversionException">If there was an error in message conversion</exception>
/// <exception cref="EMSException">if thrown by EMS API methods</exception>
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;
}
}
/// <summary>
/// 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.
/// </summary>
/// <param name="request">The original incoming message.</param>
/// <param name="response">The outgoing JMS message about to be sent.</param>
/// <exception cref="EMSException">if thrown by EMS API methods</exception>
protected virtual void PostProcessResponse(Message request, Message response)
{
response.CorrelationID = request.CorrelationID;
}
/// <summary>
/// Determine a response destination for the given message.
/// </summary>
/// <remarks>
/// <para>The default implementation first checks the JMS Reply-To
/// Destination of the supplied request; if that is not <code>null</code>
/// it is returned; if it is <code>null</code>, then the configured
/// <see cref="ResolveDefaultResponseDestination"/> default response destination}
/// is returned; if this too is <code>null</code>, then an
/// <see cref="InvalidDestinationException"/>is thrown.
/// </para>
/// </remarks>
/// <param name="request">The original incoming message.</param>
/// <param name="response">Tthe outgoing message about to be sent.</param>
/// <param name="session">The session to operate on.</param>
/// <returns>the response destination (never <code>null</code>)</returns>
/// <exception cref="EMSException">if thrown by EMS API methods</exception>
/// <exception cref="InvalidDestinationException">if no destination can be determined.</exception>
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;
}
/// <summary>
/// Resolves the default response destination into a Destination, using this
/// accessor's <see cref="DestinationResolver"/> in case of a destination name.
/// </summary>
/// <param name="session">The session to operate on.</param>
/// <returns>The located destination</returns>
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;
}
/// <summary>
/// Sends the given response message to the given destination.
/// </summary>
/// <param name="session">The session to operate on.</param>
/// <param name="destination">The destination to send to.</param>
/// <param name="response">The outgoing message about to be sent.</param>
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);
}
}
/// <summary>
/// Post-process the given message producer before using it to send the response.
/// The default implementation is empty.
/// </summary>
/// <param name="producer">The producer that will be used to send the message.</param>
/// <param name="response">The outgoing message about to be sent.</param>
protected virtual void PostProcessProducer(MessageProducer producer, Message response)
{
}
}
/// <summary>
/// Internal class combining a destination name and its target destination type (queue or topic).
/// </summary>
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; }
}
}
}

51
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/ISessionAwareMessageListener.cs Normal file
View File

@@ -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
{
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// Supported by Spring's <see cref="SimpleMessageListenerContainer"/>
/// as direct alternative to the standard MessageListener interface.
/// </remarks>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public interface ISessionAwareMessageListener
{
/// <summary> Callback for processing a received EMS message.
/// Implementors are supposed to process the given Message,
/// typically sending reply messages through the given Session.
/// </summary>
/// <param name="message">the received EMS message
/// </param>
/// <param name="session">the underlying EMS Session
/// </param>
/// <throws> EMSException if thrown by EMS methods </throws>
void OnMessage(Message message, Session session);
}
}

43
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/LocallyExposedEmsResourceHolder.cs Normal file
View File

@@ -0,0 +1,43 @@
#region License
/*
* Copyright 2002-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#endregion
using Spring.Messaging.Ems.Connections;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Listener
{
/// <summary>
/// EmsResourceHolder marker subclass that indicates local exposure,
/// i.e. that does not indicate an externally managed transaction.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class LocallyExposedEmsResourceHolder : EmsResourceHolder
{
/// <summary>
/// Initializes a new instance of the <see cref="LocallyExposedEmsResourceHolder"/> class.
/// </summary>
/// <param name="session">The session.</param>
public LocallyExposedEmsResourceHolder(Session session) : base(session)
{
}
}
}

398
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/SimpleMessageListenerContainer.cs Normal file
View File

@@ -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
{
/// <summary>
/// Message listener container that uses the plain EMS client API's
/// MessageConsumer.Listener method to create concurrent
/// MessageConsumers for the specified listeners.
/// </summary>
public class SimpleMessageListenerContainer : AbstractMessageListenerContainer, IExceptionListener
{
#region Logging
private readonly ILog logger = LogManager.GetLogger(typeof(SimpleMessageListenerContainer));
#endregion
#region fields
/// <summary>
/// The default recovery time interval between connection reconnection attempts
/// </summary>
public static TimeSpan DEFAULT_RECOVERY_INTERVAL = new TimeSpan(0, 0, 0, 5, 0);
/// <summary>
/// The total time connection recovery will be attempted.
/// </summary>
public static TimeSpan DEFAULT_MAX_RECOVERY_TIME = new TimeSpan(0, 0, 10, 0, 0);
private bool pubSubNoLocal = false;
private int concurrentConsumers = 1;
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
/// <summary>
/// Gets or sets a value indicating whether to inhibit the delivery of messages published by its own connection.
/// Default is "false".
/// </summary>
/// <value><c>true</c> if should inhibit the delivery of messages published by its own connection; otherwise, <c>false</c>.</value>
public bool PubSubNoLocal
{
get { return pubSubNoLocal; }
set { pubSubNoLocal = value; }
}
/// <summary>
/// Specify the number of concurrent consumers to create. Default is 1.
/// </summary>
/// <remarks>
/// 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.
/// <para>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.
/// </para>
/// </remarks>
/// <value>The concurrent consumers.</value>
public int ConcurrentConsumers
{
set
{
AssertUtils.IsTrue(value > 0, "'ConcurrentConsumer' value must be at least 1 (one)");
concurrentConsumers = value;
}
}
/// <summary>
/// Sets the time interval between connection recovery attempts. The default is 5 seconds.
/// </summary>
/// <value>The recovery interval.</value>
public TimeSpan RecoveryInterval
{
set { recoveryInterval = value; }
}
/// <summary>
/// Sets the max recovery time to try reconnection attempts. The default is 10 minutes.
/// </summary>
/// <value>The max recovery time.</value>
public TimeSpan MaxRecoveryTime
{
set { maxRecoveryTime = value; }
}
/// <summary>
/// Always use a shared EMS connection
/// </summary>
protected override bool SharedConnectionEnabled
{
get { return true; }
}
#endregion
/// <summary>
/// Call base class for valdation and then check that if the subscription is durable that the number of
/// concurrent consumers is equal to one.
/// </summary>
protected override void ValidateConfiguration()
{
base.ValidateConfiguration();
if (SubscriptionDurable && concurrentConsumers !=1 )
{
throw new ArgumentException("Only 1 concurrent consumer supported for durable subscription");
}
}
/// <summary>
/// Creates the specified number of concurrent consumers,
/// in the form of a JMS Session plus associated MessageConsumer
/// </summary>
/// <see cref="CreateListenerConsumer"/>
protected override void DoInitialize()
{
EstablishSharedConnection();
InitializeConsumers();
}
/// <summary>
/// Re-initializes this container's EMS message consumers,
/// if not initialized already.
/// </summary>
protected override void DoStart()
{
base.DoStart();
InitializeConsumers();
}
/// <summary>
/// Registers this listener container as EMS ExceptionListener on the shared connection.
/// </summary>
/// <param name="connection"></param>
protected override void PrepareSharedConnection(Connection connection)
{
base.PrepareSharedConnection(connection);
connection.ExceptionListener = this;
}
/// <summary>
/// <see cref="IExceptionListener"/> implementation, invoked by the EMS provider in
/// case of connection failures. Re-initializes this listener container's
/// shared connection and its sessions and consumers.
/// </summary>
/// <param name="exception">The reported connection exception.</param>
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);
}
}
/// <summary>
/// Refresh the underlying Connection, not returning before an attempt has been
/// successful. Called in case of a shared Connection as well as without shared
/// Connection, so either needs to operate on the shared Connection or on a
/// temporary Connection that just gets established for validation purposes.
/// </summary>
/// <remarks>
/// The default implementation retries until it successfully established a
/// Connection, for as long as this message listener container is active.
/// Applies the specified recovery interval between retries.
/// </remarks>
protected virtual void RefreshConnectionUntilSuccessful()
{
TimeSpan totalTryTime = new TimeSpan();
while (IsRunning)
{
try
{
RefreshSharedConnection();
break;
}
catch (Exception ex)
{
if (logger.IsInfoEnabled)
{
logger.Info("Could not refresh Connection - retrying in " + recoveryInterval, ex);
}
}
if (totalTryTime > maxRecoveryTime)
{
logger.Info("Could not refresh Connection after " + totalTryTime + ". Stopping reconnection attempts.");
throw new RecoveryTimeExceededException("Could not recover after " + totalTryTime);
}
DateTime startTime = DateTime.Now;
SleepInBetweenRecoveryAttempts();
TimeSpan sleepTimeSpan = DateTime.Now - startTime;
totalTryTime += sleepTimeSpan;
}
}
/// <summary>
/// The amount of time to sleep in between recovery attempts.
/// </summary>
protected virtual void SleepInBetweenRecoveryAttempts()
{
Thread.Sleep(recoveryInterval);
}
/// <summary>
/// Initialize the Sessions and MessageConsumers for this container.
/// </summary>
/// <exception cref="EMSException">in case of setup failure.</exception>
protected virtual void InitializeConsumers()
{
// Register Sessions and MessageConsumers
lock (consumersMonitor)
{
if (this.consumers == null)
{
logger.Debug("InitializingConsumers **********");
this.sessions = new HashedSet();
this.consumers = new HashedSet();
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);
}
}
}
}
/// <summary>
/// Creates a MessageConsumer for the given Session,
/// registering a MessageListener for the specified listener
/// </summary>
/// <param name="session">The session to work on.</param>
/// <returns>the MessageConsumer"/></returns>
/// <exception cref="EMSException">if thrown by EMS methods</exception>
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;
}
/// <summary>
/// Close the message consumers and sessions.
/// </summary>
/// <throws>EMSException if destruction failed </throws>
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;
}
/// <summary>
/// Creates a MessageConsumer for the given Session and Destination.
/// </summary>
/// <param name="session">The session to create a MessageConsumer for.</param>
/// <param name="destination">The destination to create a MessageConsumer for.</param>
/// <returns>The new MessageConsumer</returns>
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);
}
}
}
}
}