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

Sync to trunk implementations in Java (post 2.5.6 atm)

Browse Source 
CachedMessageProducer initializes its deliveryMode, priority and timeToLive fields with the target's settings 
NmsResourceHolder remove all Connections and Sessions after closing 
CachingConnectionFactory rolls back cached transacted JMS Sessions on logical close if no commit/rollback was issued
CachingConnectionFactory explicitly closes cached JMS Sessions on Connection close
CachingConnectionFactory consistently rollback and physically close Sessions on close() 
CachingConnectionFactory also caches MessageConsumers (controlled through "cache…
CachedMessageProducers pass on "disableMessageID" and "disableMessageTimestamp" properties 
SingleConnectionFactory only calls "Connection.Stop()" on reset when the shared Connection has actually been started
This commit is contained in:
markpollack
2008-08-12 19:09:13 +00:00
parent 63aebbf99c
commit 06e1dfdbe0
15 changed files with 637 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Config/spring-nms-1.2.xsd
View File

@@ -43,7 +43,7 @@
<xsd:attribute name="connection-factory" type="xsd:string" default="ConnectionFactory">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to the NMS ConnectionFactory bean.
A reference to the NMS ConnectionFactory object.
Default is "ConnectionFactory".
]]></xsd:documentation>
</xsd:annotation>

115
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedMessageConsumer .cs Normal file
View File

@@ -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
{
/// <summary>
/// NMS MessageConsumer decorator that adapts all calls
/// to a shared MessageConsumer instance underneath.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET) </author>
public class CachedMessageConsumer : IMessageConsumer
{
private IMessageConsumer target;
/// <summary>
/// Initializes a new instance of the <see cref="CachedMessageConsumer"/> class.
/// </summary>
/// <param name="target">The target.</param>
public CachedMessageConsumer(IMessageConsumer target)
{
this.target = target;
}
/// <summary>
/// Register for message events.
/// </summary>
public event MessageListener Listener
{
add
{
target.Listener += value;
}
remove
{
target.Listener -= value;
}
}
/// <summary>
/// Receives the next message produced for this message consumer.
/// </summary>
/// <returns>the next message produced for this message consumer, , or null if this message consumer is concurrently closed</returns>
public IMessage Receive()
{
return this.target.Receive();
}
/// <summary>
/// Receives the next message that arrives within the specified timeout interval.
/// </summary>
/// <param name="timeout">The timeout value.</param>
/// <returns>the next message produced for this message consumer, or null if the timeout expires or this message consumer is concurrently closed</returns>
public IMessage Receive(TimeSpan timeout)
{
return this.target.Receive(timeout);
}
/// <summary>
/// Receives the next message if one is immediately available.
/// </summary>
/// <returns>the next message produced for this message consumer, or null if one is not available</returns>
public IMessage ReceiveNoWait()
{
return this.target.ReceiveNoWait();
}
/// <summary>
/// No-op implementation since it is caching.
/// </summary>
public void Close()
{
// It's a cached MessageConsumer...
}
/// <summary>
/// Dispose of wrapped MessageConsumer
/// </summary>
public void Dispose()
{
this.target.Dispose();
}
/// <summary>
/// Description that shows this is a cached MessageConsumer
/// </summary>
/// <returns>Description that shows this is a cached MessageConsumer</returns>
public override string ToString()
{
return "Cached NMS MessageConsumer: " + this.target;
}
}
}

36
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedMessageProducer.cs
View File

@@ -24,8 +24,8 @@ using Apache.NMS;
namespace Spring.Messaging.Nms.Connections
{
/// <summary>
/// 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.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
@@ -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;
}
}
/// <summary>
/// Returns string indicated this is a wrapped MessageProducer
/// </summary>
/// <returns></returns>
public override string ToString()
{
return "Cached NMS MessageProducer: " + this.target;
}
}
}

282
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedSession.cs
View File

