From 927f1ed2097090e3c2e7308c6cb4d4a0575589e3 Mon Sep 17 00:00:00 2001 From: markpollack Date: Tue, 12 Aug 2008 22:24:08 +0000 Subject: [PATCH] TIBCO EMS Integration - SPRNET-982 --- .../Connections/ChainedExceptionListener.cs | 71 +++ .../Ems/Connections/ConnectionFactoryUtils.cs | 378 ++++++++++++++++ .../Ems/Connections/EmsResourceHolder.cs | 292 ++++++++++++ .../Ems/Connections/EmsTransactionManager.cs | 418 ++++++++++++++++++ .../SynchedLocalTransactionFailedException.cs | 64 +++ .../Support/Converter/IMessageConverter.cs | 55 +++ .../Converter/MessageConversionException.cs | 64 +++ .../Converter/SimpleMessageConverter.cs | 240 ++++++++++ .../DynamicDestinationResolver.cs | 86 ++++ .../Destinations/EmsDestinationAccessor.cs | 110 +++++ .../Destinations/IDestinationResolver.cs | 58 +++ .../Messaging/Ems/Support/EmsAccessor.cs | 170 +++++++ .../Messaging/Ems/Support/EmsUtils.cs | 243 ++++++++++ 13 files changed, 2249 insertions(+) create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ChainedExceptionListener.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ConnectionFactoryUtils.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsResourceHolder.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsTransactionManager.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/SynchedLocalTransactionFailedException.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/IMessageConverter.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/MessageConversionException.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/SimpleMessageConverter.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/DynamicDestinationResolver.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/EmsDestinationAccessor.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/IDestinationResolver.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsAccessor.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsUtils.cs diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ChainedExceptionListener.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ChainedExceptionListener.cs new file mode 100644 index 00000000..f2011087 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ChainedExceptionListener.cs @@ -0,0 +1,71 @@ +#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.Collections; +using Spring.Util; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Connections +{ + /// + /// Implementation of Spring IExceptionListener interface that supports + /// chaining allowing the addition of multiple ExceptionListener instances in order. + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class ChainedExceptionListener : IExceptionListener + { + private ArrayList listeners = new ArrayList(2); + + /// + /// Adds the exception listener to the chain + /// + /// The listener. + public void AddListener(IExceptionListener listener) + { + AssertUtils.ArgumentNotNull(listener, "listener", "ExceptionListener must not be null"); + listeners.Add(listener); + } + + /// + /// Called when an exception occurs in message processing. + /// + /// The exception. + public void OnException(EMSException exception) + { + foreach (IExceptionListener listener in listeners) + { + listener.OnException(exception); + } + } + + /// + /// Gets the exception listeners as an array. + /// + /// The exception listeners. + public IExceptionListener[] Listeners + { + get + { + return (IExceptionListener[]) listeners.ToArray(typeof (IExceptionListener)); + } + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ConnectionFactoryUtils.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ConnectionFactoryUtils.cs new file mode 100644 index 00000000..9b7622df --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ConnectionFactoryUtils.cs @@ -0,0 +1,378 @@ +#region License + +/* + * Copyright © 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 +{ + /// Helper class for obtaining transactional EMS resources + /// for a given ConnectionFactory. + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public abstract class ConnectionFactoryUtils + { + #region Logging + + private static readonly ILog LOG = LogManager.GetLogger(typeof(ConnectionFactoryUtils)); + + #endregion + + /// + /// Releases the given connection. + /// + /// The connection to release. (if this is null, the call will be ignored) + /// The ConnectionFactory that the Connection was obtained from. (may be null) + /// whether the Connection might have been started by the application. + 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); + } + } + + /// + /// Determines whether the given JMS Session is transactional, that is, + /// bound to the current thread by Spring's transaction facilities. + /// + /// The session to check. + /// The ConnectionFactory that the Session originated from + /// + /// true if is session transactional, bound to current thread; otherwise, false. + /// + 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)); + } + + /// Obtain a EMS Session that is synchronized with the current transaction, if any. + /// the ConnectionFactory to obtain a Session for + /// + /// the existing EMS Connection to obtain a Session for + /// (may be null) + /// + /// 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. + /// + /// the transactional Session, or null if none found + /// + /// EMSException in case of EMS failure + public static Session GetTransactionalSession(ConnectionFactory cf, Connection existingCon, + bool synchedLocalTransactionAllowed) + { + return + DoGetTransactionalSession(cf, + new AnonymousClassResourceFactory(existingCon, cf, + synchedLocalTransactionAllowed), true); + } + + /// + /// Obtain a EMS Session that is synchronized with the current transaction, if any. + /// + /// the TransactionSynchronizationManager key to bind to + /// (usually the ConnectionFactory) + /// the ResourceFactory to use for extracting or creating + /// EMS resources + /// 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 false. + /// + /// the transactional Session, or null if none found + /// + /// EMSException in case of EMS failure + 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 + + /// Callback interface for resource creation. + /// Serving as argument for the DoGetTransactionalSession method. + /// + public interface ResourceFactory + { + /// Fetch an appropriate Session from the given EmsResourceHolder. + /// the EmsResourceHolder + /// + /// an appropriate Session fetched from the holder, + /// or null if none found + /// + Session GetSession(EmsResourceHolder holder); + + /// Fetch an appropriate Connection from the given EmsResourceHolder. + /// the EmsResourceHolder + /// + /// an appropriate Connection fetched from the holder, + /// or null if none found + /// + Connection GetConnection(EmsResourceHolder holder); + + /// Create a new EMS Connection for registration with a EmsResourceHolder. + /// the new EMS Connection + /// + /// EMSException if thrown by EMS API methods + Connection CreateConnection(); + + /// Create a new EMS Session for registration with a EmsResourceHolder. + /// the EMS Connection to create a Session for + /// + /// the new EMS Session + /// + /// EMSException if thrown by EMS API methods + Session CreateSession(Connection con); + + + /// + /// 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 + /// + /// + bool SynchedLocalTransactionAllowed { get; } + } + + /// Callback for resource cleanup at the end of a non-native EMS transaction + /// + 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 + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsResourceHolder.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsResourceHolder.cs new file mode 100644 index 00000000..d3b9c428 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsResourceHolder.cs @@ -0,0 +1,292 @@ +#region License + +/* + * Copyright © 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 +{ + /// Connection holder, wrapping a EMS Connection and a EMS Session. + /// EmsTransactionManager binds instances of this class to the thread, + /// for a given EMS ConnectionFactory. + /// + ///

