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

SPRNET-1233 - Support for sending to remote private queue.

Browse Source 
Changes to MessageQueueTemplate
1. Registers MessageQueueMetadataCache in ApplicationContext if not set explicitly.  
2. Logs warnings if MetadataCache property is set to null, for example in stand alone API usage.  
3. No longer has DefaultMessageQueueObjectName as a required .

DefaultMessageQueueFactory updates MessageQueueMetadataCache in ApplicationContext when a new MessageQueue is registered.
This commit is contained in:
markpollack
2009-07-25 05:05:43 +00:00
parent c217365748
commit 64df937a58
13 changed files with 128 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/Spring/Spring.Messaging/Messaging/Core/DefaultMessageQueueFactory.cs
View File

@@ -66,6 +66,11 @@ namespace Spring.Messaging.Core
MessageQueueFactoryObject mqfo = new MessageQueueFactoryObject();
mqfo.MessageCreatorDelegate = messageQueueCreatorDelegate;
applicationContext.ObjectFactory.RegisterSingleton(messageQueueObjectName, mqfo);
IDictionary caches = applicationContext.GetObjectsOfType(typeof(MessageQueueMetadataCache));
foreach (DictionaryEntry entry in caches)
{
((MessageQueueMetadataCache) entry.Value).Insert(mqfo.Path, new MessageQueueMetadata(mqfo.RemoteQueue, mqfo.RemoteQueueIsTransactional));
}
}
/// <summary>

89
src/Spring/Spring.Messaging/Messaging/Core/MessageQueueTemplate.cs
View File