@@ -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;
/// <summary>
/// Initializes a new instance of the <see cref="CachedSession"/> class.
/// </summary>
/// <param name="targetSession">The target session.</param>
/// <param name="sessionList">The session list.</param>
/// <param name="sessionCacheSize">Size of the session cache.</param>
/// <param name="cacheProducers">if set to <c>true</c> to cache message producers.</param>
public CachedSession(ISession targetSession, LinkedList sessionList, int sessionCacheSize, bool cacheProducers)
/// <param name="ccf">The CachingConnectionFactory.</param>
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
/// </summary>
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
/// <summary>
/// Creates the consumer.
/// </summary>
@@ -199,7 +262,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public IMessageConsumer CreateConsumer(IDestination destination)
{
return target.CreateConsumer(destination);
return CreateConsumer(destination, null, false, null);
}
/// <summary>
@@ -210,7 +273,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public IMessageConsumer CreateConsumer(IDestination destination, string selector)
{
return target.CreateConsumer(destination, selector);
return CreateConsumer(destination, selector, false, null);
}
/// <summary>
@@ -222,23 +285,85 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public IMessageConsumer CreateConsumer(IDestination destination, string selector, bool noLocal)
{
return target.CreateConsumer(destination, selector, noLocal);
return CreateConsumer(destination, selector, noLocal, null);
}
/// <summary>
/// Creates the durable consumer.
/// </summary>
/// <param name="destination">The destination.</param>
/// <param name="name">The name.</param>
/// <param name="subscription">The name of the durable subscription.</param>
/// <param name="selector">The selector.</param>
/// <param name="noLocal">if set to <c>true</c> [no local].</param>
/// <returns></returns>
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);
}
}
/// <summary>
/// Creates the consumer.
/// </summary>
/// <param name="destination">The destination.</param>
/// <param name="selector">The selector.</param>
/// <param name="noLocal">if set to <c>true</c> [no local].</param>
/// <param name="subscription">The subscription.</param>
/// <returns></returns>
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
/// <summary>
/// Gets the queue.
/// </summary>
@@ -246,6 +371,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public IQueue GetQueue(string name)
{
this.transactionOpen = true;
return target.GetQueue(name);
}
@@ -256,6 +382,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public ITopic GetTopic(string name)
{
this.transactionOpen = true;
return target.GetTopic(name);
}
@@ -265,6 +392,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public ITemporaryQueue CreateTemporaryQueue()
{
this.transactionOpen = true;
return target.CreateTemporaryQueue();
}
@@ -274,6 +402,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public ITemporaryTopic CreateTemporaryTopic()
{
this.transactionOpen = true;
return target.CreateTemporaryTopic();
}
@@ -283,6 +412,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public IMessage CreateMessage()
{
this.transactionOpen = true;
return target.CreateMessage();
}
@@ -292,6 +422,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public ITextMessage CreateTextMessage()
{
this.transactionOpen = true;
return target.CreateTextMessage();
}
@@ -302,6 +433,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public ITextMessage CreateTextMessage(string text)
{
this.transactionOpen = true;
return target.CreateTextMessage(text);
}
@@ -311,6 +443,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public IMapMessage CreateMapMessage()
{
this.transactionOpen = true;
return target.CreateMapMessage();
}
@@ -321,6 +454,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public IObjectMessage CreateObjectMessage(object body)
{
this.transactionOpen = true;
return target.CreateObjectMessage(body);
}
@@ -330,6 +464,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public IBytesMessage CreateBytesMessage()
{
this.transactionOpen = true;
return target.CreateBytesMessage();
}
@@ -340,6 +475,7 @@ namespace Spring.Messaging.Nms.Connections
/// <returns></returns>
public IBytesMessage CreateBytesMessage(byte[] body)
{
this.transactionOpen = true;
return target.CreateBytesMessage(body);
}
@@ -348,6 +484,7 @@ namespace Spring.Messaging.Nms.Connections
/// </summary>
public void Commit()
{
this.transactionOpen = false;
target.Commit();
}
@@ -356,6 +493,7 @@ namespace Spring.Messaging.Nms.Connections
/// </summary>
public void Rollback()
{
this.transactionOpen = false;
target.Rollback();
}
@@ -365,7 +503,11 @@ namespace Spring.Messaging.Nms.Connections
/// <value><c>true</c> if transacted; otherwise, <c>false</c>.</value>
public bool Transacted
{
get { return target.Transacted; }
get
{
this.transactionOpen = true;
return target.Transacted;
}
}
/// <summary>
@@ -374,7 +516,11 @@ namespace Spring.Messaging.Nms.Connections
/// <value>The acknowledgement mode.</value>
public AcknowledgementMode AcknowledgementMode
{
get { return target.AcknowledgementMode; }
get
{
this.transactionOpen = true;
return target.AcknowledgementMode;
}
}
/// <summary>
@@ -382,8 +528,62 @@ namespace Spring.Messaging.Nms.Connections
/// </summary>
public void Dispose()
{
this.transactionOpen = true;
target.Dispose();
}
#endregion
/// <summary>
/// Returns a <see cref="T:System.String"/> that represents the current <see cref="T:System.Object"/>.
/// </summary>
/// <returns>
/// A <see cref="T:System.String"/> that represents the current <see cref="T:System.Object"/>.
/// </returns>
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();
}
}
}

