diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Config/spring-nms-1.2.xsd b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Config/spring-nms-1.2.xsd index 3d5ac234..ee49fc2d 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Config/spring-nms-1.2.xsd +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Config/spring-nms-1.2.xsd @@ -43,7 +43,7 @@ diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedMessageConsumer .cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedMessageConsumer .cs new file mode 100644 index 00000000..d61c598f --- /dev/null +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedMessageConsumer .cs @@ -0,0 +1,115 @@ +#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 Apache.NMS; + +namespace Spring.Messaging.Nms.Connections +{ + /// + /// NMS MessageConsumer decorator that adapts all calls + /// to a shared MessageConsumer instance underneath. + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class CachedMessageConsumer : IMessageConsumer + { + private IMessageConsumer target; + + /// + /// Initializes a new instance of the class. + /// + /// The target. + public CachedMessageConsumer(IMessageConsumer target) + { + this.target = target; + } + + /// + /// Register for message events. + /// + public event MessageListener Listener + { + add + { + target.Listener += value; + } + remove + { + target.Listener -= value; + } + } + + /// + /// Receives the next message produced for this message consumer. + /// + /// the next message produced for this message consumer, , or null if this message consumer is concurrently closed + public IMessage Receive() + { + return this.target.Receive(); + } + + /// + /// Receives the next message that arrives within the specified timeout interval. + /// + /// The timeout value. + /// the next message produced for this message consumer, or null if the timeout expires or this message consumer is concurrently closed + public IMessage Receive(TimeSpan timeout) + { + return this.target.Receive(timeout); + } + + /// + /// Receives the next message if one is immediately available. + /// + /// the next message produced for this message consumer, or null if one is not available + public IMessage ReceiveNoWait() + { + return this.target.ReceiveNoWait(); + } + + /// + /// No-op implementation since it is caching. + /// + public void Close() + { + // It's a cached MessageConsumer... + } + + + /// + /// Dispose of wrapped MessageConsumer + /// + public void Dispose() + { + this.target.Dispose(); + } + + + /// + /// Description that shows this is a cached MessageConsumer + /// + /// Description that shows this is a cached MessageConsumer + public override string ToString() + { + return "Cached NMS MessageConsumer: " + this.target; + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedMessageProducer.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedMessageProducer.cs index 91b7e078..c4fbafa8 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedMessageProducer.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedMessageProducer.cs @@ -24,8 +24,8 @@ using Apache.NMS; namespace Spring.Messaging.Nms.Connections { /// - /// MessageProducer decorator that adapts specific settings - /// to a shared MessageProducer instance underneath. + /// MessageProducer decorator that adapts calls to a shared MessageProducer + /// instance underneath, managing QoS settings locally within the decorator. /// /// Juergen Hoeller /// Mark Pollack (.NET) @@ -33,17 +33,10 @@ namespace Spring.Messaging.Nms.Connections { private readonly IMessageProducer target; - private bool disableMessageID; - private object originalDisableMessageID = null; - private bool disableMessageTimestamp; - private object originalDisableMessageTimestamp = null; - //Not part of NMS spce - //private int deliveryMode; - private bool persistent; private byte priority; @@ -58,6 +51,9 @@ namespace Spring.Messaging.Nms.Connections public CachedMessageProducer(IMessageProducer target) { this.target = target; + this.persistent = target.Persistent; + this.priority = target.Priority; + this.timeToLive = target.TimeToLive; } @@ -220,15 +216,15 @@ namespace Spring.Messaging.Nms.Connections { get { - return disableMessageID; + return target.DisableMessageID; } set { if (originalDisableMessageID == null) { - originalDisableMessageID = value; + originalDisableMessageID = target.DisableMessageID; } - disableMessageID = value; + target.DisableMessageID = value; } } @@ -242,15 +238,15 @@ namespace Spring.Messaging.Nms.Connections { get { - return disableMessageTimestamp; + return target.DisableMessageTimestamp; } set { if (originalDisableMessageTimestamp == null) { - originalDisableMessageTimestamp = value; + originalDisableMessageTimestamp = target.DisableMessageTimestamp; } - disableMessageTimestamp = value; + target.DisableMessageTimestamp = value; } } @@ -271,5 +267,15 @@ namespace Spring.Messaging.Nms.Connections originalDisableMessageTimestamp = null; } } + + + /// + /// Returns string indicated this is a wrapped MessageProducer + /// + /// + public override string ToString() + { + return "Cached NMS MessageProducer: " + this.target; + } } } \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedSession.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedSession.cs index 8d57fb65..579577eb 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedSession.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedSession.cs @@ -22,6 +22,7 @@ using System.Collections; using Apache.NMS; using Common.Logging; using Spring.Collections; +using Spring.Util; using IQueue=Apache.NMS.IQueue; namespace Spring.Messaging.Nms.Connections @@ -46,22 +47,27 @@ namespace Spring.Messaging.Nms.Connections private LinkedList sessionList; private int sessionCacheSize; private IDictionary cachedProducers = new Hashtable(); + private IDictionary cachedConsumers = new Hashtable(); private IMessageProducer cachedUnspecifiedDestinationMessageProducer; private bool shouldCacheProducers; + private bool shouldCacheConsumers; + private bool transactionOpen = false; + private CachingConnectionFactory ccf; /// /// Initializes a new instance of the class. /// /// The target session. /// The session list. - /// Size of the session cache. - /// if set to true to cache message producers. - public CachedSession(ISession targetSession, LinkedList sessionList, int sessionCacheSize, bool cacheProducers) + /// The CachingConnectionFactory. + public CachedSession(ISession targetSession, LinkedList sessionList, CachingConnectionFactory ccf) { target = targetSession; this.sessionList = sessionList; - this.sessionCacheSize = sessionCacheSize; - shouldCacheProducers = cacheProducers; + this.sessionCacheSize = ccf.SessionCacheSize; + shouldCacheProducers = ccf.CacheProducers; + shouldCacheConsumers = ccf.CacheConsumers; + this.ccf = ccf; } @@ -95,16 +101,18 @@ namespace Spring.Messaging.Nms.Connections } else { - cachedUnspecifiedDestinationMessageProducer = target.CreateProducer(); #region Logging if (LOG.IsDebugEnabled) { - LOG.Debug("Created cached MessageProducer for unspecified destination"); + LOG.Debug("Creating cached MessageProducer for unspecified destination"); } #endregion + cachedUnspecifiedDestinationMessageProducer = target.CreateProducer(); + } + this.transactionOpen = true; return new CachedMessageProducer(cachedUnspecifiedDestinationMessageProducer); } else @@ -137,16 +145,18 @@ namespace Spring.Messaging.Nms.Connections else { producer = target.CreateProducer(destination); - cachedProducers.Add(destination, producer); #region Logging if (LOG.IsDebugEnabled) { - LOG.Debug("Created cached MessageProducer for destination [" + destination + "]"); + LOG.Debug("Creating cached MessageProducer for destination [" + destination + "]"); } #endregion + cachedProducers.Add(destination, producer); + } + this.transactionOpen = true; return new CachedMessageProducer(producer); } else @@ -161,37 +171,90 @@ namespace Spring.Messaging.Nms.Connections /// public void Close() { - lock (sessionList) + if (ccf.IsActive) { - if (sessionList.Count < sessionCacheSize) - { //don't pass the call to the underlying target. - if (!sessionList.Contains(this)) - { - sessionList.Add(this); //add to end of linked list. - #region Logging - if (LOG.IsDebugEnabled) - { - LOG.Debug("Returned cached Session: " + target); - } - #endregion - } - } - else + //don't pass the call to the underlying target. + lock (sessionList) { - foreach (DictionaryEntry entry in cachedProducers) + if (sessionList.Count < sessionCacheSize) { - ((IMessageProducer)entry.Value).Dispose(); - } - target.Close(); - if (LOG.IsDebugEnabled) - { - LOG.Debug("Closed cached Session: " + target); + LogicalClose(); + // Remain open in the session list. + return; } } } + // If we get here, we're supposed to shut down. + PhysicalClose(); + } + + private void LogicalClose() + { + // Preserve rollback-on-close semantics. + if (this.transactionOpen && this.target.Transacted) + { + this.transactionOpen = false; + this.target.Rollback(); + } + + // Physically close durable subscribers at time of Session close call. + IList ToRemove = new ArrayList(); + foreach (DictionaryEntry dictionaryEntry in cachedConsumers) + { + ConsumerCacheKey key = (ConsumerCacheKey) dictionaryEntry.Key; + if (key.Subscription != null) + { + ((IMessageConsumer) dictionaryEntry.Value).Close(); + ToRemove.Add(key); + } + } + foreach (ConsumerCacheKey key in ToRemove) + { + cachedConsumers.Remove(key); + } + + // Allow for multiple close calls... + if (!sessionList.Contains(this)) + { + #region Logging + + if (LOG.IsDebugEnabled) + { + LOG.Debug("Returning cached Session: " + target); + } + + #endregion + + sessionList.Add(this); //add to end of linked list. + } + } + + private void PhysicalClose() + { + if (LOG.IsDebugEnabled) + { + LOG.Debug("Closing cached Session: " + this.target); + } + // Explicitly close all MessageProducers and MessageConsumers that + // this Session happens to cache... + try + { + foreach (DictionaryEntry entry in cachedProducers) + { + ((IMessageProducer)entry.Value).Dispose(); + } + foreach (DictionaryEntry entry in cachedConsumers) + { + ((IMessageConsumer)entry.Value).Close(); + } + } + finally + { + // Now actually close the Session. + target.Close(); + } } - #region Pass through implementations /// /// Creates the consumer. /// @@ -199,7 +262,7 @@ namespace Spring.Messaging.Nms.Connections /// public IMessageConsumer CreateConsumer(IDestination destination) { - return target.CreateConsumer(destination); + return CreateConsumer(destination, null, false, null); } /// @@ -210,7 +273,7 @@ namespace Spring.Messaging.Nms.Connections /// public IMessageConsumer CreateConsumer(IDestination destination, string selector) { - return target.CreateConsumer(destination, selector); + return CreateConsumer(destination, selector, false, null); } /// @@ -222,23 +285,85 @@ namespace Spring.Messaging.Nms.Connections /// public IMessageConsumer CreateConsumer(IDestination destination, string selector, bool noLocal) { - return target.CreateConsumer(destination, selector, noLocal); + return CreateConsumer(destination, selector, noLocal, null); } - /// /// Creates the durable consumer. /// /// The destination. - /// The name. + /// The name of the durable subscription. /// The selector. /// if set to true [no local]. /// - public IMessageConsumer CreateDurableConsumer(ITopic destination, string name, string selector, bool noLocal) + public IMessageConsumer CreateDurableConsumer(ITopic destination, string subscription, string selector, bool noLocal) { - return target.CreateDurableConsumer(destination, name, selector, noLocal); + this.transactionOpen = true; + if (shouldCacheConsumers) + { + return GetCachedConsumer(destination, selector, noLocal, subscription); + } + else + { + return target.CreateDurableConsumer(destination, subscription, selector, noLocal); + } } + /// + /// Creates the consumer. + /// + /// The destination. + /// The selector. + /// if set to true [no local]. + /// The subscription. + /// + protected IMessageConsumer CreateConsumer(IDestination destination, string selector, bool noLocal, string subscription) + { + this.transactionOpen = true; + if (shouldCacheConsumers) + { + return GetCachedConsumer(destination, selector, noLocal, subscription); + } + else + { + return target.CreateConsumer(destination, selector, noLocal); + } + } + + private IMessageConsumer GetCachedConsumer(IDestination destination, string selector, bool noLocal, string subscription) + { + object cacheKey = new ConsumerCacheKey(destination, selector, noLocal, null); + IMessageConsumer consumer = (IMessageConsumer)cachedConsumers[cacheKey]; + if (consumer != null) + { + if (LOG.IsDebugEnabled) + { + LOG.Debug("Found cached NMS MessageConsumer for destination [" + destination + "]: " + consumer); + } + } + else + { + if (destination is ITopic) + { + consumer = (subscription != null + ? target.CreateDurableConsumer((ITopic)destination, subscription, selector, noLocal) + : target.CreateConsumer(destination, selector, noLocal)); + } + else + { + return target.CreateConsumer(destination, selector); + } + if (LOG.IsDebugEnabled) + { + LOG.Debug("Creating cached NMS MessageConsumer for destination [" + destination + "]: " + consumer); + } + cachedConsumers[cacheKey] = consumer; + } + return new CachedMessageConsumer(consumer); + } + + #region Pass through implementations + /// /// Gets the queue. /// @@ -246,6 +371,7 @@ namespace Spring.Messaging.Nms.Connections /// public IQueue GetQueue(string name) { + this.transactionOpen = true; return target.GetQueue(name); } @@ -256,6 +382,7 @@ namespace Spring.Messaging.Nms.Connections /// public ITopic GetTopic(string name) { + this.transactionOpen = true; return target.GetTopic(name); } @@ -265,6 +392,7 @@ namespace Spring.Messaging.Nms.Connections /// public ITemporaryQueue CreateTemporaryQueue() { + this.transactionOpen = true; return target.CreateTemporaryQueue(); } @@ -274,6 +402,7 @@ namespace Spring.Messaging.Nms.Connections /// public ITemporaryTopic CreateTemporaryTopic() { + this.transactionOpen = true; return target.CreateTemporaryTopic(); } @@ -283,6 +412,7 @@ namespace Spring.Messaging.Nms.Connections /// public IMessage CreateMessage() { + this.transactionOpen = true; return target.CreateMessage(); } @@ -292,6 +422,7 @@ namespace Spring.Messaging.Nms.Connections /// public ITextMessage CreateTextMessage() { + this.transactionOpen = true; return target.CreateTextMessage(); } @@ -302,6 +433,7 @@ namespace Spring.Messaging.Nms.Connections /// public ITextMessage CreateTextMessage(string text) { + this.transactionOpen = true; return target.CreateTextMessage(text); } @@ -311,6 +443,7 @@ namespace Spring.Messaging.Nms.Connections /// public IMapMessage CreateMapMessage() { + this.transactionOpen = true; return target.CreateMapMessage(); } @@ -321,6 +454,7 @@ namespace Spring.Messaging.Nms.Connections /// public IObjectMessage CreateObjectMessage(object body) { + this.transactionOpen = true; return target.CreateObjectMessage(body); } @@ -330,6 +464,7 @@ namespace Spring.Messaging.Nms.Connections /// public IBytesMessage CreateBytesMessage() { + this.transactionOpen = true; return target.CreateBytesMessage(); } @@ -340,6 +475,7 @@ namespace Spring.Messaging.Nms.Connections /// public IBytesMessage CreateBytesMessage(byte[] body) { + this.transactionOpen = true; return target.CreateBytesMessage(body); } @@ -348,6 +484,7 @@ namespace Spring.Messaging.Nms.Connections /// public void Commit() { + this.transactionOpen = false; target.Commit(); } @@ -356,6 +493,7 @@ namespace Spring.Messaging.Nms.Connections /// public void Rollback() { + this.transactionOpen = false; target.Rollback(); } @@ -365,7 +503,11 @@ namespace Spring.Messaging.Nms.Connections /// true if transacted; otherwise, false. public bool Transacted { - get { return target.Transacted; } + get + { + this.transactionOpen = true; + return target.Transacted; + } } /// @@ -374,7 +516,11 @@ namespace Spring.Messaging.Nms.Connections /// The acknowledgement mode. public AcknowledgementMode AcknowledgementMode { - get { return target.AcknowledgementMode; } + get + { + this.transactionOpen = true; + return target.AcknowledgementMode; + } } /// @@ -382,8 +528,62 @@ namespace Spring.Messaging.Nms.Connections /// public void Dispose() { + this.transactionOpen = true; target.Dispose(); } #endregion + + /// + /// Returns a that represents the current . + /// + /// + /// A that represents the current . + /// + public override string ToString() + { + return "Cached NMS Session: " + this.target; + } + } + + internal class ConsumerCacheKey + { + private IDestination destination; + private string selector; + private bool noLocal; + private string subscription; + + public ConsumerCacheKey(IDestination destination, string selector, bool noLocal, string subscription) + { + this.destination = destination; + this.selector = selector; + this.noLocal = noLocal; + this.subscription = subscription; + } + + public string Subscription + { + get { return subscription; } + } + + protected bool Equals(ConsumerCacheKey consumerCacheKey) + { + if (consumerCacheKey == null) return false; + if (!Equals(destination, consumerCacheKey.destination)) return false; + if (!ObjectUtils.NullSafeEquals(selector, consumerCacheKey.selector)) return false; + if (!Equals(noLocal, consumerCacheKey.noLocal)) return false; + if (!ObjectUtils.NullSafeEquals(subscription, consumerCacheKey.subscription)) return false; + return true; + } + + public override bool Equals(object obj) + { + if (ReferenceEquals(this, obj)) return true; + return Equals(obj as ConsumerCacheKey); + } + + public override int GetHashCode() + { + return destination.GetHashCode(); + } } } \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachingConnectionFactory.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachingConnectionFactory.cs index 6df52d50..de297899 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachingConnectionFactory.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachingConnectionFactory.cs @@ -18,6 +18,7 @@ #endregion +using System; using System.Collections; using Apache.NMS; using Common.Logging; @@ -38,7 +39,22 @@ namespace Spring.Messaging.Nms.Connections /// By default, only one single Session will be cached, with further requested /// Sessions being created and disposed on demand. Consider raising the /// SessionCacheSize property in case of a high-concurrency environment. + /// + /// NOTE: This ConnectionFactory requires explicit closing of all Sessions + /// obtained from its shared Connection. This is the usual recommendation for + /// native NMS access code anyway. However, with this ConnectionFactory, its use + /// is mandatory in order to actually allow for Session reuse. + /// + /// + /// Note also that MessageConsumers obtained from a cached Session won't get + /// closed until the Session will eventually be removed from the pool. This may + /// lead to semantic side effects in some cases. For a durable subscriber, the + /// logical Session.Close() call will also close the subscription. + /// Re-registering a durable consumer for the same subscription on the same + /// Session handle is not supported; close and reobtain a cached Session first. + /// /// + /// /// Juergen Hoeller /// Mark Pollack (.NET) public class CachingConnectionFactory : SingleConnectionFactory @@ -53,6 +69,10 @@ namespace Spring.Messaging.Nms.Connections private bool cacheProducers = true; + private bool cacheConsumers = true; + + private volatile bool active = true; + private IDictionary cachedSessions = new Hashtable(); @@ -65,6 +85,16 @@ namespace Spring.Messaging.Nms.Connections ReconnectOnException = true; } + /// + /// Initializes a new instance of the class for the given + /// IConnectionFactory + /// + /// The target connection factory. + public CachingConnectionFactory(IConnectionFactory targetConnectionFactory) : base(targetConnectionFactory) + { + ReconnectOnException = true; + } + /// /// Gets or sets the size of the session cache. @@ -99,8 +129,8 @@ namespace Spring.Messaging.Nms.Connections /// and Session). /// /// - /// Default is "true". Switch this to "false" in order to recreate, - /// MessageProducers on demand. + /// Default is "true". Switch this to "false" in order to always, + /// recreate MessageProducers on demand. /// /// /// true if should cache message producers; otherwise, false. @@ -110,15 +140,67 @@ namespace Spring.Messaging.Nms.Connections set { cacheProducers = value; } } + + /// + /// Gets or sets a value indicating whether o cache JMS MessageConsumers per + /// NMS Session instance. + /// + /// + /// Mmore specifically: one MessageConsumer per Destination, selector String + /// and Session. Note that durable subscribers will only be cached until + /// logical closing of the Session handle. + /// + /// Default is "true". Switch this to "false" in order to always + /// recreate MessageConsumers on demand. + /// + /// + /// true to cache consumers per session instance; otherwise, false. + public bool CacheConsumers + { + get { return cacheConsumers; } + set { cacheConsumers = value; } + } + + /// + /// Gets or sets a value indicating whether this instance is active. + /// + /// true if this instance is active; otherwise, false. + public bool IsActive + { + get { return active; } + set { active = value; } + } + /// /// Resets the Session cache as well as resetting the connection. /// public override void ResetConnection() { + this.active = false; lock (cachedSessions) { + foreach (DictionaryEntry dictionaryEntry in cachedSessions) + { + LinkedList sessionList = (LinkedList) dictionaryEntry.Value; + lock (sessionList) + { + foreach (ISession session in sessionList) + { + try + { + session.Close(); + } + catch (Exception ex) + { + LOG.Trace("Could not close cached NMS Session", ex); + } + } + } + } cachedSessions.Clear(); } + this.active = true; + // Now proceed with actual closing of the shared Connection... base.ResetConnection(); } @@ -155,16 +237,17 @@ namespace Spring.Messaging.Nms.Connections { if (LOG.IsDebugEnabled) { - LOG.Debug("Found cached Session for mode " + mode + ": " + session); + LOG.Debug("Found cached Session for mode " + mode + ": " + + (session is IDecoratorSession ? ((IDecoratorSession) session).TargetSession : session)); } } else { - ISession targetSession = con.CreateSession(mode); - session = GetCachedSessionWrapper(targetSession, sessionList); + ISession targetSession = con.CreateSession(mode); if (LOG.IsDebugEnabled) { - LOG.Debug("Created cached Session for mode " + mode + ": " + session); + LOG.Debug("Creating cached Session for mode " + mode + ": " + targetSession); } + session = GetCachedSessionWrapper(targetSession, sessionList); } return session; } @@ -179,7 +262,7 @@ namespace Spring.Messaging.Nms.Connections /// The wrapped Session protected virtual ISession GetCachedSessionWrapper(ISession targetSession, LinkedList sessionList) { - return new CachedSession(targetSession, sessionList, SessionCacheSize, CacheProducers); + return new CachedSession(targetSession, sessionList, this); } } diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/ConnectionFactoryUtils.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/ConnectionFactoryUtils.cs index a1b635bb..81c3a3b0 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/ConnectionFactoryUtils.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/ConnectionFactoryUtils.cs @@ -78,6 +78,23 @@ namespace Spring.Messaging.Nms.Connections } } + /// + /// Return the innermost target Session of the given Session. If the given + /// Session is a decorated session, it will be unwrapped until a non-decorated + /// Session is found. Otherwise, the passed-in Session will be returned as-is. + /// + /// The session to unwrap + /// The innermost target Session, or the passed-in one if no decorator + public static ISession GetTargetSession(ISession session) + { + ISession sessionToUse = session; + while (sessionToUse is IDecoratorSession) + { + sessionToUse = ((IDecoratorSession) sessionToUse).TargetSession; + } + return sessionToUse; + } + /// /// Determines whether the given JMS Session is transactional, that is, /// bound to the current thread by Spring's transaction facilities. @@ -214,7 +231,7 @@ namespace Spring.Messaging.Nms.Connections if (resourceHolderToUse != resourceHolder) { TransactionSynchronizationManager.RegisterSynchronization( - new MessageResourceSynchronization(resourceKey, resourceHolderToUse, + new NmsResourceSynchronization(resourceHolderToUse, resourceKey, resourceFactory.SynchedLocalTransactionAllowed)); resourceHolderToUse.SynchronizedWithTransaction = true; TransactionSynchronizationManager.BindResource(resourceKey, resourceHolderToUse); @@ -329,7 +346,7 @@ namespace Spring.Messaging.Nms.Connections /// Callback for resource cleanup at the end of a non-native NMS transaction /// - private class MessageResourceSynchronization : TransactionSynchronizationAdapter + private class NmsResourceSynchronization : TransactionSynchronizationAdapter { private object resourceKey; @@ -339,7 +356,7 @@ namespace Spring.Messaging.Nms.Connections private bool holderActive = true; - public MessageResourceSynchronization(object resourceKey, NmsResourceHolder resourceHolder, bool transacted) + public NmsResourceSynchronization(NmsResourceHolder resourceHolder, object resourceKey, bool transacted) { this.resourceKey = resourceKey; this.resourceHolder = resourceHolder; diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/IDecoratorSession.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/IDecoratorSession.cs index 87f3738e..02a5bf20 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/IDecoratorSession.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/IDecoratorSession.cs @@ -28,6 +28,8 @@ namespace Spring.Messaging.Nms.Connections /// functionality. Allows access to the the underlying target Session. /// /// Mark Pollack + /// + /// public interface IDecoratorSession : ISession { /// diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsResourceHolder.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsResourceHolder.cs index 66b1329e..eaa91e05 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsResourceHolder.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsResourceHolder.cs @@ -273,6 +273,9 @@ namespace Spring.Messaging.Nms.Connections { ConnectionFactoryUtils.ReleaseConnection(connection, connectionFactory, true); } + this.connections.Clear(); + this.sessions.Clear(); + this.sessionsPerIConnection.Clear(); } /// diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsTransactionManager.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsTransactionManager.cs index 4cb6c81f..d5a5cdbc 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsTransactionManager.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsTransactionManager.cs @@ -44,11 +44,21 @@ namespace Spring.Messaging.Nms.Connections /// automatically participate in it. /// /// - /// This transaction strategy will typically be used in combination with - /// , which uses a single NMS Connection - /// for all NMS access in order to avoid the overhead of repeated Connection - /// creation. Each transaction will then share the same NMS Connection, while still using - /// its own individual Session. + /// The use of as a target for this + /// transaction manager is strongly recommended. CachingConnectionFactory + /// uses a single NMS Connection for all NMS access in order to avoid the overhead + /// of repeated Connection creation, as well as maintaining a cache of Sessions. + /// Each transaction will then share the same NMS Connection, while still using + /// its own individual NMS Session. + /// + /// The use of a raw target ConnectionFactory would not only be inefficient + /// because of the lack of resource reuse. It might also lead to strange effects + /// when your NMS provider doesn't accept MessageProducer.close() calls + /// and/or MessageConsumer.close() calls before Session.commit(), + /// with the latter supposed to commit all the messages that have been sent through the + /// producer handle and received through the consumer handle. As a safe general solution, + /// always pass in a into this transaction manager's + /// ConnectionFactory property. /// /// /// Transaction synchronization is turned off by default, as this manager might be used diff --git a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/SingleConnectionFactory.cs b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/SingleConnectionFactory.cs index 130de362..255b23b6 100644 --- a/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/SingleConnectionFactory.cs +++ b/src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/SingleConnectionFactory.cs @@ -31,8 +31,8 @@ namespace Spring.Messaging.Nms.Connections /// A ConnectionFactory adapter that returns the same Connection /// from all CreateConnection() calls, and ignores calls to /// Connection.Close(). According to the JMS Connection - /// model, this is perfectly thread-safe, check your vendor implmenetation for - /// details. + /// model, this is perfectly thread-safe. The + /// shared Connection can be automatically recovered in case of an Exception. /// /// /// @@ -83,6 +83,12 @@ namespace Spring.Messaging.Nms.Connections /// private IConnection connection; + + /// + /// Whether the shared Connection has been started + /// + private bool started = false; + /// /// Synchronization monitor for the shared Connection /// @@ -193,6 +199,27 @@ namespace Spring.Messaging.Nms.Connections set { reconnectOnException = value; } } + /// + /// Gets the connection monitor. + /// + /// The connection monitor. + internal object ConnectionMonitor + { + get { return connectionMonitor; } + } + + /// + /// Gets a value indicating whether this instance is started. + /// + /// + /// true if this instance is started; otherwise, false. + /// + internal bool IsStarted + { + get { return started;} + set { started = value; } + } + #endregion #region IConnectionFactory Members @@ -292,7 +319,7 @@ namespace Spring.Messaging.Nms.Connections /// The connection to operate on. /// The session ack mode. /// the Session to use, or null to indicate - /// creation of a default Session + /// creation of a raw standard Session public virtual ISession GetSession(IConnection con, AcknowledgementMode mode) { return null; @@ -313,11 +340,19 @@ namespace Spring.Messaging.Nms.Connections /// The connection. protected virtual void CloseConnection(IConnection con) { + if (LOG.IsDebugEnabled) + { + LOG.Debug("Closing shared NMS Connection: " + this.target); + } try { try { - con.Stop(); + if (this.started) + { + this.started = false; + con.Stop(); + } } finally { con.Close(); @@ -424,6 +459,16 @@ namespace Spring.Messaging.Nms.Connections // don't pass the call to the target. } + public void Start() + { + // Handle start method: track started state. + target.Start(); + lock (singleConnectionFactory.ConnectionMonitor) + { + singleConnectionFactory.IsStarted = true; + } + } + public void Stop() { //don't pass the call to the target as it would stop receiving for all clients sharing this connection. @@ -471,10 +516,6 @@ namespace Spring.Messaging.Nms.Connections target.Dispose(); } - public void Start() - { - target.Start(); - } public bool IsStarted { @@ -482,5 +523,13 @@ namespace Spring.Messaging.Nms.Connections } #endregion + /// + /// Add information to show this is a shared NMS connection + /// + /// Description of connection wrapper + public override string ToString() + { + return "Shared NMS Connection: " + this.target; + } } } \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Nms/Spring.Messaging.Nms.2005.csproj b/src/Spring/Spring.Messaging.Nms/Spring.Messaging.Nms.2005.csproj index 72d29e27..1f3bcc71 100644 --- a/src/Spring/Spring.Messaging.Nms/Spring.Messaging.Nms.2005.csproj +++ b/src/Spring/Spring.Messaging.Nms/Spring.Messaging.Nms.2005.csproj @@ -47,6 +47,7 @@ + diff --git a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/CachingConnectionFactoryTests.cs b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/CachingConnectionFactoryTests.cs index 7a5c5da1..7c87ae87 100644 --- a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/CachingConnectionFactoryTests.cs +++ b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/CachingConnectionFactoryTests.cs @@ -32,7 +32,6 @@ namespace Spring.Messaging.Nms.Connections /// This class contains tests for /// /// Mark Pollack - /// $Id:$ [TestFixture] public class CachingConnectionFactoryTests { diff --git a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/SingleConnectionFactoryTests.cs b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/SingleConnectionFactoryTests.cs index 7fb0e704..f0bdd898 100644 --- a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/SingleConnectionFactoryTests.cs +++ b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/SingleConnectionFactoryTests.cs @@ -232,5 +232,64 @@ namespace Spring.Messaging.Nms.Connections Assert.AreEqual(2, con.CloseCount); Assert.AreEqual(1, listener.Count); } + + [Test] + public void CachingConnectionFactory() + { + IConnectionFactory connectionFactory = (IConnectionFactory)mocks.CreateMock(typeof(IConnectionFactory)); + IConnection connection = (IConnection)mocks.CreateMock(typeof(IConnection)); + ISession txSession = (ISession)mocks.CreateMock(typeof(ISession)); + ISession nonTxSession = (ISession)mocks.CreateMock(typeof(ISession)); + Expect.Call(connectionFactory.CreateConnection()).Return(connection).Repeat.Once(); + + Expect.Call(connection.CreateSession(AcknowledgementMode.Transactional)).Return(txSession).Repeat.Once(); + Expect.Call(txSession.Transacted).Return(true).Repeat.Twice(); + txSession.Rollback(); + LastCall.Repeat.Once(); + txSession.Commit(); + LastCall.Repeat.Once(); + txSession.Close(); + LastCall.Repeat.Once(); + + Expect.Call(connection.CreateSession(AcknowledgementMode.ClientAcknowledge)).Return(nonTxSession).Repeat.Once(); + nonTxSession.Close(); + LastCall.Repeat.Once(); + connection.Start(); + LastCall.Repeat.Twice(); + connection.Stop(); + LastCall.Repeat.Once(); + connection.Close(); + LastCall.Repeat.Once(); + + mocks.ReplayAll(); + + CachingConnectionFactory scf = new CachingConnectionFactory(connectionFactory); + scf.ReconnectOnException = false; + + IConnection con1 = scf.CreateConnection(); + ISession session1 = con1.CreateSession(AcknowledgementMode.Transactional); + bool b = session1.Transacted; + session1.Close(); // should be ignored + session1 = con1.CreateSession(AcknowledgementMode.ClientAcknowledge); + session1.Close(); // should be ignored + con1.Start(); + con1.Close(); // should be ignored + IConnection con2 = scf.CreateConnection(); + ISession session2 = con2.CreateSession(AcknowledgementMode.ClientAcknowledge); + session2.Close(); // should be ignored + session2 = con2.CreateSession(AcknowledgementMode.Transactional); + session2.Commit(); + session2.Close(); // should be ignored + con2.Start(); + con2.Close(); + scf.Dispose(); + + mocks.Verify(connectionFactory); + mocks.Verify(connection); + mocks.Verify(txSession); + mocks.Verify(nonTxSession); + + + } } } \ No newline at end of file diff --git a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestMessageProducer.cs b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestMessageProducer.cs index b1a5d755..b26cac7e 100644 --- a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestMessageProducer.cs +++ b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestMessageProducer.cs @@ -26,6 +26,9 @@ namespace Spring.Messaging.Nms.Connections public class TestMessageProducer : IMessageProducer { + private bool persistent; + private TimeSpan timeToLive = new TimeSpan(0,0,0,60,0); + public void Send(IMessage message) { throw new NotImplementedException(); @@ -83,20 +86,20 @@ namespace Spring.Messaging.Nms.Connections public bool Persistent { - get { throw new NotImplementedException(); } - set { throw new NotImplementedException(); } + get { return persistent; } + set { persistent = value; } } public TimeSpan TimeToLive { - get { throw new NotImplementedException(); } - set { throw new NotImplementedException(); } + get { return timeToLive; } + set { timeToLive = value; } } public byte Priority { - get { throw new NotImplementedException(); } - set { throw new NotImplementedException(); } + get { return 1; } + set { } } public bool DisableMessageID diff --git a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestSession.cs b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestSession.cs index 82c6c401..0322d8f9 100644 --- a/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestSession.cs +++ b/test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestSession.cs @@ -143,17 +143,17 @@ namespace Spring.Messaging.Nms.Connections public void Commit() { - throw new NotImplementedException(); + } public void Rollback() { - throw new NotImplementedException(); + } public bool Transacted { - get { throw new NotImplementedException(); } + get { return true; } } public AcknowledgementMode AcknowledgementMode