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

TIBCO EMS Integration - SPRNET-982

Browse Source
This commit is contained in:
markpollack
2008-08-12 22:24:08 +00:00
parent c321c0da5c
commit 927f1ed209
13 changed files with 2249 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ChainedExceptionListener.cs Normal file
View File

@@ -0,0 +1,71 @@
#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.Collections;
using Spring.Util;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Connections
{
/// <summary>
/// Implementation of Spring IExceptionListener interface that supports
/// chaining allowing the addition of multiple ExceptionListener instances in order.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class ChainedExceptionListener : IExceptionListener
{
private ArrayList listeners = new ArrayList(2);
/// <summary>
/// Adds the exception listener to the chain
/// </summary>
/// <param name="listener">The listener.</param>
public void AddListener(IExceptionListener listener)
{
AssertUtils.ArgumentNotNull(listener, "listener", "ExceptionListener must not be null");
listeners.Add(listener);
}
/// <summary>
/// Called when an exception occurs in message processing.
/// </summary>
/// <param name="exception">The exception.</param>
public void OnException(EMSException exception)
{
foreach (IExceptionListener listener in listeners)
{
listener.OnException(exception);
}
}
/// <summary>
/// Gets the exception listeners as an array.
/// </summary>
/// <value>The exception listeners.</value>
public IExceptionListener[] Listeners
{
get
{
return (IExceptionListener[]) listeners.ToArray(typeof (IExceptionListener));
}
}
}
}

378
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ConnectionFactoryUtils.cs Normal file
View File