97
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachingConnectionFactory.cs
View File

@@ -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.
/// <para>
/// 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.
/// </para>
/// <para>
/// 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 <code>Session.Close()</code> 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.
/// </para>
/// </remarks>
///
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
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;
}
/// <summary>
/// Initializes a new instance of the <see cref="CachingConnectionFactory"/> class for the given
/// IConnectionFactory
/// </summary>
/// <param name="targetConnectionFactory">The target connection factory.</param>
public CachingConnectionFactory(IConnectionFactory targetConnectionFactory) : base(targetConnectionFactory)
{
ReconnectOnException = true;
}
/// <summary>
/// Gets or sets the size of the session cache.
@@ -99,8 +129,8 @@ namespace Spring.Messaging.Nms.Connections
/// and Session).
/// </summary>
/// <remarks>
/// <para>Default is "true". Switch this to "false" in order to recreate,
/// MessageProducers on demand.
/// <para>Default is "true". Switch this to "false" in order to always,
/// recreate MessageProducers on demand.
/// </para>
/// </remarks>
/// <value><c>true</c> if should cache message producers; otherwise, <c>false</c>.</value>
@@ -110,15 +140,67 @@ namespace Spring.Messaging.Nms.Connections
set { cacheProducers = value; }
}
/// <summary>
/// Gets or sets a value indicating whether o cache JMS MessageConsumers per
/// NMS Session instance.
/// </summary>
/// <remarks>
/// 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.
/// <para>
/// Default is "true". Switch this to "false" in order to always
/// recreate MessageConsumers on demand.
/// </para>
/// </remarks>
/// <value><c>true</c> to cache consumers per session instance; otherwise, <c>false</c>.</value>
public bool CacheConsumers
{
get { return cacheConsumers; }
set { cacheConsumers = value; }
}
/// <summary>
/// Gets or sets a value indicating whether this instance is active.
/// </summary>
/// <value><c>true</c> if this instance is active; otherwise, <c>false</c>.</value>
public bool IsActive
{
get { return active; }
set { active = value; }
}
/// <summary>
/// Resets the Session cache as well as resetting the connection.
/// </summary>
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
/// <returns>The wrapped Session</returns>
protected virtual ISession GetCachedSessionWrapper(ISession targetSession, LinkedList sessionList)
{
return new CachedSession(targetSession, sessionList, SessionCacheSize, CacheProducers);
return new CachedSession(targetSession, sessionList, this);
}
}

23
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/ConnectionFactoryUtils.cs
View File

@@ -78,6 +78,23 @@ namespace Spring.Messaging.Nms.Connections
}
}
/// <summary>
/// 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.
/// </summary>
/// <param name="session">The session to unwrap</param>
/// <returns>The innermost target Session, or the passed-in one if no decorator</returns>
public static ISession GetTargetSession(ISession session)
{
ISession sessionToUse = session;
while (sessionToUse is IDecoratorSession)
{
sessionToUse = ((IDecoratorSession) sessionToUse).TargetSession;
}
return sessionToUse;
}
/// <summary>
/// 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
/// <summary> Callback for resource cleanup at the end of a non-native NMS transaction
/// </summary>
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;

2
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/IDecoratorSession.cs
View File

