/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You 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. */ using System; using System.Threading; using Apache.NMS; using Apache.NMS.Util; using Apache.NMS.XMS.Util; namespace Apache.NMS.XMS { class MessageProducer : Apache.NMS.IMessageProducer { protected readonly Apache.NMS.XMS.Session nmsSession; public IBM.XMS.IMessageProducer xmsMessageProducer; private TimeSpan requestTimeout = NMSConstants.defaultRequestTimeout; private bool closed = false; private bool disposed = false; public MessageProducer(Apache.NMS.XMS.Session session, IBM.XMS.IMessageProducer producer) { this.nmsSession = session; this.xmsMessageProducer = producer; this.RequestTimeout = session.RequestTimeout; } ~MessageProducer() { Dispose(false); } private Apache.NMS.XMS.Message GetXMSMessage(Apache.NMS.IMessage message) { Apache.NMS.XMS.Message msg = (Apache.NMS.XMS.Message) message; if(this.ProducerTransformer != null) { IMessage transformed = this.ProducerTransformer(this.nmsSession, this, message); if(transformed != null) { msg = (Apache.NMS.XMS.Message) transformed; } } return msg; } #region IMessageProducer Members /// /// Sends the message to the default destination for this producer. /// public void Send(Apache.NMS.IMessage message) { Apache.NMS.XMS.Message msg = GetXMSMessage(message); long timeToLive = (long) message.NMSTimeToLive.TotalMilliseconds; if(0 == timeToLive) { timeToLive = this.xmsMessageProducer.TimeToLive; } try { msg.OnSend(); this.xmsMessageProducer.Send( msg.xmsMessage, this.xmsMessageProducer.DeliveryMode, this.xmsMessageProducer.Priority, timeToLive); } catch(Exception ex) { ExceptionUtil.WrapAndThrowNMSException(ex); } } /// /// Sends the message to the default destination with the explicit QoS /// configuration. /// public void Send(Apache.NMS.IMessage message, MsgDeliveryMode deliveryMode, MsgPriority priority, TimeSpan timeToLive) { Apache.NMS.XMS.Message msg = GetXMSMessage(message); try { this.xmsMessageProducer.Send( msg.xmsMessage, XMSConvert.ToJMSDeliveryMode(deliveryMode), (int)priority, (long)timeToLive.TotalMilliseconds); } catch(Exception ex) { ExceptionUtil.WrapAndThrowNMSException(ex); } } /// /// Sends the message to the given destination. /// public void Send(Apache.NMS.IDestination destination, Apache.NMS.IMessage message) { Apache.NMS.XMS.Destination dest = (Apache.NMS.XMS.Destination)destination; Apache.NMS.XMS.Message msg = GetXMSMessage(message); long timeToLive = (long)message.NMSTimeToLive.TotalMilliseconds; if(0 == timeToLive) { timeToLive = this.xmsMessageProducer.TimeToLive; } try { this.xmsMessageProducer.Send( dest.xmsDestination, msg.xmsMessage, this.xmsMessageProducer.DeliveryMode, this.xmsMessageProducer.Priority, timeToLive); } catch(Exception ex) { ExceptionUtil.WrapAndThrowNMSException(ex); } } /// /// Sends the message to the given destination with the explicit QoS /// configuration. /// public void Send(Apache.NMS.IDestination destination, Apache.NMS.IMessage message, MsgDeliveryMode deliveryMode, MsgPriority priority, TimeSpan timeToLive) { Apache.NMS.XMS.Destination dest = (Apache.NMS.XMS.Destination)destination; Apache.NMS.XMS.Message msg = GetXMSMessage(message); try { this.xmsMessageProducer.Send( dest.xmsDestination, msg.xmsMessage, XMSConvert.ToJMSDeliveryMode(deliveryMode), (int)priority, (long)timeToLive.TotalMilliseconds); } catch(Exception ex) { ExceptionUtil.WrapAndThrowNMSException(ex); } } private ProducerTransformerDelegate producerTransformer; /// /// A delegate that is called each time a Message is sent from this /// Producer which allows the application to perform any needed /// transformations on the Message before it is sent. /// public ProducerTransformerDelegate ProducerTransformer { get { return this.producerTransformer; } set { this.producerTransformer = value; } } public MsgDeliveryMode DeliveryMode { get { return XMSConvert.ToNMSMsgDeliveryMode( this.xmsMessageProducer.DeliveryMode); } set { try { this.xmsMessageProducer.DeliveryMode = XMSConvert.ToJMSDeliveryMode(value); } catch(Exception ex) { ExceptionUtil.WrapAndThrowNMSException(ex); } } } public TimeSpan TimeToLive { get { return TimeSpan.FromMilliseconds( this.xmsMessageProducer.TimeToLive); } set { try { this.xmsMessageProducer.TimeToLive = (long)value.TotalMilliseconds; } catch(Exception ex) { ExceptionUtil.WrapAndThrowNMSException(ex); } } } /// /// The default timeout for network requests. /// public TimeSpan RequestTimeout { get { return requestTimeout; } set { this.requestTimeout = value; } } public MsgPriority Priority { get { return (MsgPriority) this.xmsMessageProducer.Priority; } set { try { this.xmsMessageProducer.Priority = (int) value; } catch(Exception ex) { ExceptionUtil.WrapAndThrowNMSException(ex); } } } public bool DisableMessageID { get { return this.xmsMessageProducer.DisableMessageID; } set { try { this.xmsMessageProducer.DisableMessageID = value; } catch(Exception ex) { ExceptionUtil.WrapAndThrowNMSException(ex); } } } public bool DisableMessageTimestamp { get { return this.xmsMessageProducer.DisableMessageTimestamp; } set { try { this.xmsMessageProducer.DisableMessageTimestamp = value; } catch(Exception ex) { ExceptionUtil.WrapAndThrowNMSException(ex); } } } /// /// Creates a new message with an empty body /// public Apache.NMS.IMessage CreateMessage() { return this.nmsSession.CreateMessage(); } /// /// Creates a new text message with an empty body. /// public Apache.NMS.ITextMessage CreateTextMessage() { return this.nmsSession.CreateTextMessage(); } /// /// Creates a new text message with the given body. /// public Apache.NMS.ITextMessage CreateTextMessage(string text) { return this.nmsSession.CreateTextMessage(text); } /// /// Creates a new Map message which contains primitive key and value /// pairs. /// public Apache.NMS.IMapMessage CreateMapMessage() { return this.nmsSession.CreateMapMessage(); } /// /// Creates a new object message containing the given .NET object as /// the body. /// public Apache.NMS.IObjectMessage CreateObjectMessage(object body) { return this.nmsSession.CreateObjectMessage(body); } /// /// Creates a new binary message. /// public Apache.NMS.IBytesMessage CreateBytesMessage() { return this.nmsSession.CreateBytesMessage(); } /// /// Creates a new binary message with the given body. /// public Apache.NMS.IBytesMessage CreateBytesMessage(byte[] body) { return this.nmsSession.CreateBytesMessage(body); } /// /// Creates a new stream message. /// public Apache.NMS.IStreamMessage CreateStreamMessage() { return this.nmsSession.CreateStreamMessage(); } #endregion #region IDisposable Members public void Close() { lock(this) { if(closed) { return; } try { this.xmsMessageProducer.Close(); } catch(Exception ex) { ExceptionUtil.WrapAndThrowNMSException(ex); } finally { closed = true; } } } /// ///Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. /// ///2 public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } protected void Dispose(bool disposing) { if(disposed) { return; } if(disposing) { // Dispose managed code here. } try { Close(); } catch { // Ignore errors. } disposed = true; } #endregion } }