Use execute(SessionCallback) for the general case.
+ /// Starting the EMS Connection is just necessary for receiving messages,
+ /// which is preferably achieved through the receive methods.
+ ///
+ /// callback object that exposes the session
+ ///
+ /// Start the connection before performing callback action.
+ ///
+ /// the result object from working with the session
+ ///
+ /// EMSException if there is any problem
+ public virtual object Execute(ISessionCallback action, bool startConnection)
+ {
+ AssertUtils.ArgumentNotNull(action, "Callback object must not be null");
+
+ Connection con = null;
+ Session session = null;
+ bool sessionInTLS = true;
+
+ //NOTE: Not closing session or connection unless session is not returned from
+ // ConnectionFactoryUtils.DoGetTransactionalSession and CacheEmsResources is set to false
+ try
+ {
+ Session sessionToUse =
+ ConnectionFactoryUtils.DoGetTransactionalSession(ConnectionFactory, transactionalResourceFactory,
+ startConnection);
+ if (sessionToUse == null)
+ {
+ sessionInTLS = false;
+ con = CreateConnection();
+ session = CreateSession(con);
+ if (startConnection)
+ {
+ con.Start();
+ }
+ sessionToUse = session;
+ }
+ if (logger.IsDebugEnabled)
+ {
+ logger.Debug("Executing callback on EMS Session [" + sessionToUse + "]");
+ }
+ return action.DoInEms(sessionToUse);
+ }
+ finally
+ {
+ if (!sessionInTLS && !CacheEmsResources)
+ {
+ EmsUtils.CloseSession(session);
+ ConnectionFactoryUtils.ReleaseConnection(con, ConnectionFactory, startConnection);
+ }
+ }
+ }
+
+ #endregion
+
+ #region Properties
+
+ ///
+ /// Gets or sets the default destination to be used on send/receive operations that do not
+ /// have a destination parameter.
+ ///
+ /// Alternatively, specify a "defaultDestinationName", to be
+ /// dynamically resolved via the DestinationResolver.
+ /// The default destination.
+ virtual public Destination DefaultDestination
+ {
+ get { return (defaultDestination as Destination); }
+
+ set { defaultDestination = value; }
+ }
+
+
+ ///
+ /// Gets or sets the name of the default destination name
+ /// to be used on send/receive operations that
+ /// do not have a destination parameter.
+ ///
+ ///
+ /// Alternatively, specify a EMS Destination object as "DefaultDestination"
+ ///
+ /// The name of the default destination.
+ virtual public string DefaultDestinationName
+ {
+ get { return (defaultDestination as string); }
+
+ set { defaultDestination = value; }
+ }
+
+ ///
+ /// Gets or sets the message converter for this template.
+ ///
+ ///
+ /// Used to resolve
+ /// Object parameters to convertAndSend methods and Object results
+ /// from receiveAndConvert methods.
+ ///
The default converter is a SimpleMessageConverter, which is able
+ /// to handle BytesMessages, TextMessages and ObjectMessages.
+ ///
+ /// The message converter.
+ virtual public IMessageConverter MessageConverter
+ {
+ get { return messageConverter; }
+
+ set { messageConverter = value; }
+ }
+
+ ///
+ /// Gets or sets a value indicating whether Message Ids are enabled.
+ ///
+ /// true if message id enabled; otherwise, false.
+ virtual public bool MessageIdEnabled
+ {
+ get { return messageIdEnabled; }
+
+ set { messageIdEnabled = value; }
+ }
+
+ ///
+ /// Gets or sets a value indicating whether message timestamps are enabled.
+ ///
+ ///
+ /// true if [message timestamp enabled]; otherwise, false.
+ ///
+ virtual public bool MessageTimestampEnabled
+ {
+ get { return messageTimestampEnabled; }
+
+ set { messageTimestampEnabled = value; }
+ }
+
+
+ ///
+ /// Gets or sets a value indicating whether to inhibit the delivery of messages published by its own connection.
+ ///
+ ///
+ /// true if inhibit the delivery of messages published by its own connection; otherwise, false.
+ virtual public bool PubSubNoLocal
+ {
+ get { return pubSubNoLocal; }
+
+ set { pubSubNoLocal = value; }
+ }
+
+ ///
+ /// Gets or sets the receive timeout to use for recieve calls.
+ ///
+ /// The default is -1, which means no timeout.
+ /// The receive timeout.
+ virtual public long ReceiveTimeout
+ {
+ get { return receiveTimeout; }
+
+ set { receiveTimeout = value; }
+ }
+
+ ///
+ /// Gets or sets a value indicating whether to use explicit Quality of Service values.
+ ///
+ /// If "true", then the values of deliveryMode, priority, and timeToLive
+ /// will be used when sending a message. Otherwise, the default values,
+ /// that may be set administratively, will be used
+ /// true if use explicit QoS values; otherwise, false.
+ virtual public bool ExplicitQosEnabled
+ {
+ get { return explicitQosEnabled; }
+
+ set { explicitQosEnabled = value; }
+ }
+
+ ///
+ /// Sets a value indicating the delivery mode QOS
+ ///
+ ///
+ /// This will set the delivery to persistent or non-persistent
+ /// Default value is Message.DEFAULT_DELIVERY_MODE
+ ///
+ /// true if [delivery persistent]; otherwise, false.
+ virtual public int DeliveryMode
+ {
+ get { return deliveryMode; }
+
+ set { deliveryMode = value; }
+ }
+
+ ///
+ /// Gets or sets the priority when sending.
+ ///
+ /// Since a default value may be defined administratively,
+ /// this is only used when "isExplicitQosEnabled" equals "true".
+ /// The priority.
+ virtual public int Priority
+ {
+ get { return priority; }
+
+ set { priority = value; }
+ }
+
+ ///
+ /// Gets or sets the time to live when sending
+ ///
+ /// Since a default value may be defined administratively,
+ /// this is only used when "isExplicitQosEnabled" equals "true".
+ /// The time to live.
+ virtual public long TimeToLive
+ {
+ get { return timeToLive; }
+
+ set { timeToLive = value; }
+ }
+
+ ///
+ /// Gets or sets a value indicating whether the EmsTemplate should itself
+ /// be responsible for caching EMS Connection/Session/MessageProducer as compared to
+ /// creating new instances per operation (unless such resources are already
+ /// present in Thread-Local storage either due to the use of EmsTransactionMananger or
+ /// SimpleMessageListenerContainer at an outer calling layer.
+ ///
+ /// Connection/Session/MessageProducer are thread-safe classes in TIBCO EMS.
+ /// true to locally cache ems resources; otherwise, false.
+ virtual public bool CacheEmsResources
+ {
+ get { return cacheEmsResources; }
+ set { cacheEmsResources = value; }
+ }
+
+
+ #endregion
+
+ ///
+ /// Extract the content from the given JMS message.
+ ///
+ /// The Message to convert (can be null).
+ /// The content of the message, or null if none
+ protected virtual object DoConvertFromMessage(Message message)
+ {
+ if (message != null)
+ {
+ return MessageConverter.FromMessage(message);
+ }
+ return null;
+ }
+
+ #region EMS Factory Methods
+
+ /// Fetch an appropriate Connection from the given EmsResourceHolder.
+ ///
+ /// the EmsResourceHolder
+ ///
+ /// an appropriate Connection fetched from the holder,
+ /// or null if none found
+ ///
+ protected virtual Connection GetConnection(EmsResourceHolder holder)
+ {
+ return holder.GetConnection();
+ }
+
+ /// Fetch an appropriate Session from the given EmsResourceHolder.
+ ///
+ /// the EmsResourceHolder
+ ///
+ /// an appropriate Session fetched from the holder,
+ /// or null if none found
+ ///
+ protected virtual Session GetSession(EmsResourceHolder holder)
+ {
+ return holder.GetSession();
+ }
+
+ /// Create a EMS MessageProducer for the given Session and Destination,
+ /// configuring it to disable message ids and/or timestamps (if necessary).
+ ///
Delegates to doCreateProducer for creation of the raw
+ /// EMS MessageProducer
+ ///
+ /// the EMS Session to create a MessageProducer for
+ ///
+ /// the EMS Destination to create a MessageProducer for
+ ///
+ /// the new EMS MessageProducer
+ ///
+ /// EMSException if thrown by EMS API methods
+ ///
+ ///
+ ///
+ ///
+ ///
+ ///
+ protected virtual MessageProducer CreateProducer(Session session, Destination destination)
+ {
+ MessageProducer producer = DoCreateProducer(session, destination);
+ if (!MessageIdEnabled)
+ {
+ producer.DisableMessageID = true;
+ }
+ if (!MessageTimestampEnabled)
+ {
+ producer.DisableMessageTimestamp = true;
+ }
+ return producer;
+ }
+
+
+ ///
+ /// Determines whether the given Session is locally transacted, that is, whether
+ /// its transaction is managed by this template class's Session handling
+ /// and not by an external transaction coordinator.
+ ///
+ ///
+ /// The Session's own transacted flag will already have been checked
+ /// before. This method is about finding out whether the Session's transaction
+ /// is local or externally coordinated.
+ ///
+ /// The session to check.
+ ///
+ /// true if the session is locally transacted; otherwise, false.
+ ///
+ protected virtual bool IsSessionLocallyTransacted(Session session)
+ {
+ return SessionTransacted &&
+ !ConnectionFactoryUtils.IsSessionTransactional(session, ConnectionFactory);
+ }
+
+ /// Create a raw EMS MessageProducer for the given Session and Destination.
+ ///
+ /// If CacheJmsResource is true, then the producer
+ /// will be created upon the first invocation and will retrun the same
+ /// producer (per destination) on all subsequent calls.
+ ///
+ /// the EMS Session to create a MessageProducer for
+ ///
+ /// the EMS Destination to create a MessageProducer for
+ ///
+ /// the new EMS MessageProducer
+ ///
+ /// EMSException if thrown by EMS API methods
+ protected virtual MessageProducer DoCreateProducer(Session session, Destination destination)
+ {
+ if (CacheEmsResources)
+ {
+ if (destination == null)
+ {
+ if (emsResources.UnspecifiedDestinationMessageProducer == null)
+ {
+ emsResources.UnspecifiedDestinationMessageProducer = session.CreateProducer(destination);
+ }
+ return emsResources.UnspecifiedDestinationMessageProducer;
+ }
+ MessageProducer producer = (MessageProducer)emsResources.Producers[destination];
+ if (producer != null)
+ {
+ #region Logging
+
+ if (logger.IsDebugEnabled)
+ {
+ logger.Debug("Found cached MessageProducer for destination [" + destination + "]");
+ }
+
+ #endregion
+ }
+ else
+ {
+ producer = session.CreateProducer(destination);
+ emsResources.Producers.Add(destination, producer);
+ #region Logging
+
+ if (logger.IsDebugEnabled)
+ {
+ logger.Debug("Created cached MessageProducer for destination [" + destination + "]");
+ }
+
+ #endregion
+ }
+ return producer;
+ }
+ else
+ {
+ return session.CreateProducer(destination);
+ }
+ }
+
+ /// Create a EMS MessageConsumer for the given Session and Destination.
+ ///
+ /// the EMS Session to create a MessageConsumer for
+ ///
+ /// the EMS Destination to create a MessageConsumer for
+ ///
+ /// the message selector for this consumer (can be null)
+ ///
+ /// the new EMS MessageConsumer
+ ///
+ /// EMSException if thrown by EMS API methods
+ protected virtual MessageConsumer CreateConsumer(Session session, Destination destination,
+ string messageSelector)
+ {
+ // Only pass in the NoLocal flag in case of a Topic:
+ // Some EMS providers, such as WebSphere MQ 6.0, throw IllegalStateException
+ // in case of the NoLocal flag being specified for a Queue.
+ if (PubSubDomain)
+ {
+ return session.CreateConsumer(destination, messageSelector, PubSubNoLocal);
+ }
+ else
+ {
+ return session.CreateConsumer(destination, messageSelector);
+ }
+ }
+
+ /// Create a EMS Connection via this template's ConnectionFactory.
+ ///
+ /// If CacheJmsResource is true, then the connection
+ /// will be created upon the first invocation and will retrun the same
+ /// connection on all subsequent calls.
+ ///
+ /// A EMS Connection
+ ///
+ /// EMSException if thrown by EMS API methods
+ protected override Connection CreateConnection()
+ {
+ if (CacheEmsResources)
+ {
+ if (emsResources.Connection == null)
+ {
+ emsResources.Connection = ConnectionFactory.CreateConnection();
+ }
+ return emsResources.Connection;
+
+ }
+ else
+ {
+ return ConnectionFactory.CreateConnection();
+ }
+ }
+
+ /// Create a EMS Session for the given Connection.
+ ///
+ /// If CacheJmsResource is true, then the session
+ /// will be created upon the first invocation and will retrun the same
+ /// session on all subsequent calls.
+ ///
+ /// the EMS Connection to create a Session for
+ ///
+ /// the new EMS Session
+ ///
+ /// EMSException if thrown by EMS API methods
+ protected override Session CreateSession(Connection con)
+ {
+ if (CacheEmsResources)
+ {
+ if (emsResources.Session == null)
+ {
+ emsResources.Session = emsResources.Connection.CreateSession(SessionTransacted, SessionAcknowledgeMode);
+ }
+ return emsResources.Session;
+ }
+ else
+ {
+ return con.CreateSession(SessionTransacted, SessionAcknowledgeMode);
+ }
+ }
+
+ ///
+ /// Send the given message.
+ ///
+ /// The session to operate on.
+ /// The destination to send to.
+ /// The message creator delegate callback to create a Message.
+ protected internal virtual void DoSend(Session session, Destination destination, MessageCreatorDelegate messageCreatorDelegate)
+ {
+ AssertUtils.ArgumentNotNull(messageCreatorDelegate, "IMessageCreatorDelegate must not be null");
+ DoSend(session, destination, null, messageCreatorDelegate);
+ }
+
+ ///
+ /// Send the given message.
+ ///
+ /// The session to operate on.
+ /// The destination to send to.
+ /// The message creator callback to create a Message.
+ protected internal virtual void DoSend(Session session, Destination destination, IMessageCreator messageCreator)
+ {
+ AssertUtils.ArgumentNotNull(messageCreator, "IMessageCreator must not be null");
+ DoSend(session, destination, messageCreator, null);
+ }
+
+ /// Send the given EMS message.
+ /// the EMS Session to operate on
+ ///
+ /// the EMS Destination to send to
+ ///
+ /// callback to create a EMS Message
+ ///
+ /// delegate callback to create a EMS Message
+ ///
+ /// EMSException if thrown by EMS API methods
+ protected internal virtual void DoSend(Session session, Destination destination, IMessageCreator messageCreator,
+ MessageCreatorDelegate messageCreatorDelegate)
+ {
+
+
+ MessageProducer producer = CreateProducer(session, destination);
+ try
+ {
+
+ Message message = null;
+ if (messageCreator != null)
+ {
+ message = messageCreator.CreateMessage(session) ;
+ }
+ else {
+ message = messageCreatorDelegate(session);
+ }
+ if (logger.IsDebugEnabled)
+ {
+ logger.Debug("Sending created message [" + message + "]");
+ }
+ DoSend(producer, message);
+
+ // Check commit, avoid commit call is Session transaction is externally coordinated.
+ if (session.Transacted && IsSessionLocallyTransacted(session))
+ {
+ // Transacted session created by this template -> commit.
+ EmsUtils.CommitIfNecessary(session);
+ }
+ }
+ finally
+ {
+ EmsUtils.CloseMessageProducer(producer);
+ }
+ }
+
+
+ /// Actually send the given EMS message.
+ /// the EMS MessageProducer to send with
+ ///
+ /// the EMS Message to send
+ ///
+ /// EMSException if thrown by EMS API methods
+ protected virtual void DoSend(MessageProducer producer, Message message)
+ {
+ if (ExplicitQosEnabled)
+ {
+ producer.Send(message, DeliveryMode, Priority, TimeToLive);
+ }
+ else
+ {
+ producer.Send(message);
+ }
+ }
+
+
+ #endregion
+
+ #region IEmsOperations Implementation
+
+ ///
+ /// Execute the action specified by the given action object within
+ /// a EMS Session.
+ ///
+ /// delegate that exposes the session
+ ///
+ /// the result object from working with the session
+ ///
+ ///
+ /// Note that the value of PubSubDomain affects the behavior of this method.
+ /// If PubSubDomain equals true, then a Session is passed to the callback.
+ /// If false, then a Session is passed to the callback.b
+ ///
+ /// EMSException if there is any problem
+ public object Execute(SessionDelegate del)
+ {
+ return Execute(new ExecuteSessionCallbackUsingDelegate(del));
+ }
+
+ /// Execute the action specified by the given action object within
+ /// a EMS Session.
+ ///
Note: The value of PubSubDomain affects the behavior of this method.
+ /// If PubSubDomain equals true, then a Session is passed to the callback.
+ /// If false, then a Session is passed to the callback.
+ ///
+ /// callback object that exposes the session
+ ///
+ /// the result object from working with the session
+ ///
+ /// EMSException if there is any problem
+ public object Execute(ISessionCallback action)
+ {
+ return Execute(action, false);
+ }
+
+ /// Send a message to a EMS destination. The callback gives access to
+ /// the EMS session and MessageProducer in order to do more complex
+ /// send operations.
+ ///
+ /// callback object that exposes the session/producer pair
+ ///
+ /// the result object from working with the session
+ ///
+ /// EMSException if there is any problem
+ public object Execute(IProducerCallback action)
+ {
+ return Execute(new ProducerCreatorCallback(this, action));
+ }
+
+ /// Send a message to a EMS destination. The callback gives access to
+ /// the EMS session and MessageProducer in order to do more complex
+ /// send operations.
+ ///
+ /// delegate that exposes the session/producer pair
+ ///
+ /// the result object from working with the session
+ ///
+ /// EMSException if there is any problem
+ public object Execute(ProducerDelegate del)
+ {
+ return Execute(new ProducerCreatorCallback(this, del));
+ }
+
+ /// Send a message to the default destination.
+ ///
This will only work with a default destination specified!
+ ///
+ /// delegate callback to create a message
+ ///
+ /// EMSException if there is any problem
+ public void SendWithDelegate(MessageCreatorDelegate messageCreatorDelegate)
+ {
+ CheckDefaultDestination();
+ if (DefaultDestination != null)
+ {
+ SendWithDelegate(DefaultDestination, messageCreatorDelegate);
+ }
+ else
+ {
+ SendWithDelegate(DefaultDestinationName, messageCreatorDelegate);
+ }
+ }
+
+ /// Send a message to the specified destination.
+ /// The MessageCreator callback creates the message given a Session.
+ ///
+ /// the destination to send this message to
+ ///
+ /// delegate callback to create a message
+ ///
+ /// EMSException if there is any problem
+ public void SendWithDelegate(Destination destination, MessageCreatorDelegate messageCreatorDelegate)
+ {
+ Execute(new SendDestinationCallback(this, destination, messageCreatorDelegate), false);
+ }
+
+ /// Send a message to the specified destination.
+ /// The MessageCreator callback creates the message given a Session.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// delegate callback to create a message
+ ///
+ /// EMSException if there is any problem
+ public void SendWithDelegate(string destinationName, MessageCreatorDelegate messageCreatorDelegate)
+ {
+ Execute(new SendDestinationCallback(this, destinationName, messageCreatorDelegate), false);
+ }
+
+ /// Send a message to the default destination.
+ ///
This will only work with a default destination specified!
+ ///
+ /// callback to create a message
+ ///
+ /// EMSException if there is any problem
+ public void Send(IMessageCreator messageCreator)
+ {
+ CheckDefaultDestination();
+ if (DefaultDestination != null)
+ {
+ Send(DefaultDestination, messageCreator);
+ }
+ else
+ {
+ Send(DefaultDestinationName, messageCreator);
+ }
+ }
+
+ /// Send a message to the specified destination.
+ /// The MessageCreator callback creates the message given a Session.
+ ///
+ /// the destination to send this message to
+ ///
+ /// callback to create a message
+ ///
+ /// EMSException if there is any problem
+ public void Send(Destination destination, IMessageCreator messageCreator)
+ {
+
+ Execute(new SendDestinationCallback(this, destination, messageCreator), false);
+ }
+
+ /// Send a message to the specified destination.
+ /// The MessageCreator callback creates the message given a Session.
+ ///
+ /// the destination to send this message to
+ ///
+ /// callback to create a message
+ ///
+ /// EMSException if there is any problem
+ public void Send(string destinationName, IMessageCreator messageCreator)
+ {
+ Execute(new SendDestinationCallback(this, destinationName, messageCreator), false);
+ }
+ /// Send the given object to the default destination, converting the object
+ /// to a EMS message with a configured IMessageConverter.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the object to convert to a message
+ ///
+ /// EMSException if there is any problem
+ public void ConvertAndSend(object message)
+ {
+ CheckDefaultDestination();
+ if (DefaultDestination != null)
+ {
+ ConvertAndSend(DefaultDestination, message);
+ }
+ else
+ {
+ ConvertAndSend(DefaultDestinationName, message);
+ }
+ }
+
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter.
+ ///
+ /// the destination to send this message to
+ ///
+ /// the object to convert to a message
+ ///
+ /// EMSException if there is any problem
+ public void ConvertAndSend(Destination destination, object message)
+ {
+ CheckMessageConverter();
+ Send(destination, new SimpleMessageCreator(this, message));
+ }
+
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// the object to convert to a message
+ ///
+ /// EMSException if there is any problem
+ public void ConvertAndSend(string destinationName, object message)
+ {
+ Send(destinationName, new SimpleMessageCreator(this, message));
+ }
+
+ /// Send the given object to the default destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the object to convert to a message
+ ///
+ /// the callback to modify the message
+ ///
+ /// EMSException if there is any problem
+ public void ConvertAndSend(object message, IMessagePostProcessor postProcessor)
+ {
+ CheckDefaultDestination();
+ if (DefaultDestination != null)
+ {
+ ConvertAndSend(DefaultDestination, message, postProcessor);
+ }
+ else
+ {
+ ConvertAndSend(DefaultDestinationName, message, postProcessor);
+ }
+ }
+
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
+ /// the destination to send this message to
+ ///
+ /// the object to convert to a message
+ ///
+ /// the callback to modify the message
+ ///
+ /// EMSException if there is any problem
+ public void ConvertAndSend(Destination destination, object message, IMessagePostProcessor postProcessor)
+ {
+ CheckMessageConverter();
+ Send(destination, new ConvertAndSendMessageCreator(this, message, postProcessor));
+
+ }
+
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// the object to convert to a message.
+ ///
+ /// the callback to modify the message
+ ///
+ /// EMSException if there is any problem
+ public void ConvertAndSend(string destinationName, object message, IMessagePostProcessor postProcessor)
+ {
+ CheckMessageConverter();
+ Send(destinationName, new ConvertAndSendMessageCreator(this, message, postProcessor));
+ }
+ ///
+ /// Send the given object to the default destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the object to convert to a message
+ /// the callback to modify the message
+ /// EMSException if there is any problem
+ public void ConvertAndSendWithDelegate(object message, MessagePostProcessorDelegate postProcessor)
+ {
+ //Execute(new SendDestinationCallback(this, destination, messageCreatorDelegate), false);
+ CheckDefaultDestination();
+ if (DefaultDestination != null)
+ {
+ ConvertAndSendWithDelegate(DefaultDestination, message, postProcessor);
+ }
+ else
+ {
+ ConvertAndSendWithDelegate(DefaultDestinationName, message, postProcessor);
+ }
+ }
+
+ ///
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
+ /// the destination to send this message to
+ /// the object to convert to a message
+ /// the callback to modify the message
+ /// EMSException if there is any problem
+ public void ConvertAndSendWithDelegate(Destination destination, object message,
+ MessagePostProcessorDelegate postProcessor)
+ {
+ CheckMessageConverter();
+ Send(destination, new ConvertAndSendMessageCreator(this, message, postProcessor));
+ }
+
+ ///
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ /// the object to convert to a message.
+ /// the callback to modify the message
+ /// EMSException if there is any problem
+ public void ConvertAndSendWithDelegate(string destinationName, object message,
+ MessagePostProcessorDelegate postProcessor)
+ {
+ CheckMessageConverter();
+ Send(destinationName, new ConvertAndSendMessageCreator(this, message, postProcessor));
+
+ }
+
+ /// Receive a message synchronously from the default destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ public Message Receive()
+ {
+ CheckDefaultDestination();
+ if (DefaultDestination != null)
+ {
+ return Receive(DefaultDestination);
+ }
+ else
+ {
+ return Receive(DefaultDestinationName);
+ }
+ }
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the destination to receive a message from
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ public Message Receive(Destination destination)
+ {
+ return Execute(new ReceiveCallback(this, destination)) as Message;
+ }
+
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ public Message Receive(string destinationName)
+ {
+ return Execute(new ReceiveCallback(this, destinationName)) as Message;
+ }
+
+ /// Receive a message synchronously from the default destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the EMS message selector expression (or null if none).
+ /// See the EMS specification for a detailed definition of selector expressions.
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ public Message ReceiveSelected(string messageSelector)
+ {
+ CheckDefaultDestination();
+ if (DefaultDestination!= null)
+ {
+ return ReceiveSelected(DefaultDestination, messageSelector);
+ }
+ else
+ {
+ return ReceiveSelected(DefaultDestinationName, messageSelector);
+ }
+ }
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the destination to receive a message from
+ ///
+ /// the EMS message selector expression (or null if none).
+ /// See the EMS specification for a detailed definition of selector expressions.
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ public Message ReceiveSelected(Destination destination, string messageSelector)
+ {
+ return Execute(new ReceiveSelectedCallback(this, destination, messageSelector), true) as Message;
+ }
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// the EMS message selector expression (or null if none).
+ /// See the EMS specification for a detailed definition of selector expressions.
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ public Message ReceiveSelected(string destinationName, string messageSelector)
+ {
+ return Execute(new ReceiveSelectedCallback(this, destinationName, messageSelector), true) as Message;
+
+ }
+
+ ///
+ /// Receive a message.
+ ///
+ /// The session to operate on.
+ /// The destination to receive from.
+ /// The message selector for this consumer (can be null
+ /// The Message received, or null if none.
+ protected virtual Message DoReceive(Session session, Destination destination, string messageSelector)
+ {
+ return DoReceive(session, CreateConsumer(session, destination, messageSelector));
+ }
+
+ ///
+ /// Receive a message.
+ ///
+ /// The session to operate on.
+ /// The consumer to receive with.
+ /// The Message received, or null if none
+ protected virtual Message DoReceive(Session session, MessageConsumer consumer)
+ {
+ try
+ {
+ long timeout = ReceiveTimeout;
+ EmsResourceHolder resourceHolder =
+ (EmsResourceHolder)TransactionSynchronizationManager.GetResource(ConnectionFactory);
+ if (resourceHolder != null && resourceHolder.HasTimeout)
+ {
+ timeout = Convert.ToInt64(resourceHolder.TimeToLiveInMilliseconds);
+ }
+ Message message = (timeout > 0)
+ ? consumer.Receive(timeout)
+ : consumer.Receive();
+ if (session.Transacted)
+ {
+ // Commit necessary - but avoid commit call is Session transaction is externally coordinated.
+ if (IsSessionLocallyTransacted(session))
+ {
+ // Transacted session created by this template -> commit.
+ EmsUtils.CommitIfNecessary(session);
+ }
+ }
+ else if (IsClientAcknowledge(session))
+ {
+ // Manually acknowledge message, if any.
+ if (message != null)
+ {
+ message.Acknowledge();
+ }
+ }
+ return message;
+ }
+ finally
+ {
+ EmsUtils.CloseMessageConsumer(consumer);
+ }
+ }
+
+
+ /// Receive a message synchronously from the default destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the message produced for the consumer or null if the timeout expires.
+ ///
+ /// EMSException if there is any problem
+ public object ReceiveAndConvert()
+ {
+ CheckMessageConverter();
+ return DoConvertFromMessage(Receive());
+ }
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the destination to receive a message from
+ ///
+ /// the message produced for the consumer or null if the timeout expires.
+ ///
+ /// EMSException if there is any problem
+ public object ReceiveAndConvert(Destination destination)
+ {
+ CheckMessageConverter();
+ return DoConvertFromMessage(Receive(destination));
+ }
+
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// the message produced for the consumer or null if the timeout expires.
+ ///
+ /// EMSException if there is any problem
+ public object ReceiveAndConvert(string destinationName)
+ {
+ CheckMessageConverter();
+ return DoConvertFromMessage(Receive(destinationName));
+ }
+
+ /// Receive a message synchronously from the default destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the EMS message selector expression (or null if none).
+ /// See the EMS specification for a detailed definition of selector expressions.
+ ///
+ /// the message produced for the consumer or null if the timeout expires.
+ ///
+ /// EMSException if there is any problem
+ public object ReceiveSelectedAndConvert(string messageSelector)
+ {
+ CheckMessageConverter();
+ return DoConvertFromMessage(ReceiveSelected(messageSelector)); ;
+ }
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the destination to receive a message from
+ ///
+ /// the EMS message selector expression (or null if none).
+ /// See the EMS specification for a detailed definition of selector expressions.
+ ///
+ /// the message produced for the consumer or null if the timeout expires.
+ ///
+ /// EMSException if there is any problem
+ public object ReceiveSelectedAndConvert(Destination destination, string messageSelector)
+ {
+ CheckMessageConverter();
+ return DoConvertFromMessage(ReceiveSelected(destination, messageSelector));
+ }
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
Implemented by EmsTemplate. Not often used but a useful option
+ /// to enhance testability, as it can easily be mocked or stubbed.
+ ///
+ ///
Provides EmsTemplate'ssend(..) and
+ /// receive(..) methods that mirror various EMS API methods.
+ /// See the EMS specification and EMS API docs for details on those methods.
+ ///
+ ///
+ /// Mark Pollack
+ /// Juergen Hoeller
+ /// Mark Pollack (.NET)
+ public interface IEmsOperations
+ {
+ /// Execute the action specified by the given action object within
+ /// a EMS Session.
+ ///
+ /// delegate that exposes the session
+ /// the result object from working with the session
+ ///
+ /// EMSException if there is any problem
+ object Execute(SessionDelegate del);
+
+ /// Execute the action specified by the given action object within
+ /// a EMS Session.
+ ///
+ /// callback object that exposes the session
+ ///
+ /// the result object from working with the session
+ ///
+ /// EMSException if there is any problem
+ object Execute(ISessionCallback action);
+
+ /// Send a message to a EMS destination. The callback gives access to
+ /// the EMS session and MessageProducer in order to do more complex
+ /// send operations.
+ ///
+ /// callback object that exposes the session/producer pair
+ ///
+ /// the result object from working with the session
+ ///
+ /// EMSException if there is any problem
+ object Execute(IProducerCallback action);
+
+ /// Send a message to a EMS destination. The callback gives access to
+ /// the EMS session and MessageProducer in order to do more complex
+ /// send operations.
+ ///
+ /// delegate that exposes the session/producer pair
+ ///
+ /// the result object from working with the session
+ ///
+ /// EMSException if there is any problem
+ object Execute(ProducerDelegate del);
+
+
+ //-------------------------------------------------------------------------
+ // Convenience methods for sending messages
+ //-------------------------------------------------------------------------
+
+ /// Send a message to the default destination.
+ ///
This will only work with a default destination specified!
+ ///
+ /// callback to create a message
+ ///
+ /// EMSException if there is any problem
+ void Send(IMessageCreator messageCreator);
+
+ /// Send a message to the specified destination.
+ /// The IMessageCreator callback creates the message given a Session.
+ ///
+ /// the destination to send this message to
+ ///
+ /// callback to create a message
+ ///
+ /// EMSException if there is any problem
+ void Send(Destination destination, IMessageCreator messageCreator);
+
+ /// Send a message to the specified destination.
+ /// The IMessageCreator callback creates the message given a Session.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// callback to create a message
+ ///
+ /// EMSException if there is any problem
+ void Send(string destinationName, IMessageCreator messageCreator);
+
+ //-------------------------------------------------------------------------
+ // Convenience methods for sending messages
+ //-------------------------------------------------------------------------
+
+ /// Send a message to the default destination.
+ ///
This will only work with a default destination specified!
+ ///
+ /// delegate callback to create a message
+ ///
+ /// EMSException if there is any problem
+ void SendWithDelegate(MessageCreatorDelegate messageCreatorDelegate);
+
+ /// Send a message to the specified destination.
+ /// The IMessageCreator callback creates the message given a Session.
+ ///
+ /// the destination to send this message to
+ ///
+ /// delegate callback to create a message
+ ///
+ /// EMSException if there is any problem
+ void SendWithDelegate(Destination destination, MessageCreatorDelegate messageCreatorDelegate);
+
+ /// Send a message to the specified destination.
+ /// The IMessageCreator callback creates the message given a Session.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// delegate callback to create a message
+ ///
+ /// EMSException if there is any problem
+ void SendWithDelegate(string destinationName, MessageCreatorDelegate messageCreatorDelegate);
+
+ //-------------------------------------------------------------------------
+ // Convenience methods for sending auto-converted messages
+ //-------------------------------------------------------------------------
+
+ /// Send the given object to the default destination, converting the object
+ /// to a EMS message with a configured IMessageConverter.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the object to convert to a message
+ ///
+ /// EMSException if there is any problem
+ void ConvertAndSend(object message);
+
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter.
+ ///
+ /// the destination to send this message to
+ ///
+ /// the object to convert to a message
+ ///
+ /// EMSException if there is any problem
+ void ConvertAndSend(Destination destination, object message);
+
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// the object to convert to a message
+ ///
+ /// EMSException if there is any problem
+ void ConvertAndSend(string destinationName, object message);
+
+ /// Send the given object to the default destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the object to convert to a message
+ ///
+ /// the callback to modify the message
+ ///
+ /// EMSException if there is any problem
+ void ConvertAndSend(object message, IMessagePostProcessor postProcessor);
+
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
+ /// the destination to send this message to
+ ///
+ /// the object to convert to a message
+ ///
+ /// the callback to modify the message
+ ///
+ /// EMSException if there is any problem
+ void ConvertAndSend(Destination destination, object message, IMessagePostProcessor postProcessor);
+
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// the object to convert to a message.
+ ///
+ /// the callback to modify the message
+ ///
+ /// EMSException if there is any problem
+ void ConvertAndSend(string destinationName, object message, IMessagePostProcessor postProcessor);
+
+
+ ///
+ /// Send the given object to the default destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the object to convert to a message
+ /// the callback to modify the message
+ /// EMSException if there is any problem
+ void ConvertAndSendWithDelegate(object message, MessagePostProcessorDelegate postProcessor);
+
+ ///
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
+ /// the destination to send this message to
+ /// the object to convert to a message
+ /// the callback to modify the message
+ /// EMSException if there is any problem
+ void ConvertAndSendWithDelegate(Destination destination, object message, MessagePostProcessorDelegate postProcessor);
+
+ ///
+ /// Send the given object to the specified destination, converting the object
+ /// to a EMS message with a configured IMessageConverter. The IMessagePostProcessor
+ /// callback allows for modification of the message after conversion.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ /// the object to convert to a message.
+ /// the callback to modify the message
+ /// EMSException if there is any problem
+ void ConvertAndSendWithDelegate(string destinationName, object message, MessagePostProcessorDelegate postProcessor);
+
+
+ //-------------------------------------------------------------------------
+ // Convenience methods for receiving messages
+ //-------------------------------------------------------------------------
+
+ /// Receive a message synchronously from the default destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ Message Receive();
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the destination to receive a message from
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ Message Receive(Destination destination);
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ Message Receive(string destinationName);
+
+ /// Receive a message synchronously from the default destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the EMS message selector expression (or null if none).
+ /// See the EMS specification for a detailed definition of selector expressions.
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ Message ReceiveSelected(string messageSelector);
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the destination to receive a message from
+ ///
+ /// the EMS message selector expression (or null if none).
+ /// See the EMS specification for a detailed definition of selector expressions.
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ Message ReceiveSelected(Destination destination, string messageSelector);
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// the EMS message selector expression (or null if none).
+ /// See the EMS specification for a detailed definition of selector expressions.
+ ///
+ /// the message received by the consumer, or null if the timeout expires
+ ///
+ /// EMSException if there is any problem
+ Message ReceiveSelected(string destinationName, string messageSelector);
+
+
+ //-------------------------------------------------------------------------
+ // Convenience methods for receiving auto-converted messages
+ //-------------------------------------------------------------------------
+
+ /// Receive a message synchronously from the default destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the message produced for the consumer or null if the timeout expires.
+ ///
+ /// EMSException if there is any problem
+ object ReceiveAndConvert();
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the destination to receive a message from
+ ///
+ /// the message produced for the consumer or null if the timeout expires.
+ ///
+ /// EMSException if there is any problem
+ object ReceiveAndConvert(Destination destination);
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the name of the destination to send this message to
+ /// (to be resolved to an actual destination by a DestinationResolver)
+ ///
+ /// the message produced for the consumer or null if the timeout expires.
+ ///
+ /// EMSException if there is any problem
+ object ReceiveAndConvert(string destinationName);
+
+ /// Receive a message synchronously from the default destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
This will only work with a default destination specified!
+ ///
+ /// the EMS message selector expression (or null if none).
+ /// See the EMS specification for a detailed definition of selector expressions.
+ ///
+ /// the message produced for the consumer or null if the timeout expires.
+ ///
+ /// EMSException if there is any problem
+ object ReceiveSelectedAndConvert(string messageSelector);
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.
+ ///
+ /// the destination to receive a message from
+ ///
+ /// the EMS message selector expression (or null if none).
+ /// See the EMS specification for a detailed definition of selector expressions.
+ ///
+ /// the message produced for the consumer or null if the timeout expires.
+ ///
+ /// EMSException if there is any problem
+ object ReceiveSelectedAndConvert(Destination destination, string messageSelector);
+
+ /// Receive a message synchronously from the specified destination, but only
+ /// wait up to a specified time for delivery. Convert the message into an
+ /// object with a configured IMessageConverter.
+ ///
This method should be used carefully, since it will block the thread
+ /// until the message becomes available or until the timeout value is exceeded.