@@ -28,6 +28,8 @@ namespace Spring.Messaging.Nms.Connections
/// functionality. Allows access to the the underlying target Session.
/// </summary>
/// <author>Mark Pollack</author>
/// <see cref="ConnectionFactoryUtils.GetTargetSession(ISession)"/>
/// <see cref="CachingConnectionFactory"/>
public interface IDecoratorSession : ISession
{
/// <summary>

3
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsResourceHolder.cs
View File

@@ -273,6 +273,9 @@ namespace Spring.Messaging.Nms.Connections
{
ConnectionFactoryUtils.ReleaseConnection(connection, connectionFactory, true);
}
this.connections.Clear();
this.sessions.Clear();
this.sessionsPerIConnection.Clear();
}
/// <summary>

20
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/NmsTransactionManager.cs
View File

@@ -44,11 +44,21 @@ namespace Spring.Messaging.Nms.Connections
/// automatically participate in it.
/// </para>
/// <para>
/// This transaction strategy will typically be used in combination with
/// <see cref="SingleConnectionFactory"/>, 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 <see cref="CachingConnectionFactory"/>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.
/// </para>
/// <para>The use of a <i>raw</i> 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 <code>MessageProducer.close()</code> calls
/// and/or <code>MessageConsumer.close()</code> calls before <code>Session.commit()</code>,
/// 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 <see cref="CachingConnectionFactory"/> into this transaction manager's
/// ConnectionFactory property.
/// </para>
/// <para>
/// Transaction synchronization is turned off by default, as this manager might be used

65
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/SingleConnectionFactory.cs
View File

@@ -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.
/// </summary>
/// <remarks>
/// <para>
@@ -83,6 +83,12 @@ namespace Spring.Messaging.Nms.Connections
/// </summary>
private IConnection connection;
/// <summary>
/// Whether the shared Connection has been started
/// </summary>
private bool started = false;
/// <summary>
/// Synchronization monitor for the shared Connection
/// </summary>
@@ -193,6 +199,27 @@ namespace Spring.Messaging.Nms.Connections
set { reconnectOnException = value; }
}
/// <summary>
/// Gets the connection monitor.
/// </summary>
/// <value>The connection monitor.</value>
internal object ConnectionMonitor
{
get { return connectionMonitor; }
}
/// <summary>
/// Gets a value indicating whether this instance is started.
/// </summary>
/// <value>
/// <c>true</c> if this instance is started; otherwise, <c>false</c>.
/// </value>
internal bool IsStarted
{
get { return started;}
set { started = value; }
}
#endregion
#region IConnectionFactory Members
@@ -292,7 +319,7 @@ namespace Spring.Messaging.Nms.Connections
/// <param name="con">The connection to operate on.</param>
/// <param name="mode">The session ack mode.</param>
/// <returns>the Session to use, or <code>null</code> to indicate
/// creation of a default Session</returns>
/// creation of a raw standard Session</returns>
public virtual ISession GetSession(IConnection con, AcknowledgementMode mode)
{
return null;
@@ -313,11 +340,19 @@ namespace Spring.Messaging.Nms.Connections
/// <param name="con">The connection.</param>
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
/// <summary>
/// Add information to show this is a shared NMS connection
/// </summary>
/// <returns>Description of connection wrapper</returns>
public override string ToString()
{
return "Shared NMS Connection: " + this.target;
}
}
}

1
src/Spring/Spring.Messaging.Nms/Spring.Messaging.Nms.2005.csproj
View File

@@ -47,6 +47,7 @@
<Reference Include="System.Xml" />
</ItemGroup>
<ItemGroup>
<Compile Include="Messaging\Nms\Connections\CachedMessageConsumer .cs" />
<Compile Include="Messaging\Nms\Core\IExceptionListener.cs" />
<Compile Include="Messaging\Nms\Core\IMessageCreator.cs" />
<Compile Include="Messaging\Nms\Core\IMessageListener.cs" />

1
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/CachingConnectionFactoryTests.cs
View File

@@ -32,7 +32,6 @@ namespace Spring.Messaging.Nms.Connections
/// This class contains tests for
/// </summary>
/// <author>Mark Pollack</author>
/// <version>$Id:$</version>
[TestFixture]
public class CachingConnectionFactoryTests
{

59
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/SingleConnectionFactoryTests.cs
View File

@@ -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);
}
}
}

15
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestMessageProducer.cs
View File

@@ -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

6
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestSession.cs
View File

@@ -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