Note: This is an SPI class, not intended to be used by applications.

+ /// + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + 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) + + /// Create a new EmsResourceHolder that is open for resources to be added. + public EmsResourceHolder() + { + } + + + /// + /// Initializes a new instance of the class + /// at is open for resources to be added. + /// + /// The connection factory that this + /// resource holder is associated with (may be null) + /// + public EmsResourceHolder(ConnectionFactory connectionFactory) + { + this.connectionFactory = connectionFactory; + } + + /// + /// Initializes a new instance of the class for the + /// given Session. + /// + /// The session. + public EmsResourceHolder(Session session) + { + AddSession(session); + frozen = true; + } + + /// Create a new EmsResourceHolder for the given EMS resources. + /// the EMS Connection + /// + /// the EMS Session + /// + public EmsResourceHolder(Connection connection, Session session) + { + AddConnection(connection); + AddSession(session, connection); + this.frozen = true; + } + + /// + /// Initializes a new instance of the class. + /// + /// The connection factory. + /// The connection. + /// The session. + public EmsResourceHolder(ConnectionFactory connectionFactory, Connection connection, Session session) + { + this.connectionFactory = connectionFactory; + AddConnection(connection); + AddSession(session, connection); + this.frozen = true; + } + #endregion + + #region Properties + + /// + /// Gets a value indicating whether this 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. + /// + /// true if frozen; otherwise, false. + virtual public bool Frozen + { + get + { + return frozen; + } + + } + #endregion + + #region Methods + + /// + /// Adds the connection to the list of resources managed by this holder. + /// + /// The connection. + 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); + } + } + + /// + /// Adds the session to the list of resources managed by this holder. + /// + /// The session. + public void AddSession(Session session) + { + AddSession(session, null); + } + + /// + /// Adds the session and connection to the list of resources managed by this holder. + /// + /// The session. + /// The connection. + 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); + } + } + } + + /// + /// Gets the connection managed by this resource holder + /// + /// A Connection, or null if no managed connection. + public virtual Connection GetConnection() + { + return (!(this.connections.Count == 0) ? (Connection)this.connections[0] : null); + } + + /// + /// 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. + /// + /// Type of the connection. + /// The connection, or null if not found. + public virtual Connection GetConnection(Type connectionType) + { + return (Connection)CollectionUtils.FindValueOfType(this.connections, connectionType); + } + + /// + /// Gets the first session manged by this holder or null if not available. + /// + /// The session or null if not available. + public virtual Session GetSession() + { + return (!(this.sessions.Count == 0) ? (Session)this.sessions[0] : null); + } + + /// + /// Gets the session managed by this holder by type. + /// + /// Type of the session. + /// The session or null if not available. + public virtual Session GetSession(Type sessionType) + { + return GetSession(sessionType, null); + } + + /// + /// Gets the session of a given type associated with the given connection + /// + /// Type of the session. + /// The connection. + /// The sessin or null if not available. + 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); + } + + /// + /// Commits all sessions. + /// + public virtual void CommitAll() + { + foreach (Session session in sessions) + { + session.Commit(); + } + } + + /// + /// Closes all sessions then stops and closes all connections, in that order. + /// + 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); + } + } + + /// + /// Determines whether the holder contains the specified session. + /// + /// The session. + /// + /// true if the holder contains the specified session; otherwise, false. + /// + public bool ContainsSession(Session session) + { + return this.sessions.Contains(session); + } + + #endregion + } +} diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsTransactionManager.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsTransactionManager.cs new file mode 100644 index 00000000..2b1b6f3f --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsTransactionManager.cs @@ -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 +{ + /// + /// A implementation + /// for a single EMS ConnectionFactory. Binds a + /// Connection/Session pair from the specified ConnecctionFactory to the thread, + /// potentially allowing for one thread-bound Session per ConnectionFactory. + /// + /// + /// + /// Application code is required to retrieve the transactional Session via + /// . Spring's + /// will autodetect a thread-bound Session and + /// automatically participate in it. + /// + /// + /// 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. + /// + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class EmsTransactionManager : AbstractPlatformTransactionManager, + IResourceTransactionManager, IInitializingObject + { + + #region Logging Definition + + private static readonly ILog LOG = LogManager.GetLogger(typeof(EmsTransactionManager)); + + #endregion + + private ConnectionFactory connectionFactory; + + /// + /// Initializes a new instance of the class. + /// + /// + /// 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. + /// + /// 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. + /// + /// + public EmsTransactionManager() + { + TransactionSynchronization = TransactionSynchronizationState.Never; + } + + /// + /// Initializes a new instance of the class + /// given a ConnectionFactory. + /// + /// The connection factory to obtain connections from. + public EmsTransactionManager(ConnectionFactory connectionFactory) : this() + { + ConnectionFactory = connectionFactory; + AfterPropertiesSet(); + } + + + /// + /// Gets or sets the connection factory that this instance should manage transaction. + /// for. + /// + /// The connection factory. + public ConnectionFactory ConnectionFactory + { + get { return connectionFactory; } + set + { + connectionFactory = value; + } + } + + #region IInitializingObject Members + + /// + /// Make sure the ConnectionFactory has been set. + /// + public void AfterPropertiesSet() + { + if (ConnectionFactory == null) + { + throw new ArgumentException("Property 'ConnectionFactory' is required."); + } + } + + #endregion + + #region IResourceTransactionManager Members + + /// + /// Gets the resource factory that this transaction manager operates on, + /// In tihs case the ConnectionFactory + /// + /// The ConnectionFactory. + public object ResourceFactory + { + get { return ConnectionFactory; } + } + + #endregion + + + /// + /// Get the EmsTransactionObject. + /// + /// he EmsTransactionObject. + protected override object DoGetTransaction() + { + EmsTransactionObject txObject = new EmsTransactionObject(); + + txObject.ResourceHolder = + (EmsResourceHolder) TransactionSynchronizationManager.GetResource(ConnectionFactory); + return txObject; + } + + + /// + /// Begin a new transaction with the given transaction definition. + /// + /// Transaction object returned by + /// . + /// instance, describing + /// propagation behavior, isolation level, timeout etc. + /// + /// Does not have to care about applying the propagation behavior, + /// as this has already been handled by this abstract manager. + /// + /// + /// In the case of creation or system errors. + /// + 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); + } + + } + + /// + /// Suspend the resources of the current transaction. + /// + /// Transaction object returned by + /// . + /// + /// An object that holds suspended resources (will be kept unexamined for passing it into + /// .) + /// + /// + /// Transaction synchronization will already have been suspended. + /// + /// + /// in case of system errors. + /// + protected override object DoSuspend(object transaction) + { + EmsTransactionObject txObject = (EmsTransactionObject) transaction; + txObject.ResourceHolder = null; + return TransactionSynchronizationManager.UnbindResource(ConnectionFactory); + } + + /// + /// Resume the resources of the current transaction. + /// + /// Transaction object returned by + /// . + /// The object that holds suspended resources as returned by + /// . + /// Transaction synchronization will be resumed afterwards. + /// + /// + /// In the case of system errors. + /// + protected override void DoResume(object transaction, object suspendedResources) + { + EmsResourceHolder conHolder = (EmsResourceHolder) suspendedResources; + TransactionSynchronizationManager.BindResource(ConnectionFactory, conHolder); + } + + /// + /// Perform an actual commit on the given transaction. + /// + /// The status representation of the transaction. + /// + /// In the case of system errors. + /// + 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); + } + } + + /// + /// Perform an actual rollback on the given transaction. + /// + /// The status representation of the transaction. + /// + /// In the case of system errors. + /// + 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); + } + } + + + /// + /// Set the given transaction rollback-only. Only called on rollback + /// if the current transaction takes part in an existing one. + /// + /// The status representation of the transaction. + /// + /// In the case of system errors. + /// + protected override void DoSetRollbackOnly(DefaultTransactionStatus status) + { + EmsTransactionObject txObject = (EmsTransactionObject)status.Transaction; + txObject.ResourceHolder.RollbackOnly = true; + } + + /// + /// Cleanup resources after transaction completion. + /// + /// Transaction object returned by + /// . + /// + /// + /// Called after + /// and + /// + /// execution on any outcome. + /// + /// + protected override void DoCleanupAfterCompletion(object transaction) + { + EmsTransactionObject txObject = (EmsTransactionObject)transaction; + TransactionSynchronizationManager.UnbindResource(ConnectionFactory); + txObject.ResourceHolder.CloseAll(); + txObject.ResourceHolder.Clear(); + } + + /// + /// Check if the given transaction object indicates an existing transaction + /// (that is, a transaction which has already started). + /// + /// Transaction object returned by + /// . + /// + /// True if there is an existing transaction. + /// + /// + /// In the case of system errors. + /// + protected override bool IsExistingTransaction(object transaction) + { + EmsTransactionObject txObject = transaction as EmsTransactionObject; + if (txObject != null) + { + return txObject.ResourceHolder != null; + } + return false; + } + + /// + /// Creates the connection via thie manager's ConnectionFactory. + /// + /// The new Connection + /// If thrown by underlying messaging APIs + protected virtual Connection CreateConnection() + { + return ConnectionFactory.CreateConnection(); + } + + /// + /// Creates the session for the given Connection + /// + /// The connection to create a Session for. + /// the new Session + /// If thrown by underlying messaging APIs + protected virtual Session CreateSession(Connection connection) + { + return connection.CreateSession(true, Session.SESSION_TRANSACTED); + } + + + /// + /// EMS Transaction object, representing a EmsResourceHolder. + /// Used as transaction object by EMSTransactionManager + /// + 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 + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/SynchedLocalTransactionFailedException.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/SynchedLocalTransactionFailedException.cs new file mode 100644 index 00000000..dd17771a --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/SynchedLocalTransactionFailedException.cs @@ -0,0 +1,64 @@ +#region License + +/* + * Copyright © 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 +{ + /// Exception thrown when a synchronized local transaction failed to complete + /// (after the main transaction has already completed). + /// + /// Jergen Hoeller + /// Mark Pollack (.NET) + [Serializable] + public class SynchedLocalTransactionFailedException : EMSException + { + #region Constructor (s) / Destructor + + /// + /// Creates a new instance of the SynchedLocalTransactionFailedException class. with the specified message. + /// + /// + /// A message about the exception. + /// + public SynchedLocalTransactionFailedException (string message) : base(message) + { + } + + /// + /// Creates a new instance of the SynchedLocalTransactionFailedException class with the specified message + /// and root cause. + /// + /// + /// A message about the exception. + /// + /// + /// The root exception that is being wrapped. + /// + public SynchedLocalTransactionFailedException (string message, Exception rootCause) + : base(message) + { + LinkedException = rootCause; + } + + #endregion + } +} diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/IMessageConverter.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/IMessageConverter.cs new file mode 100644 index 00000000..7dc82fac --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/IMessageConverter.cs @@ -0,0 +1,55 @@ +#region License + +/* + * Copyright © 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 +{ + /// Strategy interface that specifies a IMessageConverter + /// between .NET objects and EMS messages. + /// + /// + /// Mark Pollack + /// Juergen Hoeller + /// Mark Pollack (.NET) + public interface IMessageConverter + { + /// Convert a .NET object to a EMS Message using the supplied session + /// to create the message object. + /// + /// the object to convert + /// + /// the Session to use for creating a EMS Message + /// + /// the EMS Message + /// + /// EMSException if thrown by EMS API methods + /// MessageConversionException in case of conversion failure + Message ToMessage(object objectToConvert, Session session); + + /// Convert from a EMS Message to a .NET object. + /// the message to convert + /// + /// the converted .NET object + /// + /// MessageConversionException in case of conversion failure + object FromMessage(Message messageToConvert); + } +} diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/MessageConversionException.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/MessageConversionException.cs new file mode 100644 index 00000000..b5b0ee93 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/MessageConversionException.cs @@ -0,0 +1,64 @@ +#region License + +/* + * Copyright © 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 +{ + /// Thrown by IMessageConverter implementations when the conversion + /// of an object to/from a Message fails. + /// + /// Mark Pollack + public class MessageConversionException : EMSException + { + #region Constructor (s) / Destructor + + + /// + /// Creates a new instance of the IMessageConverterException class. with the specified message. + /// + /// + /// A message about the exception. + /// + public MessageConversionException(string message) + : base(message) + { + } + + /// + /// Creates a new instance of the IMessageConverterException class with the specified message + /// and root cause. + /// + /// + /// A message about the exception. + /// + /// + /// The root exception that is being wrapped. + /// + public MessageConversionException(string message, Exception rootCause) + : base(message) + { + LinkedException = rootCause; + } + + #endregion + } +} diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/SimpleMessageConverter.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/SimpleMessageConverter.cs new file mode 100644 index 00000000..27c692ff --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/SimpleMessageConverter.cs @@ -0,0 +1,240 @@ +#region License + +/* + * Copyright © 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 +{ + /// A simple message converter that can handle TextMessages, BytesMessages, + /// MapMessages, and ObjectMessages. Used as default by EmsTemplate, for + /// ConvertAndSend and ReceiveAndConvert operations. + /// + ///

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