@@ -27,6 +27,7 @@ using Spring.Messaging.Support;
using Spring.Messaging.Support.Converters;
using Spring.Objects.Factory;
using Spring.Objects.Factory.Config;
using Spring.Util;
namespace Spring.Messaging.Core
{
@@ -82,12 +83,14 @@ namespace Spring.Messaging.Core
private string messageConverterObjectName;
private IMessageQueueFactory messageQueueFactory;
protected IApplicationContext applicationContext;
protected IConfigurableApplicationContext applicationContext;
private TimeSpan timeout = MessageQueue.InfiniteTimeout;
private MessageQueueMetadataCache metadataCache;
public const string METADATA_CACHE_NAME = "__MessageQueueMetadataCache__";
#endregion
#region Constructors
@@ -241,7 +244,17 @@ namespace Spring.Messaging.Core
public IApplicationContext ApplicationContext
{
get { return applicationContext; }
set { applicationContext = value; }
set {
AssertUtils.ArgumentNotNull(value, "An ApplicationContext instance is required");
IConfigurableApplicationContext ctx = value as IConfigurableApplicationContext;
if (ctx == null)
{
throw new InvalidOperationException(
"Implementations of IApplicationContext must also implement IConfigurableApplicationContext");
}
applicationContext = ctx;
}
}
#endregion
@@ -263,12 +276,10 @@ namespace Spring.Messaging.Core
/// </remarks>
public void AfterPropertiesSet()
{
if (DefaultMessageQueueObjectName == null)
{
throw new ArgumentException("DefaultMessageQueueObjectName is required.");
}
if (MessageQueueFactory == null)
{
AssertUtils.ArgumentNotNull(applicationContext, "MessageQueueTemplate requires the ApplicationContext property to be set if the MessageQueueFactory property is not set for automatic create of the DefaultMessageQueueFactory");
DefaultMessageQueueFactory mqf = new DefaultMessageQueueFactory();
mqf.ApplicationContext = applicationContext;
messageQueueFactory = mqf;
@@ -278,12 +289,43 @@ namespace Spring.Messaging.Core
messageConverterObjectName = QueueUtils.RegisterDefaultMessageConverter(applicationContext);
}
//If it has not been set by the user explicitly, then initialize.
CreateDefaultMetadataCache();
}
protected virtual void CreateDefaultMetadataCache()
{
if (metadataCache == null)
{
metadataCache = new MessageQueueMetadataCache();
metadataCache.ApplicationContext = ApplicationContext;
metadataCache.AfterPropertiesSet();
metadataCache.Initialize();
if (applicationContext.ContainsObject(METADATA_CACHE_NAME))
{
metadataCache = applicationContext.GetObject(METADATA_CACHE_NAME) as MessageQueueMetadataCache;
}
else
{
metadataCache = new MessageQueueMetadataCache();
if (ApplicationContext != null)
{
metadataCache.ApplicationContext = ApplicationContext;
metadataCache.AfterPropertiesSet();
metadataCache.Initialize();
applicationContext.ObjectFactory.RegisterSingleton("__MessageQueueMetadataCache__",
metadataCache);
}
else
{
#region Logging
if (LOG.IsWarnEnabled)
{
LOG.Warn(
"The ApplicationContext property has not been set, so the MessageQueueMetadataCache can not be automatically generated. " +
"Please explictly set the MessageQueueMetadataCache using the property MetadataCache or set the ApplicationContext property. " +
"This will only effect the use of MessageQueueTemplate when publishing to remote queues.");
}
#endregion
}
}
}
}
@@ -500,19 +542,30 @@ namespace Spring.Messaging.Core
protected virtual void DoSendMessageQueue(MessageQueue mq, Message msg)
{
MessageQueueTransaction transactionToUse = QueueUtils.GetMessageQueueTransaction(null);
MessageQueueMetadata mqMetadata = metadataCache.Get(mq.Path);
if (mqMetadata != null)
if (metadataCache != null)
{
if (mqMetadata.RemoteQueue)
MessageQueueMetadata mqMetadata = metadataCache.Get(mq.Path);
if (mqMetadata != null)
{
if (mqMetadata.RemoteQueueIsTransactional)
if (mqMetadata.RemoteQueue)
{
// DefaultMessageQueue transaction is externally managed.
DoSendMessageTransaction(mq, transactionToUse, msg);
if (mqMetadata.RemoteQueueIsTransactional)
{
// DefaultMessageQueue transaction is externally managed.
DoSendMessageTransaction(mq, transactionToUse, msg);
return;
}
DoSendMessageQueueNonTransactional(mq, transactionToUse, msg);
return;
}
DoSendMessageQueueNonTransactional(mq, transactionToUse, msg);
return;
}
} else
{
if (LOG.IsWarnEnabled)
{
LOG.Warn("MetadataCache has not been initialized. Set the MetadataCache explicitly in standalone usage and/or " +
"configure the MessageQueueTemplate in an ApplicationContext. If deployed in an ApplicationContext by default " +
"the MetadataCache will automaticaly populated.");
}
}
// Handle assuming these are local queues.

5
src/Spring/Spring.Messaging/Messaging/Listener/AbstractPeekingMessageListenerContainer.cs
View File

@@ -445,11 +445,10 @@ namespace Spring.Messaging.Listener
/// messageQueue.Receive().
/// </summary>
/// <remarks>It allows subclasses to modify the state of the MessageQueue
/// before receiving which maybe required when using remote queues, for example
/// to set a MessageFormatter.</remarks>
/// before receiving which maybe required when using remote queues</remarks>
/// <param name="messageQueue"></param>
protected virtual void BeforeMessageReceived(MessageQueue messageQueue)
{
{
}
#endregion

23
src/Spring/Spring.Messaging/Messaging/Listener/MessageListenerAdapter.cs
View File

@@ -230,6 +230,12 @@ namespace Spring.Messaging.Listener
{
messageConverterObjectName = QueueUtils.RegisterDefaultMessageConverter(applicationContext);
}
if (messageQueueTemplate == null)
{
messageQueueTemplate = new MessageQueueTemplate();
messageQueueTemplate.ApplicationContext = ApplicationContext;
messageQueueTemplate.AfterPropertiesSet();
}
}
#endregion
@@ -351,6 +357,18 @@ namespace Spring.Messaging.Listener
}
}
/// <summary>
/// Sets the message queue template.
/// </summary>
/// <remarks>If not set, will create one for it own internal use whne MessageListenerAdapter is constructed.
/// It maybe useful to share an existing instance if you have an extensively configured MessageQueueTemplate.
/// </remarks>
/// <value>The message queue template.</value>
public MessageQueueTemplate MessageQueueTemplate
{
set { messageQueueTemplate = value; }
}
#region IMessageListener Members
/// <summary>
@@ -395,7 +413,6 @@ namespace Spring.Messaging.Listener
protected virtual void InitDefaultStrategies()
{
processingExpression = Expression.Parse(defaultHandlerMethod + "(#convertedObject)");
messageQueueTemplate = new MessageQueueTemplate();
}
/// <summary>
@@ -435,6 +452,10 @@ namespace Spring.Messaging.Listener
protected virtual void SendResponse(MessageQueue destination, Message response)
{
//Will send with appropriate transaction semantics
if (logger.IsDebugEnabled)
{
logger.Debug("Sending response message to path = [" + destination.Path + "]");
}
messageQueueTemplate.Send(destination, response);
}

