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

Caching of NMS sessions/message producers

Browse Source 
Add IsTrue utility method to AssertUtils
This commit is contained in:
markpollack
2008-07-18 15:30:44 +00:00
parent e51c7e9bc7
commit 6084d8f068
25 changed files with 1859 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/Spring/Spring.Core/Util/AssertUtils.cs
View File

@@ -191,6 +191,37 @@ namespace Spring.Util
{
throw new ArgumentException(message, argumentName);
}
}
/// <summary>
/// Assert a boolean expression, throwing <code>ArgumentException</code>
/// if the test result is <code>false</code>.
/// </summary>
/// <param name="expression">a boolean expression.</param>
/// <param name="message">The exception message to use if the assertion fails.</param>
/// <exception cref="ArgumentException">
/// if expression is <code>false</code>
/// </exception>
public static void IsTrue(bool expression, string message)
{
if (!expression)
{
throw new ArgumentException(message);
}
}
/// <summary>
/// Assert a boolean expression, throwing <code>ArgumentException</code>
/// if the test result is <code>false</code>.
/// </summary>
/// <param name="expression">a boolean expression.</param>
/// <exception cref="ArgumentException">
/// if expression is <code>false</code>
/// </exception>
public static void IsTrue(bool expression)
{
IsTrue(expression, "[Assertion failed] - this expression must be true");
}
/// <summary>

187
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedMessageProducer.cs Normal file
View File

@@ -0,0 +1,187 @@
#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.Connection
{
/// <summary>
/// MessageProducer decorator that adapts specific settings
/// to a shared MessageProducer instance underneath.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class CachedMessageProducer : IMessageProducer
{
private 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;
private TimeSpan timeToLive;
public CachedMessageProducer(IMessageProducer target)
{
this.target = target;
}
public IMessageProducer Target
{
get { return target; }
}
public void Send(IMessage message)
{
target.Send(message);
}
public void Send(IMessage message, bool persistent, byte priority, TimeSpan timeToLive)
{
target.Send(message, persistent, priority, timeToLive);
}
public void Send(IDestination destination, IMessage message)
{
target.Send(destination, message);
}
public void Send(IDestination destination, IMessage message, bool persistent, byte priority, TimeSpan timeToLive)
{
target.Send(destination, message, persistent, priority, timeToLive);
}
#region Odd Message Creationg Methods on IMessageProducer - not in-line with JMS APIs.
public IMessage CreateMessage()
{
return target.CreateMessage();
}
public ITextMessage CreateTextMessage()
{
return target.CreateTextMessage();
}
public ITextMessage CreateTextMessage(string text)
{
return target.CreateTextMessage(text);
}
public IMapMessage CreateMapMessage()
{
return target.CreateMapMessage();
}
public IObjectMessage CreateObjectMessage(object body)
{
return target.CreateObjectMessage(body);
}
public IBytesMessage CreateBytesMessage()
{
return target.CreateBytesMessage();
}
public IBytesMessage CreateBytesMessage(byte[] body)
{
return target.CreateBytesMessage(body);
}
#endregion
public bool Persistent
{
get { return persistent; }
set { persistent = value; }
}
public TimeSpan TimeToLive
{
get { return timeToLive; }
set { timeToLive = value; }
}
public byte Priority
{
get { return priority; }
set { priority = value;}
}
public bool DisableMessageID
{
get
{
return disableMessageID;
}
set
{
if (originalDisableMessageID == null)
{
originalDisableMessageID = value;
}
disableMessageID = value;
}
}
public bool DisableMessageTimestamp
{
get
{
return disableMessageTimestamp;
}
set
{
if (originalDisableMessageTimestamp == null)
{
originalDisableMessageTimestamp = value;
}
disableMessageTimestamp = value;
}
}
public void Dispose()
{
// It's a cached MessageProducer... reset properties only.
if (originalDisableMessageID != null)
{
target.DisableMessageID = (bool) originalDisableMessageID;
originalDisableMessageID = null;
}
if (originalDisableMessageTimestamp != null)
{
target.DisableMessageTimestamp = (bool) originalDisableMessageTimestamp;
originalDisableMessageTimestamp = null;
}
}
}
}

296
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachedSession.cs Normal file
View File