+ /// + /// + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class SimpleMessageConverter : IMessageConverter + { + /// Convert a .NET object to a EMS Message using the supplied session + /// to create the message object. + /// + /// the object to convert + /// + /// the Session to use for creating a EMS Message + /// + /// the EMS Message + /// + /// EMSException if thrown by EMS API methods + /// MessageConversionException in case of conversion failure + 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"); + } + } + + /// Convert from a EMS Message to a .NET object. + /// the message to convert + /// + /// the converted .NET object + /// + /// MessageConversionException in case of conversion failure + 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 + + /// Create a EMS TextMessage for the given String. + /// the String to convert + /// + /// current EMS session + /// + /// the resulting message + /// + /// EMSException if thrown by EMS methods + protected virtual TextMessage CreateMessageForString(string text, Session session) + { + return session.CreateTextMessage((text)); + } + + /// Create a EMS BytesMessage for the given byte array. + /// the byyte array to convert + /// + /// current EMS session + /// + /// the resulting message + /// + /// EMSException if thrown by EMS methods + protected virtual BytesMessage CreateMessageForByteArray(byte[] bytes, Session session) + { + BytesMessage message = session.CreateBytesMessage(); + message.WriteBytes(bytes); + return message; + } + + /// Create a EMS MapMessage for the given Map. + /// the Map to convert + /// + /// current EMS session + /// + /// the resulting message + /// + /// EMSException if thrown by EMS methods + 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; + } + + /// Create a EMS ObjectMessage for the given Serializable object. + /// the Serializable object to convert + /// + /// current EMS session + /// + /// the resulting message + /// + /// EMSException if thrown by EMS methods + protected virtual ObjectMessage CreateMessageForSerializable( + ISerializable objectToSend, Session session) + { + return session.CreateObjectMessage(objectToSend); + } + + #endregion + + #region From Converter Mehtods + + /// Extract a String from the given TextMessage. + /// the message to convert + /// + /// the resulting String + /// + /// EMSException if thrown by EMS methods + protected virtual string ExtractStringFromMessage(TextMessage message) + { + return message.Text; + } + + /// Extract a byte array from the given BytesMessage. + /// the message to convert + /// + /// the resulting byte array + /// + /// EMSException if thrown by EMS methods + protected virtual byte[] ExtractByteArrayFromMessage(BytesMessage message) + { + byte[] bytes = new byte[(int)message.BodyLength]; + message.ReadBytes(bytes); + return bytes; + } + + /// Extract a IDictionary from the given MapMessage. + /// the message to convert + /// + /// the resulting Map + /// + /// EMSException if thrown by EMS methods + 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; + } + + /// + /// Extracts the serializable object from the given object message. + /// + /// The message to convert. + /// The resulting serializable object. + protected virtual object ExtractSerializableFromMessage( + ObjectMessage message) + { + return message.TheObject as ISerializable; + } + + #endregion + } +} diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/DynamicDestinationResolver.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/DynamicDestinationResolver.cs new file mode 100644 index 00000000..e70c3394 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/DynamicDestinationResolver.cs @@ -0,0 +1,86 @@ +#region License + +/* + * Copyright © 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 +{ + /// Simple DestinationResolver implementation resolving destination names + /// as dynamic destinations. + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class DynamicDestinationResolver : DestinationResolver + { + /// Resolve the given destination name, either as located resource + /// or as dynamic destination. + /// + /// the current EMS Session + /// + /// the name of the destination + /// + /// true if the domain is pub-sub, false if P2P + /// + /// the EMS destination (either a topic or a queue) + /// + /// EMSException if resolution failed + 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); + } + } + + + /// Resolve the given destination name to a Topic. + /// the current EMS Session + /// + /// the name of the desired Topic. + /// + /// the EMS Topic name + /// + /// EMSException if resolution failed + protected internal virtual Destination ResolveTopic(Session session, System.String topicName) + { + return session.CreateTopic(topicName); + } + + /// Resolve the given destination name to a Queue. + /// the current EMS Session + /// + /// the name of the desired Queue. + /// + /// the EMS Queue name + /// + /// EMSException if resolution failed + protected internal virtual Destination ResolveQueue(Session session, string queueName) + { + return session.CreateQueue(queueName); + } + } +} diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/EmsDestinationAccessor.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/EmsDestinationAccessor.cs new file mode 100644 index 00000000..45500292 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/EmsDestinationAccessor.cs @@ -0,0 +1,110 @@ +#region License + +/* + * Copyright © 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 +{ + /// Base class for EmsTemplate} and other + /// EMS-accessing gateway helpers, adding destination-related properties to + /// EmsAccessor's common properties. + /// + /// + ///

