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

removed as other containers provide this functionality

Browse Source
This commit is contained in:
markpollack
2008-10-14 01:57:11 +00:00
parent 2be04c1a7e
commit c98113b4e8
1 changed files with 0 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files

229
src/Spring/Spring.Messaging/Messaging/Listener/SimpleMessageListenerContainer.cs
View File

@@ -1,229 +0,0 @@
#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 System;
using System.Messaging;
using System.Threading;
using Common.Logging;
namespace Spring.Messaging.Listener
{
public class SimpleMessageListenerContainer : AbstractMessageListenerContainer
{
#region Logging Definition
private static readonly ILog LOG = LogManager.GetLogger(typeof (SimpleMessageListenerContainer));
private Thread dispatcherThread;
protected ManualResetEvent stopEvent = new ManualResetEvent(false);
private IExceptionHandler exceptionHandler;
private MessageQueue messageQueue;
public MessageQueue MessageQueue
{
get { return messageQueue; }
set { messageQueue = value; }
}
public IExceptionHandler ExceptionHandler
{
get { return exceptionHandler; }
set { exceptionHandler = value; }
}
#endregion
protected override void DoInitialize()
{
messageQueue = ApplicationContext.GetObject(MessageQueueObjectName, typeof (MessageQueue)) as MessageQueue;
}
/// <summary>
/// Unsubscribe for messaging events and closethe queue
/// </summary>
protected override void DoShutdown()
{
CloseQueueHandle(MessageQueue);
if (dispatcherThread != null)
{
LOG.Debug("Waiting to join dispatcher thread.");
dispatcherThread.Join();
dispatcherThread = null;
LOG.Debug("Dispatcher thread terminated.");
}
}
/// <summary>
/// Re-initializes this container's consumers, if not initialized already.
/// </summary>
protected override void DoStart()
{
base.DoStart();
stopEvent = new ManualResetEvent(false);
dispatcherThread = new Thread(new ThreadStart(StartListening));
dispatcherThread.Start();
}
/// <summary>
/// Stops the container from listening to message events.
/// </summary>
public override void DoStop()
{
base.DoStop();
CloseQueueHandle(MessageQueue);
stopEvent.Set();
if (dispatcherThread != null)
{
LOG.Debug("Waiting to join dispatcher thread.");
dispatcherThread.Join();
dispatcherThread = null;
LOG.Debug("Dispatcher thread terminated.");
}
}
/// <summary>
/// Starts listening off the queue.
/// </summary>
protected virtual void StartListening()
{
if (Running)
{
try
{
IAsyncResult asynchResult = MessageQueue.BeginReceive();
LOG.Debug("WaitAny");
int firedWaitHandle = WaitHandle.WaitAny(new WaitHandle[] {asynchResult.AsyncWaitHandle, stopEvent});
if (firedWaitHandle == 0)
{
ReceiveCompleted(asynchResult);
}
else
{
//Do the endreceive?
return;
}
}
catch (Exception ex)
{
LOG.Error(
"Exception executing DefaultMessageQueue.BeginReceive. Reinvoking after recovery interval [" +
RecoveryTimeSpan + "]", ex);
Thread.Sleep(RecoveryTimeSpan);
StartListening();
}
}
}
protected virtual void ReceiveCompleted(IAsyncResult asyncResult)
{
Message message;
#region Receive Message
try
{
LOG.Debug("ReceiveCompleted called.");
// Get reference to the queue.
// End the asynchronous receive operation.
message = MessageQueue.EndReceive(asyncResult);
}
catch (MessageQueueException ex)
{
if (ex.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout)
{
if (LOG.IsTraceEnabled)
{
LOG.Trace("IOTimeout: Message to receive was already processed by another thread.");
}
}
else
{
// A real issue in receiving the message
LOG.Error("Error receiving message from DefaultMessageQueue = [" + MessageQueue.QueueName + "]");
Thread.Sleep(RecoveryTimeSpan);
//InvokeReceiveExceptionHandler(ex);?
}
MessageQueue.Close();
StartListening();
return;
}
#endregion
if (message == null)
{
LOG.Error("Message recieved is null");
StartListening();
return;
}
try
{
bool queued = ThreadPool.QueueUserWorkItem(new WaitCallback(ExecuteListener), message);
#region Could not enqueue
if (!queued)
{
LOG.Warn("Could not queue work item into thread pool. Retrying.");
Thread.Sleep(RecoveryTimeSpan);
}
#endregion
}
catch (Exception e)
{
LOG.Error("Error enqueue message in thread pool DefaultMessageQueue = [" + MessageQueue.QueueName + "]", e);
Thread.Sleep(RecoveryTimeSpan);
}
finally
{
StartListening();
}
}
protected virtual void ExecuteListener(object state)
{
Message message = state as Message;
try
{
DoExecuteListener(message);
}
catch (Exception e)
{
HandleListenerException(e, message);
}
}
private void HandleListenerException(Exception e, Message message)
{
IExceptionHandler exceptionHandler = ExceptionHandler;
if (exceptionHandler != null)
{
exceptionHandler.OnException(e, message);
}
}
}
}