@@ -0,0 +1,296 @@
#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 Apache.NMS;
using Common.Logging;
using Spring.Collections;
using IQueue=Apache.NMS.IQueue;
namespace Spring.Messaging.Nms.Connection
{
/// <summary>
/// Wrapper for ISession that caches producers and registers itself as available
/// to the session cache when being closed. Generally used for testing purposes or
/// if need to get at the wrapped Session object via the TargetSession property (for
/// vendor specific methods).
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack</author>
public class CachedSession : IDecoratorSession
{
#region Logging Definition
private static readonly ILog LOG = LogManager.GetLogger(typeof(CachedSession));
#endregion
private ISession target;
private LinkedList sessionList;
private int sessionCacheSize;
private IDictionary cachedProducers = new Hashtable();
private IMessageProducer cachedUnspecifiedDestinationMessageProducer;
private bool shouldCacheProducers;
/// <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)
{
target = targetSession;
this.sessionList = sessionList;
this.sessionCacheSize = sessionCacheSize;
shouldCacheProducers = cacheProducers;
}
/// <summary>
/// Gets the target, for testing purposes.
/// </summary>
/// <value>The target.</value>
public ISession TargetSession
{
get { return target; }
}
/// <summary>
/// Creates the producer, potentially returning a cached instance.
/// </summary>
/// <returns>A message producer, potentially cached.</returns>
public IMessageProducer CreateProducer()
{
if (shouldCacheProducers)
{
if (cachedUnspecifiedDestinationMessageProducer != null)
{
#region Logging
if (LOG.IsDebugEnabled)
{
LOG.Debug("Found cached MessageProducer for unspecified destination");
}
#endregion
}
else
{
cachedUnspecifiedDestinationMessageProducer = target.CreateProducer();
#region Logging
if (LOG.IsDebugEnabled)
{
LOG.Debug("Created cached MessageProducer for unspecified destination");
}
#endregion
}
return new CachedMessageProducer(cachedUnspecifiedDestinationMessageProducer);
}
else
{
return target.CreateProducer();
}
}
/// <summary>
/// Creates the producer, potentially returning a cached instance.
/// </summary>
/// <param name="destination">The destination.</param>
/// <returns></returns>
public IMessageProducer CreateProducer(IDestination destination)
{
if (shouldCacheProducers)
{
IMessageProducer producer = (IMessageProducer)cachedProducers[destination];
if (producer != null)
{
#region Logging
if (LOG.IsDebugEnabled)
{
LOG.Debug("Found cached MessageProducer for destination [" + destination + "]");
}
#endregion
}
else
{
producer = target.CreateProducer(destination);
cachedProducers.Add(destination, producer);
#region Logging
if (LOG.IsDebugEnabled)
{
LOG.Debug("Created cached MessageProducer for destination [" + destination + "]");
}
#endregion
}
return new CachedMessageProducer(producer);
}
else
{
return target.CreateProducer(destination);
}
}
/// <summary>
/// If have not yet reached session cache size, cache the session, otherwise
/// dispose of all cached message producers and close the session.
/// </summary>
public void Close()
{
lock (sessionList)
{
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
{
foreach (DictionaryEntry entry in cachedProducers)
{
((IMessageProducer)entry.Value).Dispose();
}
target.Close();
if (LOG.IsDebugEnabled)
{
LOG.Debug("Closed cached Session: " + target);
}
}
}
}
#region Pass through implementations
public IMessageConsumer CreateConsumer(IDestination destination)
{
return target.CreateConsumer(destination);
}
public IMessageConsumer CreateConsumer(IDestination destination, string selector)
{
return target.CreateConsumer(destination, selector);
}
public IMessageConsumer CreateConsumer(IDestination destination, string selector, bool noLocal)
{
return target.CreateConsumer(destination, selector, noLocal);
}
public IMessageConsumer CreateDurableConsumer(ITopic destination, string name, string selector, bool noLocal)
{
return target.CreateDurableConsumer(destination, name, selector, noLocal);
}
public IQueue GetQueue(string name)
{
return target.GetQueue(name);
}
public ITopic GetTopic(string name)
{
return target.GetTopic(name);
}
public ITemporaryQueue CreateTemporaryQueue()
{
return target.CreateTemporaryQueue();
}
public ITemporaryTopic CreateTemporaryTopic()
{
return target.CreateTemporaryTopic();
}
public IMessage CreateMessage()
{
return target.CreateMessage();
}
public ITextMessage CreateTextMessage()
{
return target.CreateTextMessage();
}
public ITextMessage CreateTextMessage(string text)
{
return target.CreateTextMessage(text);
}
public IMapMessage CreateMapMessage()
{
return target.CreateMapMessage();
}
public IObjectMessage CreateObjectMessage(object body)
{
return target.CreateObjectMessage(body);
}
public IBytesMessage CreateBytesMessage()
{
return target.CreateBytesMessage();
}
public IBytesMessage CreateBytesMessage(byte[] body)
{
return target.CreateBytesMessage(body);
}
public void Commit()
{
target.Commit();
}
public void Rollback()
{
target.Rollback();
}
public bool Transacted
{
get { return target.Transacted; }
}
public AcknowledgementMode AcknowledgementMode
{
get { return target.AcknowledgementMode; }
}
public void Dispose()
{
target.Dispose();
}
#endregion
}
}

179
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/CachingConnectionFactory.cs Normal file
View File

@@ -0,0 +1,179 @@
#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 Apache.NMS;
using Common.Logging;
using Spring.Collections;
using Spring.Util;
using IQueue=Apache.NMS.IQueue;
namespace Spring.Messaging.Nms.Connection
{
/// <summary>
/// <see cref="SingleConnectionFactory"/> subclass that adds
/// ISession and IMessageProducer caching. This ConnectionFactory
/// also switches the ReconnectOnException property to true
/// by default, allowing for automatic recovery of the underlying
/// Connection.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack</author>
public class CachingConnectionFactory : SingleConnectionFactory
{
#region Logging Definition
private static readonly ILog LOG = LogManager.GetLogger(typeof(CachingConnectionFactory));
#endregion
private int sessionCacheSize = 1;
private bool cacheProducers = true;
private IDictionary cachedSessions = new Hashtable();
/// <summary>
/// Initializes a new instance of the <see cref="CachingConnectionFactory"/> class.
/// and sets the ReconnectOnException to true
/// </summary>
public CachingConnectionFactory()
{
ReconnectOnException = true;
}
/// <summary>
/// Gets or sets the size of the session cache.
/// </summary>
/// <remarks>
/// This cache size is the maximum limit for the number of cached Sessions
/// per session acknowledgement type (auto, client, dups_ok, transacted).
/// As a consequence, the actual number of cached Sessions may be up to
/// four times as high as the specified value - in the unlikely case
/// of mixing and matching different acknowledgement types.
/// <para>
/// Default is 1: caching a single Session, (re-)creating further ones on
/// demand. Specify a number like 10 if you'd like to raise the number of cached
/// Sessions; that said, 1 may be sufficient for low-concurrency scenarios.
/// </para>
/// </remarks>
/// <value>The size of the session cache.</value>
public int SessionCacheSize
{
get { return sessionCacheSize; }
set
{
AssertUtils.IsTrue(value >= 1, "Session cache size must be 1 or higher");
sessionCacheSize = value;
}
}
/// <summary>
/// Gets or sets a value indicating whether to cache MessageProducers per
/// Session instance. (more specifically: one MessageProducer per Destination
/// and Session).
/// </summary>
/// <remarks>
/// <para>Default is "true". Switch this to "false" in order to recreate,
/// MessageProducers on demand.
/// </para>
/// </remarks>
/// <value><c>true</c> if should cache message producers; otherwise, <c>false</c>.</value>
public bool CacheProducers
{
get { return cacheProducers; }
set { cacheProducers = value; }
}
/// <summary>
/// Resets the Session cache as well as resetting the connection.
/// </summary>
public override void ResetConnection()
{
lock (cachedSessions)
{
cachedSessions.Clear();
}
base.ResetConnection();
}
/// <summary>
/// Obtaining a cached Session.
/// </summary>
/// <param name="con">The connection to operate on.</param>
/// <param name="mode">The session ack mode.</param>
/// <returns>The Session to use
/// </returns>
public override ISession GetSession(IConnection con, AcknowledgementMode mode)
{
LinkedList sessionList;
lock (cachedSessions)
{
sessionList = (LinkedList) cachedSessions[mode];
if (sessionList == null)
{
sessionList = new LinkedList();
cachedSessions.Add(mode, sessionList);
}
}
ISession session = null;
lock (sessionList)
{
if (sessionList.Count > 0)
{
session = (ISession) sessionList[0];
sessionList.RemoveAt(0);
}
}
if (session != null)
{
if (LOG.IsDebugEnabled)
{
LOG.Debug("Found cached Session for mode " + mode + ": " + session);
}
} else
{
ISession targetSession = con.CreateSession(mode);
session = GetCachedSessionWrapper(targetSession, sessionList);
if (LOG.IsDebugEnabled)
{
LOG.Debug("Created cached Session for mode " + mode + ": " + session);
}
}
return session;
}
protected virtual ISession GetCachedSessionWrapper(ISession targetSession, LinkedList sessionList)
{
return new CachedSession(targetSession, sessionList, SessionCacheSize, CacheProducers);
}
}
}