@@ -0,0 +1,378 @@
#region License
/*
* Copyright <20> 2002-2006 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.Support;
using Spring.Transaction.Support;
using Spring.Util;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Connections
{
/// <summary> Helper class for obtaining transactional EMS resources
/// for a given ConnectionFactory.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public abstract class ConnectionFactoryUtils
{
#region Logging
private static readonly ILog LOG = LogManager.GetLogger(typeof(ConnectionFactoryUtils));
#endregion
/// <summary>
/// Releases the given connection.
/// </summary>
/// <param name="connection">The connection to release. (if this is <code>null</code>, the call will be ignored)</param>
/// <param name="cf">The ConnectionFactory that the Connection was obtained from. (may be <code>null</code>)</param>
/// <param name="started">whether the Connection might have been started by the application.</param>
public static void ReleaseConnection(Connection connection, ConnectionFactory cf, bool started)
{
if (connection == null)
{
return;
}
try
{
connection.Close();
} catch (Exception ex)
{
LOG.Debug("Could not close EMS Connection", ex);
}
}
/// <summary>
/// Determines whether the given JMS Session is transactional, that is,
/// bound to the current thread by Spring's transaction facilities.
/// </summary>
/// <param name="session">The session to check.</param>
/// <param name="cf">The ConnectionFactory that the Session originated from</param>
/// <returns>
/// <c>true</c> if is session transactional, bound to current thread; otherwise, <c>false</c>.
/// </returns>
public static bool IsSessionTransactional(Session session, ConnectionFactory cf)
{
if (session == null || cf == null)
{
return false;
}
EmsResourceHolder resourceHolder = (EmsResourceHolder) TransactionSynchronizationManager.GetResource(cf);
return (resourceHolder != null && resourceHolder.ContainsSession(session));
}
/// <summary> Obtain a EMS Session that is synchronized with the current transaction, if any.</summary>
/// <param name="cf">the ConnectionFactory to obtain a Session for
/// </param>
/// <param name="existingCon">the existing EMS Connection to obtain a Session for
/// (may be <code>null</code>)
/// </param>
/// <param name="synchedLocalTransactionAllowed">whether to allow for a local EMS transaction
/// that is synchronized with a Spring-managed transaction (where the main transaction
/// might be a ADO.NET-based one for a specific DataSource, for example), with the EMS
/// transaction committing right after the main transaction. If not allowed, the given
/// ConnectionFactory needs to handle transaction enlistment underneath the covers.
/// </param>
/// <returns> the transactional Session, or <code>null</code> if none found
/// </returns>
/// <throws> EMSException in case of EMS failure </throws>
public static Session GetTransactionalSession(ConnectionFactory cf, Connection existingCon,
bool synchedLocalTransactionAllowed)
{
return
DoGetTransactionalSession(cf,
new AnonymousClassResourceFactory(existingCon, cf,
synchedLocalTransactionAllowed), true);
}
/// <summary>
/// Obtain a EMS Session that is synchronized with the current transaction, if any.
/// </summary>
/// <param name="resourceKey">the TransactionSynchronizationManager key to bind to
/// (usually the ConnectionFactory)</param>
/// <param name="resourceFactory">the ResourceFactory to use for extracting or creating
/// EMS resources</param>
/// <param name="startConnection">whether the underlying Connection approach should be
/// started in order to allow for receiving messages. Note that a reused Connection
/// may already have been started before, even if this flag is <code>false</code>.</param>
/// <returns>
/// the transactional Session, or <code>null</code> if none found
/// </returns>
/// <throws>EMSException in case of EMS failure </throws>
public static Session DoGetTransactionalSession(Object resourceKey, ResourceFactory resourceFactory, bool startConnection)
{
AssertUtils.ArgumentNotNull(resourceKey, "Resource key must not be null");
AssertUtils.ArgumentNotNull(resourceKey, "ResourceFactory must not be null");
EmsResourceHolder resourceHolder =
(EmsResourceHolder)TransactionSynchronizationManager.GetResource(resourceKey);
if (resourceHolder != null)
{
Session rhSession = resourceFactory.GetSession(resourceHolder);
if (rhSession != null)
{
if (startConnection)
{
Connection conn = resourceFactory.GetConnection(resourceHolder);
if (conn != null)
{
conn.Start();
}
}
return rhSession;
}
}
if (!TransactionSynchronizationManager.SynchronizationActive)
{
return null;
}
EmsResourceHolder resourceHolderToUse = resourceHolder;
if (resourceHolderToUse == null)
{
resourceHolderToUse = new EmsResourceHolder();
}
Connection con = resourceFactory.GetConnection(resourceHolderToUse);
Session session = null;
try
{
bool isExistingCon = (con != null);
if (!isExistingCon)
{
con = resourceFactory.CreateConnection();
resourceHolderToUse.AddConnection(con);
}
session = resourceFactory.CreateSession(con);
resourceHolderToUse.AddSession(session, con);
if (startConnection)
{
con.Start();
}
}
catch (EMSException)
{
if (session != null)
{
try
{
session.Close();
}
catch (Exception)
{
// ignore
}
}
if (con != null)
{
try
{
con.Close();
}
catch (Exception)
{
// ignore
}
}
throw;
}
if (resourceHolderToUse != resourceHolder)
{
TransactionSynchronizationManager.RegisterSynchronization(
new EmsResourceSynchronization(resourceKey, resourceHolderToUse,
resourceFactory.SynchedLocalTransactionAllowed));
resourceHolderToUse.SynchronizedWithTransaction = true;
TransactionSynchronizationManager.BindResource(resourceKey, resourceHolderToUse);
}
return session;
}
#region ResourceFactory helper classes
private class AnonymousClassResourceFactory : ResourceFactory
{
private Connection existingCon;
private ConnectionFactory cf;
private bool synchedLocalTransactionAllowed;
public AnonymousClassResourceFactory(Connection existingCon, ConnectionFactory cf,
bool synchedLocalTransactionAllowed)
{
InitBlock(existingCon, cf, synchedLocalTransactionAllowed);
}
private void InitBlock(Connection existingCon, ConnectionFactory cf, bool synchedLocalTransactionAllowed)
{
this.existingCon = existingCon;
this.cf = cf;
this.synchedLocalTransactionAllowed = synchedLocalTransactionAllowed;
}
public virtual Session GetSession(EmsResourceHolder holder)
{
return holder.GetSession(typeof(Session), existingCon);
}
public virtual Connection GetConnection(EmsResourceHolder holder)
{
return (existingCon != null ? existingCon : holder.GetConnection());
}
public virtual Connection CreateConnection()
{
return cf.CreateConnection();
}
public virtual Session CreateSession(Connection con)
{
return con.CreateSession(synchedLocalTransactionAllowed, Session.SESSION_TRANSACTED);
}
public bool SynchedLocalTransactionAllowed
{
get { return synchedLocalTransactionAllowed; }
}
}
#endregion
#region Helper classes/interfaces
/// <summary> Callback interface for resource creation.
/// Serving as argument for the <code>DoGetTransactionalSession</code> method.
/// </summary>
public interface ResourceFactory
{
/// <summary> Fetch an appropriate Session from the given EmsResourceHolder.</summary>
/// <param name="holder">the EmsResourceHolder
/// </param>
/// <returns> an appropriate Session fetched from the holder,
/// or <code>null</code> if none found
/// </returns>
Session GetSession(EmsResourceHolder holder);
/// <summary> Fetch an appropriate Connection from the given EmsResourceHolder.</summary>
/// <param name="holder">the EmsResourceHolder
/// </param>
/// <returns> an appropriate Connection fetched from the holder,
/// or <code>null</code> if none found
/// </returns>
Connection GetConnection(EmsResourceHolder holder);
/// <summary> Create a new EMS Connection for registration with a EmsResourceHolder.</summary>
/// <returns> the new EMS Connection
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
Connection CreateConnection();
/// <summary> Create a new EMS Session for registration with a EmsResourceHolder.</summary>
/// <param name="con">the EMS Connection to create a Session for
/// </param>
/// <returns> the new EMS Session
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
Session CreateSession(Connection con);
/// <summary>
/// Return whether to allow for a local EMS transaction that is synchronized with
/// a Spring-managed transaction (where the main transaction might be a ADO.NET-based
/// one for a specific IDbProvider, for example), with the EMS transaction
/// committing right after the main transaction.
/// Returns whether to allow for synchronizing a local EMS transaction
/// </summary>
///
bool SynchedLocalTransactionAllowed { get; }
}
/// <summary> Callback for resource cleanup at the end of a non-native EMS transaction
/// </summary>
private class EmsResourceSynchronization : TransactionSynchronizationAdapter
{
private object resourceKey;
private EmsResourceHolder resourceHolder;
private bool transacted;
private bool holderActive = true;
public EmsResourceSynchronization(object resourceKey, EmsResourceHolder resourceHolder, bool transacted)
{
this.resourceKey = resourceKey;
this.resourceHolder = resourceHolder;
this.transacted = transacted;
}
public override void Suspend()
{
if (holderActive)
{
TransactionSynchronizationManager.UnbindResource(resourceKey);
}
}
public override void Resume()
{
if (holderActive)
{
TransactionSynchronizationManager.BindResource(resourceKey, resourceHolder);
}
}
public override void BeforeCompletion()
{
TransactionSynchronizationManager.UnbindResource(resourceKey);
holderActive = false;
if (!transacted)
{
resourceHolder.CloseAll();
}
}
public override void AfterCommit()
{
if (transacted)
{
try
{
resourceHolder.CommitAll();
}
catch (EMSException ex)
{
throw new SynchedLocalTransactionFailedException("Local EMS transaction failed to commit", ex);
}
}
}
public override void AfterCompletion(TransactionSynchronizationStatus status)
{
if (transacted)
{
resourceHolder.CloseAll();
}
}
}
#endregion
}
}

292
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsResourceHolder.cs Normal file
View File

@@ -0,0 +1,292 @@
#region License
/*
* Copyright <20> 2002-2006 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#endregion
using System;
using System.Collections;
using Common.Logging;
using Spring.Collections;
using Spring.Transaction.Support;
using Spring.Util;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Connections
{
/// <summary>Connection holder, wrapping a EMS Connection and a EMS Session.
/// EmsTransactionManager binds instances of this class to the thread,
/// for a given EMS ConnectionFactory.
///
/// <p>Note: This is an SPI class, not intended to be used by applications.</p>
///
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class EmsResourceHolder : ResourceHolderSupport
{
#region Logging
private static readonly ILog logger = LogManager.GetLogger(typeof(EmsResourceHolder));
#endregion
#region Fields
private ConnectionFactory connectionFactory;
private bool frozen = false;
private IList connections = new LinkedList();
private IList sessions = new LinkedList();
private IDictionary sessionsPerConnection = new Hashtable();
#endregion
#region Constructor (s)
/// <summary> Create a new EmsResourceHolder that is open for resources to be added.</summary>
public EmsResourceHolder()
{
}
/// <summary>
/// Initializes a new instance of the <see cref="EmsResourceHolder"/> class
/// at is open for resources to be added.
/// </summary>
/// <param name="connectionFactory">The connection factory that this
/// resource holder is associated with (may be <code>null</code>)
/// </param>
public EmsResourceHolder(ConnectionFactory connectionFactory)
{
this.connectionFactory = connectionFactory;
}
/// <summary>
/// Initializes a new instance of the <see cref="EmsResourceHolder"/> class for the
/// given Session.
/// </summary>
/// <param name="session">The session.</param>
public EmsResourceHolder(Session session)
{
AddSession(session);
frozen = true;
}
/// <summary> Create a new EmsResourceHolder for the given EMS resources.</summary>
/// <param name="connection">the EMS Connection
/// </param>
/// <param name="session">the EMS Session
/// </param>
public EmsResourceHolder(Connection connection, Session session)
{
AddConnection(connection);
AddSession(session, connection);
this.frozen = true;
}
/// <summary>
/// Initializes a new instance of the <see cref="EmsResourceHolder"/> class.
/// </summary>
/// <param name="connectionFactory">The connection factory.</param>
/// <param name="connection">The connection.</param>
/// <param name="session">The session.</param>
public EmsResourceHolder(ConnectionFactory connectionFactory, Connection connection, Session session)
{
this.connectionFactory = connectionFactory;
AddConnection(connection);
AddSession(session, connection);
this.frozen = true;
}
#endregion
#region Properties
/// <summary>
/// Gets a value indicating whether this <see cref="EmsResourceHolder"/> is frozen, namely that
/// additional resources can be registered with the holder. If using any of the constructors with
/// a Session, the holder will be set to the frozen state.
/// </summary>
/// <value><c>true</c> if frozen; otherwise, <c>false</c>.</value>
virtual public bool Frozen
{
get
{
return frozen;
}
}
#endregion
#region Methods
/// <summary>
/// Adds the connection to the list of resources managed by this holder.
/// </summary>
/// <param name="connection">The connection.</param>
public void AddConnection(Connection connection)
{
AssertUtils.IsTrue(!frozen, "Cannot add Connection because EmsResourceHolder is frozen");
AssertUtils.ArgumentNotNull(connection, "Connection must not be null");
if (!connections.Contains(connection))
{
connections.Add(connection);
}
}
/// <summary>
/// Adds the session to the list of resources managed by this holder.
/// </summary>
/// <param name="session">The session.</param>
public void AddSession(Session session)
{
AddSession(session, null);
}
/// <summary>
/// Adds the session and connection to the list of resources managed by this holder.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="connection">The connection.</param>
public void AddSession(Session session, Connection connection)
{
AssertUtils.IsTrue(!frozen, "Cannot add Session because EmsResourceHolder is frozen");
AssertUtils.ArgumentNotNull(session, "Session must not be null");
if (!sessions.Contains(session))
{
sessions.Add(session);
if (connection != null)
{
IList sessionsList = (IList)sessionsPerConnection[connection];
if (sessionsList == null)
{
sessionsList = new LinkedList();
sessionsPerConnection[connection] = sessionsList;
}
sessionsList.Add(session);
}
}
}
/// <summary>
/// Gets the connection managed by this resource holder
/// </summary>
/// <returns>A Connection, or null if no managed connection.</returns>
public virtual Connection GetConnection()
{
return (!(this.connections.Count == 0) ? (Connection)this.connections[0] : null);
}
/// <summary>
/// Gets the connection of a given type managed by this resource holder. This is used
/// when storing Queue or Topic Connections (from the older 1.0.2 API) as compared to the
/// 'unified domain' API , just Connection, in the newer 1.2 API.
/// </summary>
/// <param name="connectionType">Type of the connection.</param>
/// <returns>The connection, or null if not found.</returns>
public virtual Connection GetConnection(Type connectionType)
{
return (Connection)CollectionUtils.FindValueOfType(this.connections, connectionType);
}
/// <summary>
/// Gets the first session manged by this holder or null if not available.
/// </summary>
/// <returns>The session or null if not available.</returns>
public virtual Session GetSession()
{
return (!(this.sessions.Count == 0) ? (Session)this.sessions[0] : null);
}
/// <summary>
/// Gets the session managed by this holder by type.
/// </summary>
/// <param name="sessionType">Type of the session.</param>
/// <returns>The session or null if not available.</returns>
public virtual Session GetSession(Type sessionType)
{
return GetSession(sessionType, null);
}
/// <summary>
/// Gets the session of a given type associated with the given connection
/// </summary>
/// <param name="sessionType">Type of the session.</param>
/// <param name="connection">The connection.</param>
/// <returns>The sessin or null if not available.</returns>
public virtual Session GetSession(Type sessionType, Connection connection)
{
IList sessions = this.sessions;
if (connection != null)
{
sessions = (IList)sessionsPerConnection[connection];
}
return (Session)CollectionUtils.FindValueOfType(sessions, sessionType);
}
/// <summary>
/// Commits all sessions.
/// </summary>
public virtual void CommitAll()
{
foreach (Session session in sessions)
{
session.Commit();
}
}
/// <summary>
/// Closes all sessions then stops and closes all connections, in that order.
/// </summary>
public virtual void CloseAll()
{
foreach (Session session in sessions)
{
try
{
session.Close();
}
catch (Exception ex)
{
logger.Debug("Could not close EMS Session after transaction", ex);
}
}
foreach (Connection connection in connections)
{
ConnectionFactoryUtils.ReleaseConnection(connection, connectionFactory, true);
}
}
/// <summary>
/// Determines whether the holder contains the specified session.
/// </summary>
/// <param name="session">The session.</param>
/// <returns>
/// <c>true</c> if the holder contains the specified session; otherwise, <c>false</c>.
/// </returns>
public bool ContainsSession(Session session)
{
return this.sessions.Contains(session);
}
#endregion
}
}

418
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsTransactionManager.cs Normal file
View File

@@ -0,0 +1,418 @@
#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.Data;
using Common.Logging;
using Spring.Messaging.Ems.Core;
using Spring.Objects.Factory;
using Spring.Transaction;
using Spring.Transaction.Support;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Connections
{
/// <summary>
/// A <see cref="AbstractPlatformTransactionManager"/> implementation
/// for a single EMS <code>ConnectionFactory</code>. Binds a
/// Connection/Session pair from the specified ConnecctionFactory to the thread,
/// potentially allowing for one thread-bound Session per ConnectionFactory.
/// </summary>
/// <remarks>
/// <para>
/// Application code is required to retrieve the transactional Session via
/// <see cref="ConnectionFactoryUtils.GetTransactionalSession"/>. Spring's
/// <see cref="EmsTemplate"/> will autodetect a thread-bound Session and
/// automatically participate in it.
/// </para>
/// <para>
/// Transaction synchronization is turned off by default, as this manager might be used
/// alongside an IDbProvider based Spring transaction manager such as the
/// AdoPlatformTransactionManager, which has stronger needs for synchronization.
/// </para>
/// </remarks>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class EmsTransactionManager : AbstractPlatformTransactionManager,
IResourceTransactionManager, IInitializingObject
{
#region Logging Definition
private static readonly ILog LOG = LogManager.GetLogger(typeof(EmsTransactionManager));
#endregion
private ConnectionFactory connectionFactory;
/// <summary>
/// Initializes a new instance of the <see cref="EmsTransactionManager"/> class.
/// </summary>
/// <remarks>
/// The ConnectionFactory has to be set before using the instance.
/// This constructor can be used to prepare a EmsTemplate via a ApplicationContext,
/// typically setting the ConnectionFactory via ConnectionFactory property.
/// <para>
/// Turns off transaction synchronization by default, as this manager might
/// be used alongside a dbprovider-based Spring transaction manager like
/// AdoPlatformTransactionManager, which has stronger needs for synchronization.
/// Only one manager is allowed to drive synchronization at any point of time.
/// </para>
/// </remarks>
public EmsTransactionManager()
{
TransactionSynchronization = TransactionSynchronizationState.Never;
}
/// <summary>
/// Initializes a new instance of the <see cref="EmsTransactionManager"/> class
/// given a ConnectionFactory.
/// </summary>
/// <param name="connectionFactory">The connection factory to obtain connections from.</param>
public EmsTransactionManager(ConnectionFactory connectionFactory) : this()
{
ConnectionFactory = connectionFactory;
AfterPropertiesSet();
}
/// <summary>
/// Gets or sets the connection factory that this instance should manage transaction.
/// for.
/// </summary>
/// <value>The connection factory.</value>
public ConnectionFactory ConnectionFactory
{
get { return connectionFactory; }
set
{
connectionFactory = value;
}
}
#region IInitializingObject Members
/// <summary>
/// Make sure the ConnectionFactory has been set.
/// </summary>
public void AfterPropertiesSet()
{
if (ConnectionFactory == null)
{
throw new ArgumentException("Property 'ConnectionFactory' is required.");
}
}
#endregion
#region IResourceTransactionManager Members
/// <summary>
/// Gets the resource factory that this transaction manager operates on,
/// In tihs case the ConnectionFactory
/// </summary>
/// <value>The ConnectionFactory.</value>
public object ResourceFactory
{
get { return ConnectionFactory; }
}
#endregion
/// <summary>
/// Get the EmsTransactionObject.
/// </summary>
/// <returns>he EmsTransactionObject.</returns>
protected override object DoGetTransaction()
{
EmsTransactionObject txObject = new EmsTransactionObject();
txObject.ResourceHolder =
(EmsResourceHolder) TransactionSynchronizationManager.GetResource(ConnectionFactory);
return txObject;
}
/// <summary>
/// Begin a new transaction with the given transaction definition.
/// </summary>
/// <param name="transaction">Transaction object returned by
/// <see cref="Spring.Transaction.Support.AbstractPlatformTransactionManager.DoGetTransaction"/>.</param>
/// <param name="definition"><see cref="Spring.Transaction.ITransactionDefinition"/> instance, describing
/// propagation behavior, isolation level, timeout etc.</param>
/// <remarks>
/// Does not have to care about applying the propagation behavior,
/// as this has already been handled by this abstract manager.
/// </remarks>
/// <exception cref="Spring.Transaction.TransactionException">
/// In the case of creation or system errors.
/// </exception>
protected override void DoBegin(object transaction, ITransactionDefinition definition)
{
//This is the default value defined in DefaultTransactionDefinition
if (definition.TransactionIsolationLevel != IsolationLevel.ReadCommitted)
{
throw new InvalidIsolationLevelException("EMS does not support an isoliation level concept");
}
EmsTransactionObject txObject = (EmsTransactionObject) transaction;
Connection con = null;
Session session = null;
try
{
con = CreateConnection();
session = CreateSession(con);
if (LOG.IsDebugEnabled)
{
log.Debug("Created EMS transaction on Session [" + session + "] from Connection [" + con + "]");
}
txObject.ResourceHolder = new EmsResourceHolder(ConnectionFactory, con, session);
txObject.ResourceHolder.SynchronizedWithTransaction = true;
int timeout = DetermineTimeout(definition);
if (timeout != DefaultTransactionDefinition.TIMEOUT_DEFAULT)
{
txObject.ResourceHolder.TimeoutInSeconds = timeout;
}
TransactionSynchronizationManager.BindResource(ConnectionFactory, txObject.ResourceHolder);
} catch (EMSException ex)
{
if (session != null)
{
try
{
session.Close();
} catch (Exception)
{}
}
if (con != null)
{
try
{
con.Close();
} catch (Exception){}
}
throw new CannotCreateTransactionException("Could not create EMS Transaction", ex);
}
}
/// <summary>
/// Suspend the resources of the current transaction.
/// </summary>
/// <param name="transaction">Transaction object returned by
/// <see cref="Spring.Transaction.Support.AbstractPlatformTransactionManager.DoGetTransaction"/>.</param>
/// <returns>
/// An object that holds suspended resources (will be kept unexamined for passing it into
/// <see cref="Spring.Transaction.Support.AbstractPlatformTransactionManager.DoResume"/>.)
/// </returns>
/// <remarks>
/// Transaction synchronization will already have been suspended.
/// </remarks>
/// <exception cref="Spring.Transaction.TransactionException">
/// in case of system errors.
/// </exception>
protected override object DoSuspend(object transaction)
{
EmsTransactionObject txObject = (EmsTransactionObject) transaction;
txObject.ResourceHolder = null;
return TransactionSynchronizationManager.UnbindResource(ConnectionFactory);
}
/// <summary>
/// Resume the resources of the current transaction.
/// </summary>
/// <param name="transaction">Transaction object returned by
/// <see cref="Spring.Transaction.Support.AbstractPlatformTransactionManager.DoGetTransaction"/>.</param>
/// <param name="suspendedResources">The object that holds suspended resources as returned by
/// <see cref="Spring.Transaction.Support.AbstractPlatformTransactionManager.DoSuspend"/>.</param>
/// <remarks>Transaction synchronization will be resumed afterwards.
/// </remarks>
/// <exception cref="Spring.Transaction.TransactionException">
/// In the case of system errors.
/// </exception>
protected override void DoResume(object transaction, object suspendedResources)
{
EmsResourceHolder conHolder = (EmsResourceHolder) suspendedResources;
TransactionSynchronizationManager.BindResource(ConnectionFactory, conHolder);
}
/// <summary>
/// Perform an actual commit on the given transaction.
/// </summary>
/// <param name="status">The status representation of the transaction.</param>
/// <exception cref="Spring.Transaction.TransactionException">
/// In the case of system errors.
/// </exception>
protected override void DoCommit(DefaultTransactionStatus status)
{
EmsTransactionObject txObject = (EmsTransactionObject)status.Transaction;
Session session = txObject.ResourceHolder.GetSession();
try
{
if (status.Debug)
{
LOG.Debug("Committing EMS transaction on Session [" + session + "]");
}
session.Commit();
}
catch (TransactionRolledBackException ex)
{
throw new UnexpectedRollbackException("EMS transaction rolled back", ex);
}
catch (EMSException ex)
{
throw new TransactionSystemException("Could not commit EMS transaction.", ex);
}
}
/// <summary>
/// Perform an actual rollback on the given transaction.
/// </summary>
/// <param name="status">The status representation of the transaction.</param>
/// <exception cref="Spring.Transaction.TransactionException">
/// In the case of system errors.
/// </exception>
protected override void DoRollback(DefaultTransactionStatus status)
{
EmsTransactionObject txObject = (EmsTransactionObject)status.Transaction;
Session session = txObject.ResourceHolder.GetSession();
try
{
if (status.Debug)
{
LOG.Debug("Rolling back EMS transaction on Session [" + session + "]");
}
session.Rollback();
}
catch (EMSException ex)
{
throw new TransactionSystemException("Could not roll back EMS transaction.", ex);
}
}
/// <summary>
/// Set the given transaction rollback-only. Only called on rollback
/// if the current transaction takes part in an existing one.
/// </summary>
/// <param name="status">The status representation of the transaction.</param>
/// <exception cref="Spring.Transaction.TransactionException">
/// In the case of system errors.
/// </exception>
protected override void DoSetRollbackOnly(DefaultTransactionStatus status)
{
EmsTransactionObject txObject = (EmsTransactionObject)status.Transaction;
txObject.ResourceHolder.RollbackOnly = true;
}
/// <summary>
/// Cleanup resources after transaction completion.
/// </summary>
/// <param name="transaction">Transaction object returned by
/// <see cref="Spring.Transaction.Support.AbstractPlatformTransactionManager.DoGetTransaction"/>.</param>
/// <remarks>
/// <para>
/// Called after <see cref="Spring.Transaction.Support.AbstractPlatformTransactionManager.DoCommit"/>
/// and
/// <see cref="Spring.Transaction.Support.AbstractPlatformTransactionManager.DoRollback"/>
/// execution on any outcome.
/// </para>
/// </remarks>
protected override void DoCleanupAfterCompletion(object transaction)
{
EmsTransactionObject txObject = (EmsTransactionObject)transaction;
TransactionSynchronizationManager.UnbindResource(ConnectionFactory);
txObject.ResourceHolder.CloseAll();
txObject.ResourceHolder.Clear();
}
/// <summary>
/// Check if the given transaction object indicates an existing transaction
/// (that is, a transaction which has already started).
/// </summary>
/// <param name="transaction">Transaction object returned by
/// <see cref="Spring.Transaction.Support.AbstractPlatformTransactionManager.DoGetTransaction"/>.</param>
/// <returns>
/// True if there is an existing transaction.
/// </returns>
/// <exception cref="Spring.Transaction.TransactionException">
/// In the case of system errors.
/// </exception>
protected override bool IsExistingTransaction(object transaction)
{
EmsTransactionObject txObject = transaction as EmsTransactionObject;
if (txObject != null)
{
return txObject.ResourceHolder != null;
}
return false;
}
/// <summary>
/// Creates the connection via thie manager's ConnectionFactory.
/// </summary>
/// <returns>The new Connection</returns>
/// <exception cref="EMSException">If thrown by underlying messaging APIs</exception>
protected virtual Connection CreateConnection()
{
return ConnectionFactory.CreateConnection();
}
/// <summary>
/// Creates the session for the given Connection
/// </summary>
/// <param name="connection">The connection to create a Session for.</param>
/// <returns>the new Session</returns>
/// <exception cref="EMSException">If thrown by underlying messaging APIs</exception>
protected virtual Session CreateSession(Connection connection)
{
return connection.CreateSession(true, Session.SESSION_TRANSACTED);
}
/// <summary>
/// EMS Transaction object, representing a EmsResourceHolder.
/// Used as transaction object by EMSTransactionManager
/// </summary>
internal class EmsTransactionObject : ISmartTransactionObject
{
private EmsResourceHolder resourceHolder;
public EmsResourceHolder ResourceHolder
{
get { return resourceHolder; }
set { resourceHolder = value; }
}
#region ISmartTransactionObject Members
public bool RollbackOnly
{
get { return resourceHolder.RollbackOnly; }
}
#endregion
}
}
}

64
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/SynchedLocalTransactionFailedException.cs Normal file
View File

@@ -0,0 +1,64 @@
#region License
/*
* Copyright <20> 2002-2006 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.Connections
{
/// <summary> Exception thrown when a synchronized local transaction failed to complete
/// (after the main transaction has already completed).
/// </summary>
/// <author>Jergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
[Serializable]
public class SynchedLocalTransactionFailedException : EMSException
{
#region Constructor (s) / Destructor
/// <summary>
/// Creates a new instance of the SynchedLocalTransactionFailedException class. with the specified message.
/// </summary>
/// <param name="message">
/// A message about the exception.
/// </param>
public SynchedLocalTransactionFailedException (string message) : base(message)
{
}
/// <summary>
/// Creates a new instance of the SynchedLocalTransactionFailedException class with the specified message
/// and root cause.
/// </summary>
/// <param name="message">
/// A message about the exception.
/// </param>
/// <param name="rootCause">
/// The root exception that is being wrapped.
/// </param>
public SynchedLocalTransactionFailedException (string message, Exception rootCause)
: base(message)
{
LinkedException = rootCause;
}
#endregion
}
}

55
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/IMessageConverter.cs Normal file
View File

@@ -0,0 +1,55 @@
#region License
/*
* Copyright <20> 2002-2006 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.Support.Converter
{
/// <summary> Strategy interface that specifies a IMessageConverter
/// between .NET objects and EMS messages.
///
/// </summary>
/// <author>Mark Pollack</author>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public interface IMessageConverter
{
/// <summary> Convert a .NET object to a EMS Message using the supplied session
/// to create the message object.
/// </summary>
/// <param name="objectToConvert">the object to convert
/// </param>
/// <param name="session">the Session to use for creating a EMS Message
/// </param>
/// <returns> the EMS Message
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
/// <throws>MessageConversionException in case of conversion failure </throws>
Message ToMessage(object objectToConvert, Session session);
/// <summary> Convert from a EMS Message to a .NET object.</summary>
/// <param name="messageToConvert">the message to convert
/// </param>
/// <returns> the converted .NET object
/// </returns>
/// <throws>MessageConversionException in case of conversion failure </throws>
object FromMessage(Message messageToConvert);
}
}

64
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/MessageConversionException.cs Normal file
View File

@@ -0,0 +1,64 @@
#region License
/*
* Copyright <20> 2002-2006 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.Support.Converter
{
/// <summary> Thrown by IMessageConverter implementations when the conversion
/// of an object to/from a Message fails.
/// </summary>
/// <author>Mark Pollack</author>
public class MessageConversionException : EMSException
{
#region Constructor (s) / Destructor
/// <summary>
/// Creates a new instance of the IMessageConverterException class. with the specified message.
/// </summary>
/// <param name="message">
/// A message about the exception.
/// </param>
public MessageConversionException(string message)
: base(message)
{
}
/// <summary>
/// Creates a new instance of the IMessageConverterException class with the specified message
/// and root cause.
/// </summary>
/// <param name="message">
/// A message about the exception.
/// </param>
/// <param name="rootCause">
/// The root exception that is being wrapped.
/// </param>
public MessageConversionException(string message, Exception rootCause)
: base(message)
{
LinkedException = rootCause;
}
#endregion
}
}

240
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/SimpleMessageConverter.cs Normal file
View File

@@ -0,0 +1,240 @@
#region License
/*
* Copyright <20> 2002-2006 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#endregion
using System;
using System.Collections;
using System.Runtime.Serialization;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Support.Converter
{
/// <summary> A simple message converter that can handle TextMessages, BytesMessages,
/// MapMessages, and ObjectMessages. Used as default by EmsTemplate, for
/// <code>ConvertAndSend</code> and <code>ReceiveAndConvert</code> operations.
///
/// <p>Converts a String to a EMS TextMessage, a byte array to a EMS BytesMessage,
/// a Map to a EMS MapMessage, and a Serializable object to a EMS ObjectMessage
/// (or vice versa).</p>
///
///
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class SimpleMessageConverter : IMessageConverter
{
/// <summary> Convert a .NET object to a EMS Message using the supplied session
/// to create the message object.
/// </summary>
/// <param name="objectToConvert">the object to convert
/// </param>
/// <param name="session">the Session to use for creating a EMS Message
/// </param>
/// <returns> the EMS Message
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
/// <throws>MessageConversionException in case of conversion failure </throws>
public Message ToMessage(object objectToConvert, Session session)
{
if (objectToConvert is Message)
{
return (Message) objectToConvert;
}
else if (objectToConvert is string)
{
return CreateMessageForString((string) objectToConvert, session);
}
else if (objectToConvert is sbyte[])
{
return CreateMessageForByteArray((byte[]) objectToConvert, session);
}
else if (objectToConvert is IDictionary)
{
return CreateMessageForMap((IDictionary) objectToConvert, session);
}
else if (objectToConvert is ISerializable)
{
return
CreateMessageForSerializable(((ISerializable) objectToConvert), session);
}
else
{
throw new MessageConversionException("Cannot convert object [" + objectToConvert + "] to EMS message");
}
}
/// <summary> Convert from a EMS Message to a .NET object.</summary>
/// <param name="messageToConvert">the message to convert
/// </param>
/// <returns> the converted .NET object
/// </returns>
/// <throws>MessageConversionException in case of conversion failure </throws>
public object FromMessage(Message messageToConvert)
{
if (messageToConvert is TextMessage)
{
return ExtractStringFromMessage((TextMessage) messageToConvert);
}
else if (messageToConvert is BytesMessage)
{
return ExtractByteArrayFromMessage((BytesMessage) messageToConvert);
}
else if (messageToConvert is MapMessage)
{
return ExtractMapFromMessage((MapMessage) messageToConvert);
}
else if (messageToConvert is ObjectMessage)
{
return ExtractSerializableFromMessage((ObjectMessage) messageToConvert);
}
else
{
return messageToConvert;
}
}
#region To Converter Methods
/// <summary> Create a EMS TextMessage for the given String.</summary>
/// <param name="text">the String to convert
/// </param>
/// <param name="session">current EMS session
/// </param>
/// <returns> the resulting message
/// </returns>
/// <throws> EMSException if thrown by EMS methods </throws>
protected virtual TextMessage CreateMessageForString(string text, Session session)
{
return session.CreateTextMessage((text));
}
/// <summary> Create a EMS BytesMessage for the given byte array.</summary>
/// <param name="bytes">the byyte array to convert
/// </param>
/// <param name="session">current EMS session
/// </param>
/// <returns> the resulting message
/// </returns>
/// <throws> EMSException if thrown by EMS methods </throws>
protected virtual BytesMessage CreateMessageForByteArray(byte[] bytes, Session session)
{
BytesMessage message = session.CreateBytesMessage();
message.WriteBytes(bytes);
return message;
}
/// <summary> Create a EMS MapMessage for the given Map.</summary>
/// <param name="map">the Map to convert
/// </param>
/// <param name="session">current EMS session
/// </param>
/// <returns> the resulting message
/// </returns>
/// <throws> EMSException if thrown by EMS methods </throws>
protected virtual MapMessage CreateMessageForMap(IDictionary map, Session session)
{
MapMessage mapMessage = session.CreateMapMessage();
foreach (DictionaryEntry entry in map)
{
if (!(entry.Key is string))
{
throw new MessageConversionException("Cannot convert non-String key of type [" +
(entry.Key != null ? entry.Key.GetType().FullName : null) +
"] to MapMessage entry");
}
mapMessage.SetObject(entry.Key.ToString(), entry.Value);
}
return mapMessage;
}
/// <summary> Create a EMS ObjectMessage for the given Serializable object.</summary>
/// <param name="objectToSend">the Serializable object to convert
/// </param>
/// <param name="session">current EMS session
/// </param>
/// <returns> the resulting message
/// </returns>
/// <throws> EMSException if thrown by EMS methods </throws>
protected virtual ObjectMessage CreateMessageForSerializable(
ISerializable objectToSend, Session session)
{
return session.CreateObjectMessage(objectToSend);
}
#endregion
#region From Converter Mehtods
/// <summary> Extract a String from the given TextMessage.</summary>
/// <param name="message">the message to convert
/// </param>
/// <returns> the resulting String
/// </returns>
/// <throws> EMSException if thrown by EMS methods </throws>
protected virtual string ExtractStringFromMessage(TextMessage message)
{
return message.Text;
}
/// <summary> Extract a byte array from the given BytesMessage.</summary>
/// <param name="message">the message to convert
/// </param>
/// <returns> the resulting byte array
/// </returns>
/// <throws> EMSException if thrown by EMS methods </throws>
protected virtual byte[] ExtractByteArrayFromMessage(BytesMessage message)
{
byte[] bytes = new byte[(int)message.BodyLength];
message.ReadBytes(bytes);
return bytes;
}
/// <summary> Extract a IDictionary from the given MapMessage.</summary>
/// <param name="message">the message to convert
/// </param>
/// <returns> the resulting Map
/// </returns>
/// <throws>EMSException if thrown by EMS methods </throws>
protected virtual IDictionary ExtractMapFromMessage(MapMessage message)
{
IDictionary dictionary = new Hashtable();
IEnumerator e = message.MapNames;
while (e.MoveNext())
{
String key = (String)e.Current;
dictionary.Add(key, message.GetObject(key));
}
return dictionary;
}
/// <summary>
/// Extracts the serializable object from the given object message.
/// </summary>
/// <param name="message">The message to convert.</param>
/// <returns>The resulting serializable object.</returns>
protected virtual object ExtractSerializableFromMessage(
ObjectMessage message)
{
return message.TheObject as ISerializable;
}
#endregion
}
}

86
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/DynamicDestinationResolver.cs Normal file
View File

@@ -0,0 +1,86 @@
#region License
/*
* Copyright <20> 2002-2006 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.Util;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Support.Destinations
{
/// <summary> Simple DestinationResolver implementation resolving destination names
/// as dynamic destinations.</summary>
///
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class DynamicDestinationResolver : DestinationResolver
{
/// <summary> Resolve the given destination name, either as located resource
/// or as dynamic destination.
/// </summary>
/// <param name="session">the current EMS Session
/// </param>
/// <param name="destinationName">the name of the destination
/// </param>
/// <param name="pubSubDomain"><code>true</code> if the domain is pub-sub, <code>false</code> if P2P
/// </param>
/// <returns> the EMS destination (either a topic or a queue)
/// </returns>
/// <throws>EMSException if resolution failed </throws>
public Destination ResolveDestinationName(Session session, string destinationName, bool pubSubDomain)
{
AssertUtils.ArgumentNotNull(session, "Session must not be null");
AssertUtils.ArgumentNotNull(destinationName, "Destination name must not be null");
if (pubSubDomain)
{
return ResolveTopic(session, destinationName);
}
else
{
return ResolveQueue(session, destinationName);
}
}
/// <summary> Resolve the given destination name to a Topic.</summary>
/// <param name="session">the current EMS Session
/// </param>
/// <param name="topicName">the name of the desired Topic.
/// </param>
/// <returns> the EMS Topic name
/// </returns>
/// <throws>EMSException if resolution failed </throws>
protected internal virtual Destination ResolveTopic(Session session, System.String topicName)
{
return session.CreateTopic(topicName);
}
/// <summary> Resolve the given destination name to a Queue.</summary>
/// <param name="session">the current EMS Session
/// </param>
/// <param name="queueName">the name of the desired Queue.
/// </param>
/// <returns> the EMS Queue name
/// </returns>
/// <throws>EMSException if resolution failed </throws>
protected internal virtual Destination ResolveQueue(Session session, string queueName)
{
return session.CreateQueue(queueName);
}
}
}

110
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/EmsDestinationAccessor.cs Normal file
View File

@@ -0,0 +1,110 @@
#region License
/*
* Copyright <20> 2002-2006 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.Util;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Support.Destinations
{
/// <summary> Base class for EmsTemplate} and other
/// EMS-accessing gateway helpers, adding destination-related properties to
/// EmsAccessor's common properties.
/// </summary>
/// <remarks>
/// <p>Not intended to be used directly. See EmsTemplate.</p>
///
/// </remarks>
/// <author>Juergen Hoeller </author>
/// <author>Mark Pollack (.NET)</author>
public class EmsDestinationAccessor : EmsAccessor
{
#region Fields
private DestinationResolver destinationResolver = new DynamicDestinationResolver();
private bool pubSubDomain = false;
#endregion
#region Properties
/// <summary>
/// Gets or sets the destination resolver that is to be used to resolve
/// Destination references for this accessor.
/// </summary>
/// <remarks>The default resolver is a DynamicDestinationResolver. Specify a
/// JndDestinationResolver for resolving destination names as JNDI locations.
/// </remarks>
/// <value>The destination resolver.</value>
virtual public DestinationResolver DestinationResolver
{
get
{
return destinationResolver;
}
set
{
AssertUtils.ArgumentNotNull(value, "DestinationResolver must not be null");
this.destinationResolver = value;
}
}
/// <summary>
/// Gets or sets a value indicating whether Publish/Subscribe
/// domain (Topics) is used. Otherwise, the Point-to-Point domain
/// (Queues) is used.
///
/// </summary>
/// <remarks>this
/// setting tells what type of destination to create if dynamic destinations are enabled.</remarks>
/// <value><c>true</c> if Publish/Subscribe domain; otherwise, <c>false</c>
/// for the Point-to-Point domain.</value>
public virtual bool PubSubDomain
{
get
{
return pubSubDomain;
}
set
{
this.pubSubDomain = value;
}
}
#endregion
/// <summary>
/// Resolves the given destination name to a EMS destination.
/// </summary>
/// <param name="session">The current session.</param>
/// <param name="destinationName">Name of the destination.</param>
/// <returns>The located Destination</returns>
/// <exception cref="EMSException">If resolution failed.</exception>
public virtual Destination ResolveDestinationName(Session session, System.String destinationName)
{
return DestinationResolver.ResolveDestinationName(session, destinationName, PubSubDomain);
}
}
}

58
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/IDestinationResolver.cs Normal file
View File

@@ -0,0 +1,58 @@
#region License
/*
* Copyright <20> 2002-2006 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.Support.Destinations
{
/// <summary> Strategy interface for resolving EMS destinations.
/// </summary>
/// <remarks>
/// <para>Used by EmsTemplate for resolving
/// destination names from simple Strings to actual
/// Destination implementation instances.
/// </para>
///
/// <para>The default DestinationResolver implementation used by
/// EmsTemplate instances is the
/// DynamicDestinationResolver class. Consider using the
/// JndDestinationResolver for more advanced scenarios.
/// </para>
/// </remarks>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public interface DestinationResolver
{
/// <summary> Resolve the given destination name, either as located resource
/// or as dynamic destination.
/// </summary>
/// <param name="session">the current EMS Session
/// </param>
/// <param name="destinationName">the name of the destination
/// </param>
/// <param name="pubSubDomain"><code>true</code> if the domain is pub-sub, <code>false</code> if P2P
/// </param>
/// <returns> the EMS destination (either a topic or a queue)
/// </returns>
/// <throws>EMSException if resolution failed </throws>
Destination ResolveDestinationName(Session session, string destinationName, bool pubSubDomain);
}
}

170
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsAccessor.cs Normal file
View File

@@ -0,0 +1,170 @@
#region License
/*
* Copyright <20> 2002-2006 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.Objects.Factory;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Support
{
/// <summary> Base class for EmsTemplate and other EMS-accessing gateway helpers</summary>
/// <remarks>It defines common properties like the ConnectionFactory}. The subclass
/// EmsDestinationAccessor adds further, destination-related properties.
/// <para>
/// Not intended to be used directly. See EmsTemplate.
/// </para>
/// </remarks>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class EmsAccessor : IInitializingObject
{
#region Logging
private readonly ILog logger = LogManager.GetLogger(typeof(EmsAccessor));
#endregion
#region Fields
private ConnectionFactory connectionFactory;
private bool sessionTransacted = false;
private int sessionAcknowledgeMode = Session.AUTO_ACKNOWLEDGE;
#endregion
#region Properties
/// <summary>
/// Gets or sets the connection factory to use for obtaining EMS Connections.
/// </summary>
/// <value>The connection factory.</value>
virtual public ConnectionFactory ConnectionFactory
{
get
{
return connectionFactory;
}
set
{
this.connectionFactory = value;
}
}
/// <summary>
/// Gets or sets the session acknowledge mode for EMS Sessions including whether or not the session is transacted
/// </summary>
/// <remarks>
/// Set the EMS acknowledgement mode that is used when creating a EMS
/// Session to send a message. The default is AUTO_ACKNOWLEDGE.
/// </remarks>
/// <value>The session acknowledge mode.</value>
virtual public int SessionAcknowledgeMode
{
get
{
return sessionAcknowledgeMode;
}
set
{
this.sessionAcknowledgeMode = value;
}
}
/// <summary>
/// Set the transaction mode that is used when creating a EMS Session.
/// Default is "false".
/// </summary>
/// <remarks>
/// <para>Setting this flag to "true" will use a short local EMS transaction
/// when running outside of a managed transaction, and a synchronized local
/// EMS transaction in case of a managed transaction being present.
/// The latter has the effect of a local EMS
/// transaction being managed alongside the main transaction (which might
/// be a native ADO.NET transaction), with the EMS transaction committing
/// right after the main transaction.
/// </para>
/// </remarks>
public bool SessionTransacted
{
get
{
return sessionTransacted;
}
set
{
if (value)
{
sessionTransacted = value;
}
}
}
#endregion
/// <summary>
/// Verify that ConnectionFactory property has been set.
/// </summary>
public virtual void AfterPropertiesSet()
{
if (ConnectionFactory == null)
{
throw new ArgumentException("ConnectionFactory is required");
}
}
/// <summary>
/// Creates the connection via the ConnectionFactory.
/// </summary>
/// <returns></returns>
protected virtual Connection CreateConnection()
{
return ConnectionFactory.CreateConnection();
}
/// <summary>
/// Creates the session for the given Connection
/// </summary>
/// <param name="con">The connection to create a session for.</param>
/// <returns>The new session</returns>
protected virtual Session CreateSession(Connection con)
{
return con.CreateSession(sessionTransacted, SessionAcknowledgeMode);
}
/// <summary>
/// Returns whether the Session is in client acknowledgement mode.
/// </summary>
/// <param name="session">The session to check.</param>
/// <returns>true if in client ack mode, false otherwise</returns>
protected virtual bool IsClientAcknowledge(Session session)
{
return (session.AcknowledgeMode == Session.CLIENT_ACKNOWLEDGE);
}
}
}

243
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsUtils.cs Normal file
View File

@@ -0,0 +1,243 @@
#region License
/*
* Copyright <20> 2002-2006 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.Util;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Support
{
/// <summary>
/// Generic utility methods for working with EMS. Mainly for internal use
/// within the framework, but also useful for custom EMS access code.
/// </summary>
public abstract class EmsUtils
{
#region Logging
private static readonly ILog logger = LogManager.GetLogger(typeof(EmsUtils));
#endregion
/// <summary> Close the given EMS Connection and ignore any thrown exception.
/// This is useful for typical <code>finally</code> blocks in manual EMS code.
/// </summary>
/// <param name="con">the EMS Connection to close (may be <code>null</code>)
/// </param>
public static void CloseConnection(Connection con)
{
CloseConnection(con, false);
}
/// <summary> Close the given EMS Connection and ignore any thrown exception.
/// This is useful for typical <code>finally</code> blocks in manual EMS code.
/// </summary>
/// <param name="con">the EMS Connection to close (may be <code>null</code>)
/// </param>
/// <param name="stop">whether to call <code>stop()</code> before closing
/// </param>
public static void CloseConnection(Connection con, bool stop)
{
if (con != null)
{
try
{
if (stop)
{
try
{
con.Stop();
}
finally
{
con.Close();
}
}
else
{
con.Close();
}
}
catch (EMSException ex)
{
logger.Debug("Could not close EMS Connection", ex);
}
catch (Exception ex)
{
// We don't trust the EMS provider: It might throw another exception.
logger.Debug("Unexpected exception on closing EMS Connection", ex);
}
}
}
/// <summary> Close the given EMS Session and ignore any thrown exception.
/// This is useful for typical <code>finally</code> blocks in manual EMS code.
/// </summary>
/// <param name="session">the EMS Session to close (may be <code>null</code>)
/// </param>
public static void CloseSession(Session session)
{
if (session != null)
{
try
{
session.Close();
}
catch (EMSException ex)
{
logger.Debug("Could not close EMS Session", ex);
}
catch (Exception ex)
{
// We don't trust the EMS provider: It might throw RuntimeException or Error.
logger.Debug("Unexpected exception on closing EMS Session", ex);
}
}
}
/// <summary> Close the given EMS MessageProducer and ignore any thrown exception.
/// This is useful for typical <code>finally</code> blocks in manual EMS code.
/// </summary>
/// <param name="producer">the EMS MessageProducer to close (may be <code>null</code>)
/// </param>
public static void CloseMessageProducer(MessageProducer producer)
{
if (producer != null)
{
try
{
producer.Close();
}
catch (EMSException ex)
{
logger.Debug("Could not close EMS MessageProducer", ex);
}
catch (Exception ex)
{
// We don't trust the provider: It might throw an exception .
logger.Debug("Unexpected exception on closing EMS MessageProducer", ex);
}
}
}
/// <summary> Close the given EMS MessageConsumer and ignore any thrown exception.
/// This is useful for typical <code>finally</code> blocks in manual EMS code.
/// </summary>
/// <param name="consumer">the EMS MessageConsumer to close (may be <code>null</code>)
/// </param>
public static void CloseMessageConsumer(MessageConsumer consumer)
{
if (consumer != null)
{
try
{
consumer.Close();
}
catch (EMSException ex)
{
logger.Debug("Could not close EMS MessageConsumer", ex);
}
catch (Exception ex)
{
// We don't trust the EMS provider: It might throw RuntimeException or Error.
logger.Debug("Unexpected exception on closing EMS MessageConsumer", ex);
}
}
}
/*
/// <summary> Close the given EMS QueueRequestor and ignore any thrown exception.
/// This is useful for typical <code>finally</code> blocks in manual EMS code.
/// </summary>
/// <param name="requestor">the EMS QueueRequestor to close (may be <code>null</code>)
/// </param>
*/
// public static void CloseQueueRequestor(QueueRequestor requestor)
// {
// if (requestor != null)
// {
// try
// {
// requestor.Close();
// }
// catch (EMSException ex)
// {
// logger.Debug("Could not close EMS QueueRequestor", ex);
// }
// catch (Exception ex)
// {
// // We don't trust the EMS provider: It might throw RuntimeException or Error.
// logger.Debug("Unexpected exception on closing EMS QueueRequestor", ex);
// }
// }
// }
/// <summary> Commit the Session if not within a distributed transaction.</summary>
/// <remarks>Needs investigation - no distributed tx in .NET messaging providers</remarks>
/// <param name="session">the EMS Session to commit
/// </param>
/// <throws>EMSException if committing failed </throws>
public static void CommitIfNecessary(Session session)
{
AssertUtils.ArgumentNotNull(session, "Session must not be null");
session.Commit();
// TODO Investigate
// try {
// session.Commit();
// }
// catch (TransactionInProgressException ex) {
// // TODO Investigate
// // Ignore -> can only happen in case of a JTA transaction.
// }
// catch (IllegalStateException ex) {
// // TODO Investigate
// // Ignore -> can only happen in case of a JTA transaction.
// }
}
/// <summary> Rollback the Session if not within a distributed transaction.</summary>
/// <remarks>Needs investigation - no distributed tx in EMS</remarks>
/// <param name="session">the EMS Session to rollback
/// </param>
/// <throws> EMSException if committing failed </throws>
public static void RollbackIfNecessary(Session session)
{
AssertUtils.ArgumentNotNull(session, "Session must not be null");
session.Rollback();
// TODO Investigate
// try {
// session.Rollback();
// }
// catch (TransactionInProgressException ex) {
// // Ignore -> can only happen in case of a JTA transaction.
// }
// catch (IllegalStateException ex) {
// // Ignore -> can only happen in case of a JTA transaction.
// }
}
}
}