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

Add queue browsing functionality to TIBCO EmsTemplate

Browse Source
This commit is contained in:
markpollack
2008-08-14 00:41:21 +00:00
parent 4018ddf263
commit 6e72dc997b
5 changed files with 550 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/BrowserDelegate.cs Normal file
View File

@@ -0,0 +1,31 @@
#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 TIBCO.EMS;
namespace Spring.Messaging.Ems.Core
{
/// <summary>
/// Delegate callback for browsing the messages in an EMS queue.
/// </summary>
public delegate object BrowserDelegate(Session session, QueueBrowser browser);
}

335
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/EmsTemplate.cs
View File

@@ -28,6 +28,7 @@ using Spring.Messaging.Ems.Support.Destinations;
using Spring.Transaction.Support;
using Spring.Util;
using TIBCO.EMS;
using Queue=TIBCO.EMS.Queue;
namespace Spring.Messaging.Ems.Core
{
@@ -164,8 +165,30 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the result object from working with the session
/// </returns>
/// <throws> EMSException if there is any problem </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public virtual object Execute(ISessionCallback action, bool startConnection)
{
return Execute(action.DoInEms, startConnection);
}
/// <summary> Execute the action specified by the given action object within a
/// EMS Session.
/// </summary>
/// <remarks> Generalized version of <code>execute(SessionCallback)</code>,
/// allowing the EMS Connection to be started on the fly.
/// <p>Use <code>execute(SessionCallback)</code> for the general case.
/// Starting the EMS Connection is just necessary for receiving messages,
/// which is preferably achieved through the <code>receive</code> methods.</p>
/// </remarks>
/// <param name="action">callback object that exposes the session
/// </param>
/// <param name="startConnection">Start the connection before performing callback action.
/// </param>
/// <returns> the result object from working with the session
/// </returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public virtual object Execute(SessionDelegate action, bool startConnection)
{
AssertUtils.ArgumentNotNull(action, "Callback object must not be null");
@@ -195,7 +218,7 @@ namespace Spring.Messaging.Ems.Core
{
logger.Debug("Executing callback on EMS Session [" + sessionToUse + "]");
}
return action.DoInEms(sessionToUse);
return action(sessionToUse);
}
finally
{
@@ -433,7 +456,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the new EMS MessageProducer
/// </returns>
/// <throws> EMSException if thrown by EMS API methods </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
/// <seealso cref="DoCreateProducer">
/// </seealso>
/// <seealso cref="MessageIdEnabled">
@@ -487,7 +510,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the new EMS MessageProducer
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
protected virtual MessageProducer DoCreateProducer(Session session, Destination destination)
{
if (CacheEmsResources)
@@ -543,7 +566,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the new EMS MessageConsumer
/// </returns>
/// <throws> EMSException if thrown by EMS API methods </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
protected virtual MessageConsumer CreateConsumer(Session session, Destination destination,
string messageSelector)
{
@@ -568,7 +591,7 @@ namespace Spring.Messaging.Ems.Core
/// </remarks>
/// <returns>A EMS Connection
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
protected override Connection CreateConnection()
{
if (CacheEmsResources)
@@ -596,7 +619,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the new EMS Session
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
protected override Session CreateSession(Connection con)
{
if (CacheEmsResources)
@@ -646,7 +669,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="messageCreatorDelegate">delegate callback to create a EMS Message
/// </param>
/// <throws>EMSException if thrown by EMS API methods </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
protected internal virtual void DoSend(Session session, Destination destination, IMessageCreator messageCreator,
MessageCreatorDelegate messageCreatorDelegate)
{
@@ -656,7 +679,7 @@ namespace Spring.Messaging.Ems.Core
try
{
Message message = null;
Message message;
if (messageCreator != null)
{
message = messageCreator.CreateMessage(session) ;
@@ -689,7 +712,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="message">the EMS Message to send
/// </param>
/// <throws> EMSException if thrown by EMS API methods </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
protected virtual void DoSend(MessageProducer producer, Message message)
{
if (ExplicitQosEnabled)
@@ -720,7 +743,7 @@ namespace Spring.Messaging.Ems.Core
/// If PubSubDomain equals true, then a Session is passed to the callback.
/// If false, then a Session is passed to the callback.</para>b
/// </remarks>
/// <throws>EMSException if there is any problem </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object Execute(SessionDelegate del)
{
return Execute(new ExecuteSessionCallbackUsingDelegate(del));
@@ -736,7 +759,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the result object from working with the session
/// </returns>
/// <throws>EMSException if there is any problem </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object Execute(ISessionCallback action)
{
return Execute(action, false);
@@ -750,7 +773,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the result object from working with the session
/// </returns>
/// <throws>EMSException if there is any problem </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object Execute(IProducerCallback action)
{
return Execute(new ProducerCreatorCallback(this, action));
@@ -764,7 +787,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the result object from working with the session
/// </returns>
/// <throws>EMSException if there is any problem </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object Execute(ProducerDelegate del)
{
return Execute(new ProducerCreatorCallback(this, del));
@@ -775,7 +798,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <param name="messageCreatorDelegate">delegate callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void SendWithDelegate(MessageCreatorDelegate messageCreatorDelegate)
{
CheckDefaultDestination();
@@ -796,7 +819,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="messageCreatorDelegate">delegate callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void SendWithDelegate(Destination destination, MessageCreatorDelegate messageCreatorDelegate)
{
Execute(new SendDestinationCallback(this, destination, messageCreatorDelegate), false);
@@ -810,7 +833,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="messageCreatorDelegate">delegate callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void SendWithDelegate(string destinationName, MessageCreatorDelegate messageCreatorDelegate)
{
Execute(new SendDestinationCallback(this, destinationName, messageCreatorDelegate), false);
@@ -821,7 +844,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <param name="messageCreator">callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void Send(IMessageCreator messageCreator)
{
CheckDefaultDestination();
@@ -842,7 +865,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="messageCreator">callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void Send(Destination destination, IMessageCreator messageCreator)
{
@@ -856,7 +879,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="messageCreator">callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void Send(string destinationName, IMessageCreator messageCreator)
{
Execute(new SendDestinationCallback(this, destinationName, messageCreator), false);
@@ -867,7 +890,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <param name="message">the object to convert to a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void ConvertAndSend(object message)
{
CheckDefaultDestination();
@@ -888,7 +911,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="message">the object to convert to a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void ConvertAndSend(Destination destination, object message)
{
CheckMessageConverter();
@@ -903,7 +926,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="message">the object to convert to a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void ConvertAndSend(string destinationName, object message)
{
Send(destinationName, new SimpleMessageCreator(this, message));
@@ -918,7 +941,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="postProcessor">the callback to modify the message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void ConvertAndSend(object message, IMessagePostProcessor postProcessor)
{
CheckDefaultDestination();
@@ -942,7 +965,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="postProcessor">the callback to modify the message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void ConvertAndSend(Destination destination, object message, IMessagePostProcessor postProcessor)
{
CheckMessageConverter();
@@ -961,7 +984,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="postProcessor">the callback to modify the message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void ConvertAndSend(string destinationName, object message, IMessagePostProcessor postProcessor)
{
CheckMessageConverter();
@@ -975,7 +998,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <param name="message">the object to convert to a message</param>
/// <param name="postProcessor">the callback to modify the message</param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void ConvertAndSendWithDelegate(object message, MessagePostProcessorDelegate postProcessor)
{
//Execute(new SendDestinationCallback(this, destination, messageCreatorDelegate), false);
@@ -998,7 +1021,7 @@ namespace Spring.Messaging.Ems.Core
/// <param name="destination">the destination to send this message to</param>
/// <param name="message">the object to convert to a message</param>
/// <param name="postProcessor">the callback to modify the message</param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void ConvertAndSendWithDelegate(Destination destination, object message,
MessagePostProcessorDelegate postProcessor)
{
@@ -1015,7 +1038,7 @@ namespace Spring.Messaging.Ems.Core
/// (to be resolved to an actual destination by a DestinationResolver)</param>
/// <param name="message">the object to convert to a message.</param>
/// <param name="postProcessor">the callback to modify the message</param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public void ConvertAndSendWithDelegate(string destinationName, object message,
MessagePostProcessorDelegate postProcessor)
{
@@ -1032,7 +1055,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public Message Receive()
{
CheckDefaultDestination();
@@ -1055,7 +1078,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public Message Receive(Destination destination)
{
return Execute(new ReceiveCallback(this, destination)) as Message;
@@ -1072,7 +1095,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public Message Receive(string destinationName)
{
return Execute(new ReceiveCallback(this, destinationName)) as Message;
@@ -1089,7 +1112,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public Message ReceiveSelected(string messageSelector)
{
CheckDefaultDestination();
@@ -1115,7 +1138,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public Message ReceiveSelected(Destination destination, string messageSelector)
{
return Execute(new ReceiveSelectedCallback(this, destination, messageSelector), true) as Message;
@@ -1134,7 +1157,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public Message ReceiveSelected(string destinationName, string messageSelector)
{
return Execute(new ReceiveSelectedCallback(this, destinationName, messageSelector), true) as Message;
@@ -1208,7 +1231,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object ReceiveAndConvert()
{
CheckMessageConverter();
@@ -1225,7 +1248,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object ReceiveAndConvert(Destination destination)
{
CheckMessageConverter();
@@ -1244,7 +1267,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object ReceiveAndConvert(string destinationName)
{
CheckMessageConverter();
@@ -1263,11 +1286,11 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object ReceiveSelectedAndConvert(string messageSelector)
{
CheckMessageConverter();
return DoConvertFromMessage(ReceiveSelected(messageSelector)); ;
return DoConvertFromMessage(ReceiveSelected(messageSelector));
}
/// <summary> Receive a message synchronously from the specified destination, but only
@@ -1283,7 +1306,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object ReceiveSelectedAndConvert(Destination destination, string messageSelector)
{
CheckMessageConverter();
@@ -1304,15 +1327,243 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object ReceiveSelectedAndConvert(string destinationName, string messageSelector)
{
CheckMessageConverter();
return DoConvertFromMessage(ReceiveSelected(destinationName, messageSelector));
}
/// <summary>
/// Browses messages in the default EMS queue. The callback gives access to the EMS
/// Session and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <returns>the result object from working with the session</returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object Browse(IBrowserCallback action)
{
AssertUtils.ArgumentNotNull(action, "action");
return BrowseWithDelegate(action.DoInEms);
}
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queue">The queue to browse.</param>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <returns>the result object from working with the session</returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object Browse(Queue queue, IBrowserCallback action)
{
AssertUtils.ArgumentNotNull(action, "action");
return BrowseSelectedWithDelegate(queue, null, action.DoInEms);
}
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queueName">Name of the queue to browse,
/// (to be resolved to an actual destination by a DestinationResolver)</param>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object Browse(string queueName, IBrowserCallback action)
{
AssertUtils.ArgumentNotNull(action, "action");
return BrowseSelectedWithDelegate(queueName, null, action.DoInEms);
}
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object BrowseSelected(string messageSelector, IBrowserCallback action)
{
AssertUtils.ArgumentNotNull(action, "action");
return BrowseSelectedWithDelegate(messageSelector, action.DoInEms);
}
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queue">The queue to browse.</param>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <returns></returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object BrowseSelected(Queue queue, string messageSelector, IBrowserCallback action)
{
AssertUtils.ArgumentNotNull(action, "action");
return BrowseSelectedWithDelegate(queue, messageSelector, action.DoInEms);
}
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queueName">Name of the queue to browse,
/// (to be resolved to an actual destination by a DestinationResolver)</param>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <returns></returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object BrowseSelected(string queueName, string messageSelector, IBrowserCallback action)
{
AssertUtils.ArgumentNotNull(action, "action");
return BrowseSelectedWithDelegate(queueName, messageSelector, action.DoInEms);
}
/// <summary>
/// Browses messages in the default EMS queue. The callback gives access to the EMS
/// Session and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <returns>the result object from working with the session</returns>
public object BrowseWithDelegate(BrowserDelegate action)
{
if (DefaultDestinationName != null)
{
return BrowseSelectedWithDelegate(DefaultDestinationName, action);
}
else
{
Destination destination = DefaultDestination;
if (!(destination is Queue))
{
throw new InvalidOperationException("defaultDestination does not correspond to a Queue. Check configuration of NmsTemplate.");
}
return BrowseWithDelegate((Queue)destination, action);
}
}
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queue">The queue to browse.</param>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <returns>the result object from working with the session</returns>
public object BrowseWithDelegate(Queue queue, BrowserDelegate action)
{
return BrowseSelectedWithDelegate(queue, null, action);
}
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queueName">Name of the queue to browse,
/// (to be resolved to an actual destination by a DestinationResolver)</param>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object BrowseWithDelegate(string queueName, BrowserDelegate action)
{
return BrowseSelectedWithDelegate(queueName, null, action);
}
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object BrowseSelectedWithDelegate(string messageSelector, BrowserDelegate action)
{
if (DefaultDestinationName != null)
{
return BrowseSelectedWithDelegate(DefaultDestinationName, messageSelector, action);
}
else
{
Destination destination = DefaultDestination;
if (!(destination is Queue))
{
throw new InvalidOperationException("defaultDestination does not correspond to a Queue. Check configuration of NmsTemplate.");
}
return BrowseSelectedWithDelegate((Queue)destination, messageSelector, action);
}
}
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queue">The queue to browse.</param>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <returns></returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object BrowseSelectedWithDelegate(Queue queue, string messageSelector, BrowserDelegate action)
{
AssertUtils.ArgumentNotNull(action, "action");
return Execute(delegate(Session session)
{
QueueBrowser browser = CreateBrowser(session, queue, messageSelector);
try
{
return action(session, browser);
}
finally
{
EmsUtils.CloseQueueBrowser(browser);
}
}, true);
}
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queueName">Name of the queue to browse,
/// (to be resolved to an actual destination by a DestinationResolver)</param>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <returns></returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
public object BrowseSelectedWithDelegate(string queueName, string messageSelector, BrowserDelegate action)
{
AssertUtils.ArgumentNotNull(action, "action");
return Execute(delegate(Session session)
{
Queue queue = (Queue)DestinationResolver.ResolveDestinationName(session, queueName, false);
QueueBrowser browser = CreateBrowser(session, queue, messageSelector);
try
{
return action(session, browser);
}
finally
{
EmsUtils.CloseQueueBrowser(browser);
}
}, true);
}
#endregion
/// <summary>
/// Creates the queue browser.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="queue">The queue.</param>
/// <param name="selector">The selector.</param>
/// <returns>A new queue browser</returns>
protected virtual QueueBrowser CreateBrowser(Session session, Queue queue, string selector)
{
return session.CreateBrowser(queue, selector);
}
#region Supporting Internal Classes
/// <summary>

47
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IBrowserCallback.cs Normal file
View File

@@ -0,0 +1,47 @@
#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 TIBCO.EMS;
namespace Spring.Messaging.Ems.Core
{
/// <summary>
/// Callback for browsing the messages in an EMS queue.
/// </summary>
/// <remarks>
/// To be used with EmsTemplate's callback methods that take a IBrowserCallback argument
/// </remarks>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public interface IBrowserCallback
{
/// <summary>
/// Perform operations on the given Session and QueueBrowser
/// </summary>
/// <param name="session">The session.</param>
/// <param name="browser">The browser.</param>
/// <returns>The object from working with the Session and QueueBrowser, may be null</returns>
/// <exception cref="EMSException">If there is any problem when accessing EMS API</exception>
object DoInEms(Session session, QueueBrowser browser);
}
}

187
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IEmsOperations.cs
View File

@@ -45,7 +45,7 @@ namespace Spring.Messaging.Ems.Core
/// <param name="del">delegate that exposes the session</param>
/// <returns> the result object from working with the session
/// </returns>
/// <throws>EMSException if there is any problem </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object Execute(SessionDelegate del);
/// <summary> Execute the action specified by the given action object within
@@ -55,7 +55,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the result object from working with the session
/// </returns>
/// <throws>EMSException if there is any problem </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object Execute(ISessionCallback action);
/// <summary> Send a message to a EMS destination. The callback gives access to
@@ -77,7 +77,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the result object from working with the session
/// </returns>
/// <throws>EMSException if there is any problem </throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object Execute(ProducerDelegate del);
@@ -90,7 +90,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <param name="messageCreator">callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void Send(IMessageCreator messageCreator);
/// <summary> Send a message to the specified destination.
@@ -100,7 +100,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="messageCreator">callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void Send(Destination destination, IMessageCreator messageCreator);
/// <summary> Send a message to the specified destination.
@@ -111,7 +111,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="messageCreator">callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void Send(string destinationName, IMessageCreator messageCreator);
//-------------------------------------------------------------------------
@@ -123,7 +123,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <param name="messageCreatorDelegate">delegate callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void SendWithDelegate(MessageCreatorDelegate messageCreatorDelegate);
/// <summary> Send a message to the specified destination.
@@ -133,7 +133,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="messageCreatorDelegate">delegate callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void SendWithDelegate(Destination destination, MessageCreatorDelegate messageCreatorDelegate);
/// <summary> Send a message to the specified destination.
@@ -144,7 +144,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="messageCreatorDelegate">delegate callback to create a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void SendWithDelegate(string destinationName, MessageCreatorDelegate messageCreatorDelegate);
//-------------------------------------------------------------------------
@@ -157,7 +157,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <param name="message">the object to convert to a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void ConvertAndSend(object message);
/// <summary> Send the given object to the specified destination, converting the object
@@ -167,7 +167,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="message">the object to convert to a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void ConvertAndSend(Destination destination, object message);
/// <summary> Send the given object to the specified destination, converting the object
@@ -178,7 +178,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="message">the object to convert to a message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void ConvertAndSend(string destinationName, object message);
/// <summary> Send the given object to the default destination, converting the object
@@ -190,7 +190,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="postProcessor">the callback to modify the message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void ConvertAndSend(object message, IMessagePostProcessor postProcessor);
/// <summary> Send the given object to the specified destination, converting the object
@@ -203,7 +203,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="postProcessor">the callback to modify the message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void ConvertAndSend(Destination destination, object message, IMessagePostProcessor postProcessor);
/// <summary> Send the given object to the specified destination, converting the object
@@ -217,7 +217,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <param name="postProcessor">the callback to modify the message
/// </param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void ConvertAndSend(string destinationName, object message, IMessagePostProcessor postProcessor);
@@ -229,7 +229,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <param name="message">the object to convert to a message</param>
/// <param name="postProcessor">the callback to modify the message</param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void ConvertAndSendWithDelegate(object message, MessagePostProcessorDelegate postProcessor);
/// <summary>
@@ -240,7 +240,7 @@ namespace Spring.Messaging.Ems.Core
/// <param name="destination">the destination to send this message to</param>
/// <param name="message">the object to convert to a message</param>
/// <param name="postProcessor">the callback to modify the message</param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void ConvertAndSendWithDelegate(Destination destination, object message, MessagePostProcessorDelegate postProcessor);
/// <summary>
@@ -252,7 +252,7 @@ namespace Spring.Messaging.Ems.Core
/// (to be resolved to an actual destination by a DestinationResolver)</param>
/// <param name="message">the object to convert to a message.</param>
/// <param name="postProcessor">the callback to modify the message</param>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
void ConvertAndSendWithDelegate(string destinationName, object message, MessagePostProcessorDelegate postProcessor);
@@ -268,7 +268,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
Message Receive();
/// <summary> Receive a message synchronously from the specified destination, but only
@@ -280,7 +280,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
Message Receive(Destination destination);
/// <summary> Receive a message synchronously from the specified destination, but only
@@ -293,7 +293,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
Message Receive(string destinationName);
/// <summary> Receive a message synchronously from the default destination, but only
@@ -307,7 +307,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
Message ReceiveSelected(string messageSelector);
/// <summary> Receive a message synchronously from the specified destination, but only
@@ -322,7 +322,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
Message ReceiveSelected(Destination destination, string messageSelector);
/// <summary> Receive a message synchronously from the specified destination, but only
@@ -338,7 +338,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message received by the consumer, or <code>null</code> if the timeout expires
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
Message ReceiveSelected(string destinationName, string messageSelector);
@@ -355,7 +355,7 @@ namespace Spring.Messaging.Ems.Core
/// </summary>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object ReceiveAndConvert();
/// <summary> Receive a message synchronously from the specified destination, but only
@@ -368,7 +368,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object ReceiveAndConvert(Destination destination);
/// <summary> Receive a message synchronously from the specified destination, but only
@@ -382,7 +382,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object ReceiveAndConvert(string destinationName);
/// <summary> Receive a message synchronously from the default destination, but only
@@ -397,7 +397,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object ReceiveSelectedAndConvert(string messageSelector);
/// <summary> Receive a message synchronously from the specified destination, but only
@@ -413,7 +413,7 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object ReceiveSelectedAndConvert(Destination destination, string messageSelector);
/// <summary> Receive a message synchronously from the specified destination, but only
@@ -430,7 +430,134 @@ namespace Spring.Messaging.Ems.Core
/// </param>
/// <returns> the message produced for the consumer or <code>null</code> if the timeout expires.
/// </returns>
/// <throws>EMSException if there is any problem</throws>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object ReceiveSelectedAndConvert(string destinationName, string messageSelector);
/// <summary>
/// Browses messages in the default EMS queue. The callback gives access to the EMS
/// Session and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <returns>the result object from working with the session</returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object Browse(IBrowserCallback action);
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queue">The queue to browse.</param>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <returns>the result object from working with the session</returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object Browse(Queue queue, IBrowserCallback action);
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queueName">Name of the queue to browse,
/// (to be resolved to an actual destination by a DestinationResolver)</param>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object Browse(string queueName, IBrowserCallback action);
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object BrowseSelected(string messageSelector, IBrowserCallback action);
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queue">The queue to browse.</param>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <returns></returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object BrowseSelected(Queue queue, string messageSelector, IBrowserCallback action);
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queueName">Name of the queue to browse,
/// (to be resolved to an actual destination by a DestinationResolver)</param>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback object that exposes the session/browser pair.</param>
/// <returns></returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object BrowseSelected(string queueName, string messageSelector, IBrowserCallback action);
/// <summary>
/// Browses messages in the default EMS queue. The callback gives access to the EMS
/// Session and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <returns>the result object from working with the session</returns>
object BrowseWithDelegate(BrowserDelegate action);
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queue">The queue to browse.</param>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <returns>the result object from working with the session</returns>
object BrowseWithDelegate(Queue queue, BrowserDelegate action);
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queueName">Name of the queue to browse,
/// (to be resolved to an actual destination by a DestinationResolver)</param>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object BrowseWithDelegate(string queueName, BrowserDelegate action);
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object BrowseSelectedWithDelegate(string messageSelector, BrowserDelegate action);
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queue">The queue to browse.</param>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <returns></returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object BrowseSelectedWithDelegate(Queue queue, string messageSelector, BrowserDelegate action);
/// <summary>
/// Browses messages in a EMS queue. The callback gives access to the EMS Session
/// and QueueBrowser in order to browse the queue and react to the contents.
/// </summary>
/// <param name="queueName">Name of the queue to browse,
/// (to be resolved to an actual destination by a DestinationResolver)</param>
/// <param name="messageSelector">The EMS message selector expression (or <code>null</code> if none).</param>
/// <param name="action">The action callback delegate that exposes the session/browser pair.</param>
/// <returns></returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
object BrowseSelectedWithDelegate(string queueName, string messageSelector, BrowserDelegate action);
}
}

23
src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsUtils.cs
View File

@@ -238,6 +238,27 @@ namespace Spring.Messaging.Ems.Support
// // Ignore -> can only happen in case of a JTA transaction.
// }
}
/// <summary>
/// Closes the given queue browser and ignore any thrown exception.
/// This is useful for typical <code>finally</code> blocks in manual EMS code.
/// </summary>
/// <param name="browser">The queue browser to close (may be <code>null</code>.</param>
public static void CloseQueueBrowser(QueueBrowser browser)
{
if (browser != null)
{
try
{
browser.Close();
} catch (EMSException ex)
{
logger.Debug("Could not close EMS QueueBrowser", ex);
} catch (Exception ex)
{
logger.Debug("Unexpected exception on closing EMS QueueBrowser", ex);
}
}
}
}
}