Not intended to be used directly. See EmsTemplate.

+ /// + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class EmsDestinationAccessor : EmsAccessor + { + #region Fields + + private DestinationResolver destinationResolver = new DynamicDestinationResolver(); + + private bool pubSubDomain = false; + + #endregion + + #region Properties + + /// + /// Gets or sets the destination resolver that is to be used to resolve + /// Destination references for this accessor. + /// + /// The default resolver is a DynamicDestinationResolver. Specify a + /// JndDestinationResolver for resolving destination names as JNDI locations. + /// + /// The destination resolver. + virtual public DestinationResolver DestinationResolver + { + get + { + return destinationResolver; + } + + set + { + AssertUtils.ArgumentNotNull(value, "DestinationResolver must not be null"); + this.destinationResolver = value; + } + + } + + + /// + /// Gets or sets a value indicating whether Publish/Subscribe + /// domain (Topics) is used. Otherwise, the Point-to-Point domain + /// (Queues) is used. + /// + /// + /// this + /// setting tells what type of destination to create if dynamic destinations are enabled. + /// true if Publish/Subscribe domain; otherwise, false + /// for the Point-to-Point domain. + public virtual bool PubSubDomain + { + get + { + return pubSubDomain; + } + + set + { + this.pubSubDomain = value; + } + + } + + #endregion + + /// + /// Resolves the given destination name to a EMS destination. + /// + /// The current session. + /// Name of the destination. + /// The located Destination + /// If resolution failed. + public virtual Destination ResolveDestinationName(Session session, System.String destinationName) + { + return DestinationResolver.ResolveDestinationName(session, destinationName, PubSubDomain); + } + } +} diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/IDestinationResolver.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/IDestinationResolver.cs new file mode 100644 index 00000000..f3b94207 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/IDestinationResolver.cs @@ -0,0 +1,58 @@ +#region License + +/* + * Copyright © 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 +{ + /// Strategy interface for resolving EMS destinations. + /// + /// + /// Used by EmsTemplate for resolving + /// destination names from simple Strings to actual + /// Destination implementation instances. + /// + /// + /// The default DestinationResolver implementation used by + /// EmsTemplate instances is the + /// DynamicDestinationResolver class. Consider using the + /// JndDestinationResolver for more advanced scenarios. + /// + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public interface DestinationResolver + { + /// Resolve the given destination name, either as located resource + /// or as dynamic destination. + /// + /// the current EMS Session + /// + /// the name of the destination + /// + /// true if the domain is pub-sub, false if P2P + /// + /// the EMS destination (either a topic or a queue) + /// + /// EMSException if resolution failed + Destination ResolveDestinationName(Session session, string destinationName, bool pubSubDomain); + + } +} diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsAccessor.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsAccessor.cs new file mode 100644 index 00000000..d80c5824 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsAccessor.cs @@ -0,0 +1,170 @@ +#region License + +/* + * Copyright © 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 +{ + /// Base class for EmsTemplate and other EMS-accessing gateway helpers + /// It defines common properties like the ConnectionFactory}. The subclass + /// EmsDestinationAccessor adds further, destination-related properties. + /// + /// Not intended to be used directly. See EmsTemplate. + /// + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + 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 + + + /// + /// Gets or sets the connection factory to use for obtaining EMS Connections. + /// + /// The connection factory. + virtual public ConnectionFactory ConnectionFactory + { + get + { + return connectionFactory; + } + + set + { + this.connectionFactory = value; + } + } + + + /// + /// Gets or sets the session acknowledge mode for EMS Sessions including whether or not the session is transacted + /// + /// + /// Set the EMS acknowledgement mode that is used when creating a EMS + /// Session to send a message. The default is AUTO_ACKNOWLEDGE. + /// + /// The session acknowledge mode. + virtual public int SessionAcknowledgeMode + { + get + { + return sessionAcknowledgeMode; + } + + set + { + this.sessionAcknowledgeMode = value; + } + + } + + /// + /// Set the transaction mode that is used when creating a EMS Session. + /// Default is "false". + /// + /// + /// 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. + /// + /// + public bool SessionTransacted + { + get + { + return sessionTransacted; + } + set + { + if (value) + { + sessionTransacted = value; + } + } + } + + #endregion + + + /// + /// Verify that ConnectionFactory property has been set. + /// + public virtual void AfterPropertiesSet() + { + if (ConnectionFactory == null) + { + throw new ArgumentException("ConnectionFactory is required"); + } + } + + /// + /// Creates the connection via the ConnectionFactory. + /// + /// + protected virtual Connection CreateConnection() + { + return ConnectionFactory.CreateConnection(); + } + + /// + /// Creates the session for the given Connection + /// + /// The connection to create a session for. + /// The new session + protected virtual Session CreateSession(Connection con) + { + return con.CreateSession(sessionTransacted, SessionAcknowledgeMode); + } + + /// + /// Returns whether the Session is in client acknowledgement mode. + /// + /// The session to check. + /// true if in client ack mode, false otherwise + protected virtual bool IsClientAcknowledge(Session session) + { + return (session.AcknowledgeMode == Session.CLIENT_ACKNOWLEDGE); + } + } +} diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsUtils.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsUtils.cs new file mode 100644 index 00000000..7ea6e203 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsUtils.cs @@ -0,0 +1,243 @@ +#region License + +/* + * Copyright © 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 +{ + /// + /// Generic utility methods for working with EMS. Mainly for internal use + /// within the framework, but also useful for custom EMS access code. + /// + public abstract class EmsUtils + { + #region Logging + + private static readonly ILog logger = LogManager.GetLogger(typeof(EmsUtils)); + + #endregion + + /// Close the given EMS Connection and ignore any thrown exception. + /// This is useful for typical finally blocks in manual EMS code. + /// + /// the EMS Connection to close (may be null) + /// + public static void CloseConnection(Connection con) + { + CloseConnection(con, false); + } + + /// Close the given EMS Connection and ignore any thrown exception. + /// This is useful for typical finally blocks in manual EMS code. + /// + /// the EMS Connection to close (may be null) + /// + /// whether to call stop() before closing + /// + 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); + } + } + } + + /// Close the given EMS Session and ignore any thrown exception. + /// This is useful for typical finally blocks in manual EMS code. + /// + /// the EMS Session to close (may be null) + /// + 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); + } + } + } + + /// Close the given EMS MessageProducer and ignore any thrown exception. + /// This is useful for typical finally blocks in manual EMS code. + /// + /// the EMS MessageProducer to close (may be null) + /// + 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); + } + } + } + + /// Close the given EMS MessageConsumer and ignore any thrown exception. + /// This is useful for typical finally blocks in manual EMS code. + /// + /// the EMS MessageConsumer to close (may be null) + /// + 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); + } + } + } +/* + /// Close the given EMS QueueRequestor and ignore any thrown exception. + /// This is useful for typical finally blocks in manual EMS code. + /// + /// the EMS QueueRequestor to close (may be null) + /// + */ +// 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); +// } +// } +// } + + + /// Commit the Session if not within a distributed transaction. + /// Needs investigation - no distributed tx in .NET messaging providers + /// the EMS Session to commit + /// + /// EMSException if committing failed + 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. +// } + } + + /// Rollback the Session if not within a distributed transaction. + /// Needs investigation - no distributed tx in EMS + /// the EMS Session to rollback + /// + /// EMSException if committing failed + 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. +// } + } + + } +}