4
src/Spring/Spring.Messaging/Messaging/Listener/NonTransactionalMessageListenerContainer.cs
View File

@@ -73,7 +73,7 @@ namespace Spring.Messaging.Listener
/// Perform a receive opertion on the message queue and execute the
/// message listener
/// </summary>
/// <param name="mq">The DefaultMessageQueue.</param>
/// <param name="mq">The MessageQueue.</param>
/// <returns>
/// true if received a message, false otherwise
/// </returns>
@@ -119,7 +119,7 @@ namespace Spring.Messaging.Listener
if (LOG.IsErrorEnabled)
{
LOG.Error("Error receiving message from DefaultMessageQueue [" + mq.Path +
LOG.Error("Error receiving message from MessageQueue [" + mq.Path +
"], closing queue and clearing connection cache.");
}

2
test/Spring/Spring.Messaging.Tests/Messaging/Core/MessageQueueMetadataCacheTests.cs
View File

@@ -36,7 +36,7 @@ namespace Spring.Messaging.Core
{
protected override string[] ConfigLocations
{
get { return new[] {"assembly://Spring.Messaging.Tests/Spring.Messaging.Core/MessageQueueTemplateTests.xml"}; }
get { return new[] {"assembly://Spring.Messaging.Tests/Spring.Messaging.Core/MessageQueueMetadataCacheTests.xml"}; }
}
[Test]

10
test/Spring/Spring.Messaging.Tests/Messaging/Core/MessageQueueMetadataCacheTests.xml
View File

@@ -36,14 +36,4 @@
</object>
<object id='msqueue' type='Spring.Messaging.Support.MessageQueueFactoryObject, Spring.Messaging'>
<property name='Path' value='.\Private$\testqueue'/>
<property name='ProductTemplate'>
<object>
<property name='Label' value='MyLabel'/>
</object>
</property>
</object>
</objects>

2
test/Spring/Spring.Messaging.Tests/Messaging/Core/MessageQueueTemplateTests.cs
View File

@@ -116,7 +116,7 @@ namespace Spring.Messaging.Core
[Test]
[ExpectedException(typeof (ArgumentException), ExpectedMessage = "DefaultMessageQueueObjectName is required.")]
[ExpectedException(typeof (ArgumentNullException))]
public void NoMessageQueueNameSpecified()
{
MessageQueueTemplate mqt = new MessageQueueTemplate();

1
test/Spring/Spring.Messaging.Tests/Messaging/Core/MessageQueueTemplateTests.xml
View File

@@ -34,6 +34,7 @@
<property name='RemoteQueue' value="true"/>
</object>
<!-- MessageQueueTemplate -->
<object id="queue" type="Spring.Messaging.Core.MessageQueueTemplate, Spring.Messaging">
<property name="DefaultMessageQueueObjectName" value="testtxqueue"/>

2
test/Spring/Spring.Messaging.Tests/Messaging/Listener/DistributedTxMessageListenerContainerTests.xml
View File

@@ -77,7 +77,7 @@
</object>
<object id="distributedTransactionExceptionHandler" type="Spring.Messaging.Listener.SendToQueueDistributedTransactionExceptionHandler, Spring.Messaging">
<property name="DefaultMessageQueueObjectName" value="msmqtxretryqueue"/>
<property name="MessageQueueObjectName" value="msmqtxretryqueue"/>
<property name="MaxRetry" value="2"/>
</object>

11
test/Spring/Spring.Messaging.Tests/Messaging/Listener/NonTransactionalMessageListenerContainerTests.cs
View File

@@ -84,19 +84,20 @@ namespace Spring.Messaging.Listener
[Test]
public void SendAndAsyncReceive()
{
//MessageQueueTemplate q = applicationContext["testQueueTemplate"] as MessageQueueTemplate;
MessageQueueTemplate q = applicationContext["testRemoteTemplate"] as MessageQueueTemplate;
Assert.IsNotNull(q);
/*
q.ConvertAndSend("Hello World 1");
q.ConvertAndSend("Hello World 2");
q.ConvertAndSend("Hello World 3");
q.ConvertAndSend("Hello World 4");
q.ConvertAndSend("Hello World 5");
*/
Assert.AreEqual(0, listener.MessageCount);
container.Start();
@@ -107,7 +108,7 @@ namespace Spring.Messaging.Listener
container.Stop();
container.Shutdown();
Thread.Sleep(2500);
Thread.Sleep(2500);
}

19
test/Spring/Spring.Messaging.Tests/Messaging/Listener/NonTransactionalMessageListenerContainerTests.xml
View File

@@ -40,34 +40,45 @@
<object id="testRemoteTemplate" type="Spring.Messaging.Core.MessageQueueTemplate, Spring.Messaging">
<property name="DefaultMessageQueueObjectName" value="testremotequeue"/>
<property name="MessageConverterObjectName" value="messageConverter"/>
<property name="MessageConverterObjectName" value="binaryMessageConverter"/>
</object>
<object id="testQueueTemplate" type="Spring.Messaging.Core.MessageQueueTemplate, Spring.Messaging">
<property name="DefaultMessageQueueObjectName" value="msmqTestQueue"/>
<property name="MessageConverterObjectName" value="messageConverter"/>
<property name="MessageConverterObjectName" value="binaryMessageConverter"/>
</object>
<!-- Message Converters -->
<object id="messageConverter" type="Spring.Messaging.Support.Converters.XmlMessageConverter, Spring.Messaging"
singleton="false">
<property name="TargetTypes" value="System.String"/>
</object>
<object id="binaryMessageConverter" type="Spring.Messaging.Support.Converters.BinaryMessageConverter, Spring.Messaging">
</object>
<object id="nonTransactionalMessageListenerContainer" type="Spring.Messaging.Listener.NonTransactionalMessageListenerContainer, Spring.Messaging">
<property name="MessageQueueObjectName" value="testremotequeue"/>
<property name="MaxConcurrentListeners" value="2"/>
<property name="MaxConcurrentListeners" value="1"/>
<property name="ListenerTimeLimit" value="20s"/>
<property name="MessageListener" ref="messageListenerAdapter"/>
<property name="ExceptionHandler" ref="exceptionHandler"/>
<property name="AutoStartup" value="false"/>
</object>
<!-- Adapter -->
<object id="messageListenerAdapter" type="Spring.Messaging.Listener.MessageListenerAdapter, Spring.Messaging">
<property name="DefaultResponseQueueName" value="msmqTestResponseQueue"/>
<property name="MessageConverterObjectName" value="messageConverter"/>
<property name="MessageConverterObjectName" value="binaryMessageConverter"/>
<property name="HandlerObject" ref="simpleHandler"/>
</object>
<!-- Message and Exception Handlers -->
<object id="simpleHandler" type="Spring.Messaging.Listener.SimpleHandler, Spring.Messaging.Tests">
</object>

1
test/Spring/Spring.Messaging.Tests/Messaging/Listener/SimpleExceptionHandler.cs
View File

@@ -28,6 +28,7 @@ namespace Spring.Messaging.Listener
public void OnException(Exception exception, Message message)
{
LOG.Error("Exception Handler processing message id = [" + message.Id + "]");
LOG.Error("Exception = ", exception);
messageCount++;
}