/* | |
* 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; | |
namespace Apache.NMS.EMS | |
{ | |
class MessageConsumer : Apache.NMS.IMessageConsumer | |
{ | |
private readonly Apache.NMS.EMS.Dispatcher dispatcher = new Apache.NMS.EMS.Dispatcher(); | |
protected readonly Apache.NMS.EMS.Session nmsSession; | |
public TIBCO.EMS.MessageConsumer tibcoMessageConsumer; | |
private bool closed = false; | |
private bool disposed = false; | |
public MessageConsumer(Apache.NMS.EMS.Session session, TIBCO.EMS.MessageConsumer consumer) | |
{ | |
this.nmsSession = session; | |
this.tibcoMessageConsumer = consumer; | |
this.tibcoMessageConsumer.MessageHandler += this.HandleTibcoMsg; | |
} | |
~MessageConsumer() | |
{ | |
Dispose(false); | |
} | |
#region IMessageConsumer Members | |
private ConsumerTransformerDelegate consumerTransformer; | |
/// <summary> | |
/// A Delegate that is called each time a Message is dispatched to allow the client to do | |
/// any necessary transformations on the received message before it is delivered. | |
/// </summary> | |
public ConsumerTransformerDelegate ConsumerTransformer | |
{ | |
get { return this.consumerTransformer; } | |
set { this.consumerTransformer = value; } | |
} | |
/// <summary> | |
/// Waits until a message is available and returns it | |
/// </summary> | |
public Apache.NMS.IMessage Receive() | |
{ | |
return this.dispatcher.Dequeue(); | |
} | |
/// <summary> | |
/// If a message is available within the timeout duration it is returned otherwise this method returns null | |
/// </summary> | |
public Apache.NMS.IMessage Receive(TimeSpan timeout) | |
{ | |
return this.dispatcher.Dequeue(timeout); | |
} | |
/// <summary> | |
/// If a message is available immediately it is returned otherwise this method returns null | |
/// </summary> | |
public Apache.NMS.IMessage ReceiveNoWait() | |
{ | |
return this.dispatcher.DequeueNoWait(); | |
} | |
/// <summary> | |
/// An asynchronous listener which can be used to consume messages asynchronously | |
/// </summary> | |
public event Apache.NMS.MessageListener Listener; | |
/// <summary> | |
/// Closes the message consumer. | |
/// </summary> | |
/// <remarks> | |
/// Clients should close message consumers them when they are not needed. | |
/// This call blocks until a receive or message listener in progress has completed. | |
/// A blocked message consumer receive call returns null when this message consumer is closed. | |
/// </remarks> | |
public void Close() | |
{ | |
lock(this) | |
{ | |
if(closed) | |
{ | |
return; | |
} | |
} | |
// wake up any pending dequeue() call on the dispatcher | |
this.dispatcher.Close(); | |
lock(this) | |
{ | |
try | |
{ | |
if(!this.nmsSession.tibcoSession.IsClosed) | |
{ | |
// JGomes: Calling the following message handler removal code creates an | |
// instability in the TIBCO consumer and will fail to unregister a consumer. | |
// This is a very odd bug, but it has been observed that unregistering the | |
// message handler is not necessary and can be harmful. When the unregister is | |
// executed slowly step-by-step under a debugger, then it works correctly. When | |
// it is run full speed, it creates a race condition. Therefore, the code is left | |
// here in a commented out state to demonstrate what normal cleanup code should | |
// look like, but don't uncomment it. | |
// this.tibcoMessageConsumer.MessageHandler -= this.HandleTibcoMsg; | |
this.tibcoMessageConsumer.Close(); | |
} | |
} | |
catch(Exception ex) | |
{ | |
ExceptionUtil.WrapAndThrowNMSException(ex); | |
} | |
finally | |
{ | |
closed = true; | |
} | |
} | |
} | |
#endregion | |
#region IDisposable Members | |
///<summary> | |
/// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. | |
///</summary> | |
///<filterpriority>2</filterpriority> | |
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 | |
private void HandleTibcoMsg(object sender, TIBCO.EMS.EMSMessageEventArgs arg) | |
{ | |
Apache.NMS.IMessage message = EMSConvert.ToNMSMessage(arg.Message); | |
if(null != message) | |
{ | |
if(this.ConsumerTransformer != null) | |
{ | |
IMessage newMessage = ConsumerTransformer(this.nmsSession, this, message); | |
if(newMessage != null) | |
{ | |
message = newMessage; | |
} | |
} | |
if(Listener != null) | |
{ | |
try | |
{ | |
Listener(message); | |
} | |
catch(Exception ex) | |
{ | |
Apache.NMS.Tracer.Debug("Error handling message: " + ex.Message); | |
} | |
} | |
else | |
{ | |
this.dispatcher.Enqueue(message); | |
} | |
} | |
} | |
} | |
} |