72
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/ChainedExceptionListener.cs Normal file
View File

@@ -0,0 +1,72 @@
#region License
/*
* Copyright <20> 2002-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#endregion
using System;
using System.Collections;
using Spring.Util;
namespace Spring.Messaging.Nms.Connection
{
/// <summary>
/// Implementation of Spring IExceptionListener interface that supports
/// chaining allowing the addition of multiple ExceptionListener instances in order.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class ChainedExceptionListener : IExceptionListener
{
private ArrayList listeners = new ArrayList(2);
/// <summary>
/// Adds the exception listener to the chain
/// </summary>
/// <param name="listener">The listener.</param>
public void AddListener(IExceptionListener listener)
{
AssertUtils.ArgumentNotNull(listener, "listener", "ExceptionListener must not be null");
listeners.Add(listener);
}
/// <summary>
/// Called when an exception occurs in message processing.
/// </summary>
/// <param name="exception">The exception.</param>
public void OnException(Exception exception)
{
foreach (IExceptionListener listener in listeners)
{
listener.OnException(exception);
}
}
/// <summary>
/// Gets the exception listeners as an array.
/// </summary>
/// <value>The exception listeners.</value>
public IExceptionListener[] Listeners
{
get
{
return (IExceptionListener[]) listeners.ToArray(typeof (IExceptionListener));
}
}
}
}

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

@@ -24,7 +24,7 @@ using Common.Logging;
using Spring.Transaction.Support;
using Spring.Util;
namespace Spring.Messaging.Nms.IConnections
namespace Spring.Messaging.Nms.Connection
{
/// <summary> Helper class for obtaining transactional NMS resources
/// for a given IConnectionFactory.
@@ -340,7 +340,6 @@ namespace Spring.Messaging.Nms.IConnections
}
}
//TODO bring in new Spring.Data library to Integration project which has this method in interface.
public override void AfterCommit()
{
if (transacted)

40
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/IDecoratorSession.cs Normal file
View File

@@ -0,0 +1,40 @@
#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 Apache.NMS;
namespace Spring.Messaging.Nms.Connection
{
/// <summary>
/// Subinterface of ISession to be implemented by
/// implementations that wrap an ISession to provide added
/// functionality. Allows access to the the underlying target Session.
/// </summary>
/// <author>Mark Pollack</author>
public interface IDecoratorSession : ISession
{
/// <summary>
/// Gets the target session of the decorator.
/// This will typically be the native provider Session or a wrapper from a session pool.
/// </summary>
/// <value>The underlying session, never null</value>
ISession TargetSession { get; }
}
}

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

@@ -22,7 +22,7 @@
using Apache.NMS;
namespace Spring.Messaging.Nms.IConnections
namespace Spring.Messaging.Nms.Connection
{
/// <summary>
/// Extension of the <code>IConnectionFactory</code> interface,

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

@@ -26,7 +26,7 @@ using Spring.Transaction.Support;
using Spring.Util;
using Apache.NMS;
namespace Spring.Messaging.Nms.IConnections
namespace Spring.Messaging.Nms.Connection
{
/// <summary> IConnection holder, wrapping a NMS IConnection and a NMS ISession.
/// NmsTransactionManager binds instances of this class to the thread,

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

@@ -8,7 +8,7 @@ using Spring.Objects.Factory;
using Spring.Transaction;
using Spring.Transaction.Support;
namespace Spring.Messaging.Nms.IConnections
namespace Spring.Messaging.Nms.Connection
{
/// <summary>
/// A <see cref="AbstractPlatformTransactionManager"/> implementation

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

@@ -24,9 +24,9 @@ using Common.Logging;
using Spring.Objects.Factory;
using Spring.Util;
namespace Spring.Messaging.Nms.IConnections
namespace Spring.Messaging.Nms.Connection
{
public class SingleConnectionFactory : IConnectionFactory, IInitializingObject, IDisposable
public class SingleConnectionFactory : IConnectionFactory, IExceptionListener, IInitializingObject, IDisposable
{
#region Logging Definition
@@ -34,11 +34,13 @@ namespace Spring.Messaging.Nms.IConnections
#endregion
#region Fields
private IConnectionFactory targetConnectionFactory;
private string clientId;
private ExceptionListener exceptionListenerDelegate;
private IExceptionListener exceptionListener;
private bool reconnectOnException = false;
@@ -57,6 +59,9 @@ namespace Spring.Messaging.Nms.IConnections
/// </summary>
private object connectionMonitor = new object();
#endregion
#region Constructors
/// <summary>
/// Initializes a new instance of the <see cref="SingleConnectionFactory"/> class.
@@ -73,9 +78,9 @@ namespace Spring.Messaging.Nms.IConnections
/// <param name="target">The single Connection.</param>
public SingleConnectionFactory(IConnection target)
{
AssertUtils.ArgumentNotNull(target, "connection", "Target Connection must not be null");
AssertUtils.ArgumentNotNull(target, "connection", "TargetSession Connection must not be null");
this.target = target;
connection = GetSharedConnection(target);
connection = GetSharedConnection(this, target);
}
@@ -87,10 +92,14 @@ namespace Spring.Messaging.Nms.IConnections
public SingleConnectionFactory(IConnectionFactory targetConnectionFactory)
{
AssertUtils.ArgumentNotNull(targetConnectionFactory, "targetConnectionFactory",
"Target ConnectionFactory must not be null");
"TargetSession ConnectionFactory must not be null");
this.targetConnectionFactory = targetConnectionFactory;
}
#endregion
#region Properties
/// <summary>
/// Gets or sets the target connection factory which will be used to create a single
/// connection.
@@ -118,18 +127,12 @@ namespace Spring.Messaging.Nms.IConnections
}
/// <summary>
/// Gets or sets the exception listener delegate that should be registered with
/// the single connection created by this factory.
/// </summary>
/// <value>The exception listener delegate.</value>
public ExceptionListener ExceptionListenerDelegate
public IExceptionListener ExceptionListener
{
get { return exceptionListenerDelegate; }
set { exceptionListenerDelegate = value; }
get { return exceptionListener; }
set { exceptionListener = value; }
}
/// <summary>
/// Gets or sets a value indicating whether the single Connection
/// should be reset (to be subsequently renewed) when a NMSException
@@ -155,6 +158,8 @@ namespace Spring.Messaging.Nms.IConnections
set { reconnectOnException = value; }
}
#endregion
#region IConnectionFactory Members
public IConnection CreateConnection()
@@ -169,6 +174,13 @@ namespace Spring.Messaging.Nms.IConnections
}
}
public IConnection CreateConnection(string userName, string password)
{
throw new InvalidOperationException("SingleConnectionFactory does not support custom username and password.");
}
#endregion
public void InitConnection()
{
if (TargetConnectionFactory == null)
@@ -188,34 +200,61 @@ namespace Spring.Messaging.Nms.IConnections
{
LOG.Info("Established shared NMS Connection: " + this.target);
}
this.connection = GetSharedConnection(this.target);
this.connection = GetSharedConnection(this, target);
}
}
/// <summary>
/// Exception listener callback that renews the underlying single Connection.
/// </summary>
/// <param name="exception">The exception from the messaging infrastructure.</param>
public void OnException(Exception exception)
{
ResetConnection();
}
protected virtual void PrepareConnection(IConnection con)
{
if (ClientId != null)
{
con.ClientId = ClientId;
}
if (ExceptionListenerDelegate != null || ReconnectOnException)
if (ExceptionListener != null || ReconnectOnException)
{
ExceptionListener listenerToUse = ExceptionListenerDelegate;
IExceptionListener listenerToUse = ExceptionListener;
if (ReconnectOnException)
{
InternalChainedExceptionListenerSupport chained = new InternalChainedExceptionListenerSupport(this, listenerToUse);
InternalChainedExceptionListener chained = new InternalChainedExceptionListener(this, listenerToUse);
con.ExceptionListener += chained.OnException;
}
else
{
if (ExceptionListener != null)
{
con.ExceptionListener += ExceptionListener.OnException;
}
}
}
}
/// <summary>
/// Template method for obtaining a (potentially cached) Session.
/// </summary>
/// <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>
public virtual ISession GetSession(IConnection con, AcknowledgementMode mode)
{
return null;
}
protected virtual IConnection DoCreateConnection()
{
return TargetConnectionFactory.CreateConnection();
}
private void CloseConnection(IConnection con)
protected virtual void CloseConnection(IConnection con)
{
try
{
@@ -232,13 +271,6 @@ namespace Spring.Messaging.Nms.IConnections
}
}
public IConnection CreateConnection(string userName, string password)
{
throw new NotImplementedException();
}
#endregion
#region IInitializingObject Members
public void AfterPropertiesSet()
@@ -251,14 +283,12 @@ namespace Spring.Messaging.Nms.IConnections
#endregion
#region IDisposable Members
public void Dispose()
{
ResetConnection();
}
public void ResetConnection()
public virtual void ResetConnection()
{
lock (connectionMonitor)
{
@@ -271,62 +301,79 @@ namespace Spring.Messaging.Nms.IConnections
}
}
#endregion
protected virtual IConnection GetSharedConnection(IConnection target)
protected virtual IConnection GetSharedConnection(SingleConnectionFactory singleConnectionFactory, IConnection target)
{
lock (connectionMonitor)
{
return new CloseSupressingConnection(target);
return new CloseSupressingConnection(singleConnectionFactory, target);
}
}
}
internal class InternalChainedExceptionListenerSupport
internal class InternalChainedExceptionListener : ChainedExceptionListener, IExceptionListener
{
private SingleConnectionFactory factory;
private ExceptionListener listenerToUse;
public InternalChainedExceptionListenerSupport(SingleConnectionFactory factory, ExceptionListener listenerToUse)
private IExceptionListener userListener;
public InternalChainedExceptionListener(IExceptionListener internalListener, IExceptionListener userListener)
{
this.factory = factory;
this.listenerToUse = listenerToUse;
AddListener(internalListener);
if (userListener != null)
{
AddListener(userListener);
this.userListener = userListener;
}
}
public void OnException(Exception exception)
public IExceptionListener UserListener
{
//TODO exception mgmt.
get { return userListener; }
}
}
internal class CloseSupressingConnection : IConnection
{
private IConnection target;
private SingleConnectionFactory singleConnectionFactory;
public CloseSupressingConnection(IConnection target)
public CloseSupressingConnection(SingleConnectionFactory singleConnectionFactory, IConnection target)
{
this.target = target;
this.singleConnectionFactory = singleConnectionFactory;
}
public void Close()
{
// don't pass the call to the target.
}
public void Stop()
{
//don't pass the call to the target.
}
public ISession CreateSession()
{
return CreateSession(AcknowledgementMode.AutoAcknowledge);
}
public ISession CreateSession(AcknowledgementMode acknowledgementMode)
{
ISession session = singleConnectionFactory.GetSession(target, acknowledgementMode);
if (session != null)
{
return session;
}
return target.CreateSession();
}
#region Pass through implementations to the target connection
public event ExceptionListener ExceptionListener
{
add { target.ExceptionListener += value; }
remove { target.ExceptionListener -= value; }
}
public ISession CreateSession()
{
return target.CreateSession();
}
public ISession CreateSession(AcknowledgementMode acknowledgementMode)
{
return target.CreateSession(acknowledgementMode);
}
public void Close()
{
// don't pass the call to the target.
}
public AcknowledgementMode AcknowledgementMode
{
@@ -354,10 +401,7 @@ namespace Spring.Messaging.Nms.IConnections
{
get { return target.IsStarted; }
}
#endregion
public void Stop()
{
//don't pass the call to the target.
}
}
}

30
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Connections/SynchedLocalTransactionFailedException.cs
View File

@@ -19,25 +19,19 @@
#endregion
using System;
using System.Runtime.Serialization;
using Apache.NMS;
namespace Spring.Messaging.Nms.IConnections
{
//TODO should we have a generic spring exception for NMS?
namespace Spring.Messaging.Nms.Connection
{
/// <summary> Exception thrown when a synchronized local transaction failed to complete
/// (after the main transaction has already completed).
/// </summary>
/// <author>Jergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
[Serializable]
public class SynchedLocalTransactionFailedException : ApplicationException
public class SynchedLocalTransactionFailedException : NMSException
{
#region Constructor (s) / Destructor
/// <summary>Creates a new instance of the SynchedLocalTransactionFailedException class.</summary>
public SynchedLocalTransactionFailedException()
{
}
/// <summary>
/// Creates a new instance of the SynchedLocalTransactionFailedException class. with the specified message.
@@ -64,22 +58,6 @@ namespace Spring.Messaging.Nms.IConnections
{
}
/// <summary>
/// Creates a new instance of the SynchedLocalTransactionFailedException class.
/// </summary>
/// <param name="info">
/// The <see cref="System.Runtime.Serialization.SerializationInfo"/>
/// that holds the serialized object data about the exception being thrown.
/// </param>
/// <param name="context">
/// The <see cref="System.Runtime.Serialization.StreamingContext"/>
/// that contains contextual information about the source or destination.
/// </param>
protected SynchedLocalTransactionFailedException(
SerializationInfo info, StreamingContext context)
: base (info, context)
{
}
#endregion
}
}

33
src/Spring/Spring.Messaging.Nms/Messaging/Nms/IExceptionListener.cs Normal file
View File

@@ -0,0 +1,33 @@
#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;
namespace Spring.Messaging.Nms
{
/// <summary>
/// Exception handler for exceptions from the messaging infrastrcture.
/// </summary>
/// <author>Mark Pollack</author>
public interface IExceptionListener
{
void OnException(Exception exception);
}
}

2
src/Spring/Spring.Messaging.Nms/Messaging/Nms/Listener/AbstractMessageListenerContainer.cs
View File

@@ -1,7 +1,7 @@
using System;
using System.Collections;
using Spring.Context;
using Spring.Messaging.Nms.IConnections;
using Spring.Messaging.Nms.Connection;
using Spring.Messaging.Nms.Support;
using Spring.Messaging.Nms.Support.IDestinations;
using Spring.Util;

2
src/Spring/Spring.Messaging.Nms/Messaging/Nms/NmsTemplate.cs
View File

@@ -19,7 +19,7 @@
#endregion
using System;
using Spring.Messaging.Nms.IConnections;
using Spring.Messaging.Nms.Connection;
using Spring.Messaging.Nms.Support;
using Spring.Messaging.Nms.Support.Converter;
using Spring.Messaging.Nms.Support.IDestinations;

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

@@ -47,12 +47,18 @@
</ItemGroup>
<ItemGroup>
<Compile Include="Context\ILifecycle.cs" />
<Compile Include="Messaging\Nms\Connections\CachedMessageProducer.cs" />
<Compile Include="Messaging\Nms\Connections\CachedSession.cs" />
<Compile Include="Messaging\Nms\Connections\CachingConnectionFactory.cs" />
<Compile Include="Messaging\Nms\Connections\ChainedExceptionListener.cs" />
<Compile Include="Messaging\Nms\Connections\ConnectionFactoryUtils.cs" />
<Compile Include="Messaging\Nms\Connections\IDecoratorSession.cs" />
<Compile Include="Messaging\Nms\Connections\ISmartConnectionFactory.cs" />
<Compile Include="Messaging\Nms\Connections\NmsResourceHolder.cs" />
<Compile Include="Messaging\Nms\Connections\NmsTransactionManager.cs" />
<Compile Include="Messaging\Nms\Connections\SingleConnectionFactory.cs" />
<Compile Include="Messaging\Nms\Connections\SynchedLocalTransactionFailedException.cs" />
<Compile Include="Messaging\Nms\IExceptionListener.cs" />
<Compile Include="Messaging\Nms\IMessageCreator.cs" />
<Compile Include="Messaging\Nms\IMessageListener.cs" />
<Compile Include="Messaging\Nms\IMessagePostProcessor.cs" />

26
test/Spring/Spring.Core.Tests/Util/AssertUtilsTests.cs
View File

@@ -35,6 +35,32 @@ namespace Spring.Util
[TestFixture]
public sealed class AssertUtilsTests
{
[Test]
[ExpectedException(typeof(ArgumentException),ExpectedMessage = "foo")]
public void IsTrueWithMesssage()
{
AssertUtils.IsTrue(false,"foo");
}
[Test]
public void IsTrueWithMessageValidExpression()
{
AssertUtils.IsTrue(true, "foo");
}
[Test]
[ExpectedException(typeof(ArgumentException), ExpectedMessage = "[Assertion failed] - this expression must be true")]
public void IsTrue()
{
AssertUtils.IsTrue(false);
}
[Test]
public void IsTrueValidExpression()
{
AssertUtils.IsTrue(true);
}
[Test]
[ExpectedException(typeof(InvalidOperationException))]
public void StateTrue()

221
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/CachingConnectionFactoryTests.cs Normal file
View File

@@ -0,0 +1,221 @@
#region License
/*
* Copyright <20> 2002-2007 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
#region Imports
using Apache.NMS;
using NUnit.Framework;
using Rhino.Mocks;
#endregion
namespace Spring.Messaging.Nms.Connection
{
/// <summary>
/// This class contains tests for
/// </summary>
/// <author>Mark Pollack</author>
/// <version>$Id:$</version>
[TestFixture]
public class CachingConnectionFactoryTests
{
private MockRepository mocks;
[SetUp]
public void Setup()
{
mocks = new MockRepository();
}
[Test]
public void CachedSession()
{
IConnectionFactory connectionFactory = (IConnectionFactory)mocks.CreateMock(typeof(IConnectionFactory));
IConnection connection = new TestConnection();
Expect.Call(connectionFactory.CreateConnection()).Return(connection).Repeat.Once();
mocks.ReplayAll();
CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
cachingConnectionFactory.TargetConnectionFactory = connectionFactory;
IConnection con1 = cachingConnectionFactory.CreateConnection();
ISession session1 = con1.CreateSession(AcknowledgementMode.Transactional);
TestSession testSession = GetTestSession(session1);
Assert.AreEqual(1, testSession.CreatedCount);
Assert.AreEqual(0, testSession.CloseCount);
session1.Close(); // won't close, will put in session cache.
Assert.AreEqual(0, testSession.CloseCount);
ISession session2 = con1.CreateSession(AcknowledgementMode.Transactional);
TestSession testSession2 = GetTestSession(session2);
Assert.AreSame(testSession, testSession2);
Assert.AreEqual(1, testSession.CreatedCount);
Assert.AreEqual(0, testSession.CloseCount);
mocks.VerifyAll();
//don't explicitly call close on
}
private static TestSession GetTestSession(ISession session1)
{
CachedSession cachedSession = session1 as CachedSession;
Assert.IsNotNull(cachedSession);
TestSession testSession = cachedSession.TargetSession as TestSession;
Assert.IsNotNull(testSession);
return testSession;
}
[Test]
public void CachedSessionTwoRequests()
{
IConnectionFactory connectionFactory = (IConnectionFactory)mocks.CreateMock(typeof(IConnectionFactory));
IConnection connection = new TestConnection();
Expect.Call(connectionFactory.CreateConnection()).Return(connection).Repeat.Once();
mocks.ReplayAll();
CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
cachingConnectionFactory.TargetConnectionFactory = connectionFactory;
IConnection con1 = cachingConnectionFactory.CreateConnection();
ISession session1 = con1.CreateSession(AcknowledgementMode.Transactional);
TestSession testSession1 = GetTestSession(session1);
Assert.AreEqual(1, testSession1.CreatedCount);
Assert.AreEqual(0, testSession1.CloseCount);
//will create a new one, not in the cache.
ISession session2 = con1.CreateSession(AcknowledgementMode.Transactional);
TestSession testSession2 = GetTestSession(session2);
Assert.AreEqual(1, testSession2.CreatedCount);
Assert.AreEqual(0, testSession2.CloseCount);
Assert.AreNotSame(testSession1, testSession2);
Assert.AreNotSame(session1, session2);
session1.Close(); // will be put in the cache
ISession session3 = con1.CreateSession(AcknowledgementMode.Transactional);
TestSession testSession3 = GetTestSession(session3);
Assert.AreSame(testSession1, testSession3);
Assert.AreSame(session1, session3);
Assert.AreEqual(1, testSession1.CreatedCount);
Assert.AreEqual(0, testSession1.CloseCount);
mocks.VerifyAll();
}
/// <summary>
/// Tests that the same underlying instance of the message producer is returned after
/// creating a session, creating the producer (A), closing the session, and creating another
/// producer (B). Assert that (A)=(B).
/// </summary>
[Test]
public void CachedMessageProducer()
{
IConnectionFactory connectionFactory = (IConnectionFactory)mocks.CreateMock(typeof(IConnectionFactory));
IConnection connection = new TestConnection();
Expect.Call(connectionFactory.CreateConnection()).Return(connection).Repeat.Once();
mocks.ReplayAll();
CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
cachingConnectionFactory.TargetConnectionFactory = connectionFactory;
IConnection con1 = cachingConnectionFactory.CreateConnection();
ISession sessionA = con1.CreateSession(AcknowledgementMode.Transactional);
IMessageProducer producerA = sessionA.CreateProducer();
TestMessageProducer tmpA = GetTestMessageProducer(producerA);
sessionA.Close();
ISession sessionB = con1.CreateSession(AcknowledgementMode.Transactional);
IMessageProducer producerB = sessionB.CreateProducer();
TestMessageProducer tmpB = GetTestMessageProducer(producerB);
Assert.AreSame(tmpA, tmpB);
mocks.VerifyAll();
}
[Test]
public void CachedMessageProducerTwoRequests()
{
IConnectionFactory connectionFactory = (IConnectionFactory)mocks.CreateMock(typeof(IConnectionFactory));
IConnection connection = new TestConnection();
Expect.Call(connectionFactory.CreateConnection()).Return(connection).Repeat.Once();
mocks.ReplayAll();
CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
cachingConnectionFactory.TargetConnectionFactory = connectionFactory;
IConnection con1 = cachingConnectionFactory.CreateConnection();
ISession sessionA = con1.CreateSession(AcknowledgementMode.Transactional);
IMessageProducer producerA = sessionA.CreateProducer();
TestMessageProducer tmpA = GetTestMessageProducer(producerA);
ISession sessionB = con1.CreateSession(AcknowledgementMode.Transactional);
IMessageProducer producerB = sessionB.CreateProducer();
TestMessageProducer tmpB = GetTestMessageProducer(producerB);
Assert.AreNotSame(tmpA, tmpB);
sessionA.Close();
ISession sessionC = con1.CreateSession(AcknowledgementMode.Transactional);
IMessageProducer producerC = sessionC.CreateProducer();
TestMessageProducer tmpC = GetTestMessageProducer(producerC);
Assert.AreSame(tmpA, tmpC);
mocks.VerifyAll();
}
private static TestMessageProducer GetTestMessageProducer(IMessageProducer producer1)
{
CachedMessageProducer cmp1 = producer1 as CachedMessageProducer;
Assert.IsNotNull(cmp1);
TestMessageProducer tmp1 = cmp1.Target as TestMessageProducer;
Assert.IsNotNull(tmp1);
return tmp1;
}
}
}

4
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/NmsTransactionManagerTests.cs
View File

@@ -24,13 +24,13 @@ using System;
using Apache.NMS;
using NUnit.Framework;
using Rhino.Mocks;
using Spring.Messaging.Nms.IConnections;
using Spring.Messaging.Nms.Connection;
using Spring.Transaction;
using Spring.Transaction.Support;
#endregion
namespace Spring.Messaging.Nms.Connections
namespace Spring.Messaging.Nms.Connection
{
/// <summary>
/// This class contains tests for

235
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/SingleConnectionFactoryTests.cs Normal file
View File

@@ -0,0 +1,235 @@
#region License
/*
* Copyright <20> 2002-2007 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
#region Imports
using System;
using Apache.NMS;
using NUnit.Framework;
using Rhino.Mocks;
using Spring.Messaging.Nms.Connection;
#endregion
namespace Spring.Messaging.Nms.Connection
{
/// <summary>
/// This class contains tests for the SingleConnectionFactory
/// </summary>
/// <author>Mark Pollack</author>
[TestFixture]
public class SingleConnectionFactoryTests
{
private MockRepository mocks;
[SetUp]
public void Setup()
{
mocks = new MockRepository();
}
[Test]
public void UsingConnection()
{
IConnection connection = (IConnection)mocks.CreateMock(typeof(IConnection));
connection.Start();
LastCall.On(connection).Repeat.Twice();
connection.Stop();
LastCall.On(connection).Repeat.Once();
connection.Close();
LastCall.On(connection).Repeat.Once();
mocks.ReplayAll();
SingleConnectionFactory scf = new SingleConnectionFactory(connection);
IConnection con1 = scf.CreateConnection();
con1.Start();
con1.Stop(); // should be ignored
con1.Close(); // should be ignored
IConnection con2 = scf.CreateConnection();
con2.Start();
con2.Stop(); // should be ignored
con2.Close(); // should be ignored.
scf.Dispose();
mocks.VerifyAll();
}
[Test]
public void UsingConnectionFactory()
{
IConnectionFactory connectionFactory = (IConnectionFactory)mocks.CreateMock(typeof(IConnectionFactory));
IConnection connection = (IConnection)mocks.CreateMock(typeof(IConnection));
Expect.Call(connectionFactory.CreateConnection()).Return(connection).Repeat.Once();
connection.Start();
LastCall.On(connection).Repeat.Twice();
connection.Stop();
LastCall.On(connection).Repeat.Once();
connection.Close();
LastCall.On(connection).Repeat.Once();
mocks.ReplayAll();
SingleConnectionFactory scf = new SingleConnectionFactory(connectionFactory);
IConnection con1 = scf.CreateConnection();
con1.Start();
con1.Close(); // should be ignored
IConnection con2 = scf.CreateConnection();
con2.Start();
con2.Close(); //should be ignored
scf.Dispose(); //should trigger actual close
mocks.VerifyAll();
}
[Test]
public void UsingConnectionFactoryAndClientId()
{
IConnectionFactory connectionFactory = (IConnectionFactory)mocks.CreateMock(typeof(IConnectionFactory));
IConnection connection = (IConnection)mocks.CreateMock(typeof(IConnection));
Expect.Call(connectionFactory.CreateConnection()).Return(connection).Repeat.Once();
connection.ClientId = "MyId";
LastCall.On(connection).Repeat.Once();
connection.Start();
LastCall.On(connection).Repeat.Twice();
connection.Stop();
LastCall.On(connection).Repeat.Once();
connection.Close();
LastCall.On(connection).Repeat.Once();
mocks.ReplayAll();
SingleConnectionFactory scf = new SingleConnectionFactory(connectionFactory);
scf.ClientId = "MyId";
IConnection con1 = scf.CreateConnection();
con1.Start();
con1.Close(); // should be ignored
IConnection con2 = scf.CreateConnection();
con2.Start();
con2.Close(); // should be ignored
scf.Dispose(); // should trigger actual close
mocks.VerifyAll();
}
[Test]
public void UsingConnectionFactoryAndExceptionListener()
{
IConnectionFactory connectionFactory = (IConnectionFactory)mocks.CreateMock(typeof(IConnectionFactory));
IConnection connection = (IConnection)mocks.CreateMock(typeof(IConnection));
IExceptionListener listener = new ChainedExceptionListener();
Expect.Call(connectionFactory.CreateConnection()).Return(connection).Repeat.Once();
connection.ExceptionListener += listener.OnException;
LastCall.On(connection).IgnoreArguments();
connection.Start();
LastCall.On(connection).Repeat.Twice();
connection.Stop();
LastCall.On(connection).Repeat.Once();
connection.Close();
LastCall.On(connection).Repeat.Once();
mocks.ReplayAll();
SingleConnectionFactory scf = new SingleConnectionFactory(connectionFactory);
scf.ExceptionListener = listener;
IConnection con1 = scf.CreateConnection();
//can't look at invocation list on event ...grrr.
con1.Start();
con1.Stop(); // should be ignored
con1.Close(); // should be ignored
IConnection con2 = scf.CreateConnection();
con2.Start();
con2.Stop();
con2.Close();
scf.Dispose();
mocks.VerifyAll();
}
[Test]
public void UsingConnectionFactoryAndReconnectOnException()
{
IConnectionFactory connectionFactory = (IConnectionFactory)mocks.CreateMock(typeof(IConnectionFactory));
TestConnection con = new TestConnection();
Expect.Call(connectionFactory.CreateConnection()).Return(con).Repeat.Twice();
mocks.ReplayAll();
SingleConnectionFactory scf = new SingleConnectionFactory(connectionFactory);
scf.ReconnectOnException = true;
IConnection con1 = scf.CreateConnection();
con1.Start();
con.FireExcpetionEvent(new NMSException(""));
IConnection con2 = scf.CreateConnection();
con2.Start();
scf.Dispose();
mocks.VerifyAll();
Assert.AreEqual(2, con.StartCount);
Assert.AreEqual(2, con.CloseCount);
}
[Test]
public void UsingConnectionFactoryAndExceptionListenerAndReconnectOnException()
{
IConnectionFactory connectionFactory = (IConnectionFactory)mocks.CreateMock(typeof(IConnectionFactory));
TestConnection con = new TestConnection();
TestExceptionListener listener = new TestExceptionListener();
Expect.Call(connectionFactory.CreateConnection()).Return(con).Repeat.Twice();
mocks.ReplayAll();
SingleConnectionFactory scf = new SingleConnectionFactory(connectionFactory);
scf.ExceptionListener = listener;
scf.ReconnectOnException = true;
IConnection con1 = scf.CreateConnection();
//Assert.AreSame(listener, );
con1.Start();
con.FireExcpetionEvent(new NMSException(""));
IConnection con2 = scf.CreateConnection();
con2.Start();
scf.Dispose();
mocks.VerifyAll();
Assert.AreEqual(2, con.StartCount);
Assert.AreEqual(2, con.CloseCount);
Assert.AreEqual(1, listener.Count);
}
}
}

82
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestConnection.cs Normal file
View File

@@ -0,0 +1,82 @@
using System;
using Apache.NMS;
namespace Spring.Messaging.Nms.Connection
{
public class TestConnection : IConnection
{
private int startCount;
private int closeCount;
private int createSessionCount;
private int closeSessionCount;
public int StartCount
{
get { return startCount; }
}
public int CloseCount
{
get { return closeCount; }
}
public event ExceptionListener ExceptionListener;
public ISession CreateSession()
{
createSessionCount++;
return new TestSession();
}
public ISession CreateSession(AcknowledgementMode acknowledgementMode)
{
createSessionCount++;
return new TestSession();
}
public void Close()
{
closeCount++;
}
public AcknowledgementMode AcknowledgementMode
{
get { return AcknowledgementMode.ClientAcknowledge; }
set { }
}
public string ClientId
{
get { return null; }
set { }
}
public void Dispose()
{
}
public void Start()
{
startCount++;
}
public bool IsStarted
{
get
{
if (startCount > 0) return true;
return false;
}
}
public void Stop()
{
}
public void FireExcpetionEvent(Exception e)
{
ExceptionListener(e);
}
}
}

40
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestExceptionListener.cs Normal file
View File

@@ -0,0 +1,40 @@
#region License
/*
* Copyright <20> 2002-2007 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;
namespace Spring.Messaging.Nms.Connection
{
public class TestExceptionListener : IExceptionListener
{
private int count = 0;
public void OnException(Exception exception)
{
count++;
}
public int Count
{
get { return count; }
}
}
}

119
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestMessageProducer.cs Normal file
View File

@@ -0,0 +1,119 @@
#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.Connection
{
public class TestMessageProducer : IMessageProducer
{
public void Send(IMessage message)
{
throw new NotImplementedException();
}
public void Send(IMessage message, bool persistent, byte priority, TimeSpan timeToLive)
{
throw new NotImplementedException();
}
public void Send(IDestination destination, IMessage message)
{
throw new NotImplementedException();
}
public void Send(IDestination destination, IMessage message, bool persistent, byte priority, TimeSpan timeToLive)
{
throw new NotImplementedException();
}
public IMessage CreateMessage()
{
throw new NotImplementedException();
}
public ITextMessage CreateTextMessage()
{
throw new NotImplementedException();
}
public ITextMessage CreateTextMessage(string text)
{
throw new NotImplementedException();
}
public IMapMessage CreateMapMessage()
{
throw new NotImplementedException();
}
public IObjectMessage CreateObjectMessage(object body)
{
throw new NotImplementedException();
}
public IBytesMessage CreateBytesMessage()
{
throw new NotImplementedException();
}
public IBytesMessage CreateBytesMessage(byte[] body)
{
throw new NotImplementedException();
}
public bool Persistent
{
get { throw new NotImplementedException(); }
set { throw new NotImplementedException(); }
}
public TimeSpan TimeToLive
{
get { throw new NotImplementedException(); }
set { throw new NotImplementedException(); }
}
public byte Priority
{
get { throw new NotImplementedException(); }
set { throw new NotImplementedException(); }
}
public bool DisableMessageID
{
get { throw new NotImplementedException(); }
set { throw new NotImplementedException(); }
}
public bool DisableMessageTimestamp
{
get { throw new NotImplementedException(); }
set { throw new NotImplementedException(); }
}
public void Dispose()
{
throw new NotImplementedException();
}
}
}

164
test/Spring/Spring.Messaging.Nms.Tests/Messaging/Nms/Connections/TestSession.cs Normal file
View File

@@ -0,0 +1,164 @@
#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.Connection
{
public class TestSession : ISession
{
private int closeCount;
private int createdCount;
public TestSession()
{
createdCount++;
}
public int CloseCount
{
get { return closeCount; }
}
public int CreatedCount
{
get { return createdCount; }
}
public IMessageProducer CreateProducer()
{
return new TestMessageProducer();
}
public IMessageProducer CreateProducer(IDestination destination)
{
return new TestMessageProducer();
}
public IMessageConsumer CreateConsumer(IDestination destination)
{
throw new NotImplementedException();
}
public IMessageConsumer CreateConsumer(IDestination destination, string selector)
{
throw new NotImplementedException();
}
public IMessageConsumer CreateConsumer(IDestination destination, string selector, bool noLocal)
{
throw new NotImplementedException();
}
public IMessageConsumer CreateDurableConsumer(ITopic destination, string name, string selector, bool noLocal)
{
throw new NotImplementedException();
}
public IQueue GetQueue(string name)
{
throw new NotImplementedException();
}
public ITopic GetTopic(string name)
{
throw new NotImplementedException();
}
public ITemporaryQueue CreateTemporaryQueue()
{
throw new NotImplementedException();
}
public ITemporaryTopic CreateTemporaryTopic()
{
throw new NotImplementedException();
}
public IMessage CreateMessage()
{
throw new NotImplementedException();
}
public ITextMessage CreateTextMessage()
{
throw new NotImplementedException();
}
public ITextMessage CreateTextMessage(string text)
{
throw new NotImplementedException();
}
public IMapMessage CreateMapMessage()
{
throw new NotImplementedException();
}
public IObjectMessage CreateObjectMessage(object body)
{
throw new NotImplementedException();
}
public IBytesMessage CreateBytesMessage()
{
throw new NotImplementedException();
}
public IBytesMessage CreateBytesMessage(byte[] body)
{
throw new NotImplementedException();
}
public void Close()
{
closeCount++;
}
public void Commit()
{
throw new NotImplementedException();
}
public void Rollback()
{
throw new NotImplementedException();
}
public bool Transacted
{
get { throw new NotImplementedException(); }
}
public AcknowledgementMode AcknowledgementMode
{
get { throw new NotImplementedException(); }
}
public void Dispose()
{
throw new NotImplementedException();
}
}
}

10
test/Spring/Spring.Messaging.Nms.Tests/Spring.Messaging.Nms.Tests.2005.csproj
View File

@@ -74,9 +74,19 @@
<Project>{AEB1578C-9018-4D49-B440-789F38DD2F29}</Project>
<Name>Spring.Messaging.Nms.2005</Name>
</ProjectReference>
<ProjectReference Include="..\Spring.Core.Tests\Spring.Core.Tests.2005.csproj">
<Project>{44B16BAA-6DF8-447C-9D7F-3AD3D854D904}</Project>
<Name>Spring.Core.Tests.2005</Name>
</ProjectReference>
</ItemGroup>
<ItemGroup>
<Compile Include="Messaging\Nms\Connections\CachingConnectionFactoryTests.cs" />
<Compile Include="Messaging\Nms\Connections\NmsTransactionManagerTests.cs" />
<Compile Include="Messaging\Nms\Connections\SingleConnectionFactoryTests.cs" />
<Compile Include="Messaging\Nms\Connections\TestConnection.cs" />
<Compile Include="Messaging\Nms\Connections\TestExceptionListener.cs" />
<Compile Include="Messaging\Nms\Connections\TestMessageProducer.cs" />
<Compile Include="Messaging\Nms\Connections\TestSession.cs" />
</ItemGroup>
<ItemGroup>
<Content Include="Spring.Messaging.Nms.Tests.dll.config">