<commit_before>/// Copyright 2007-2008 The Apache Software Foundation.
/// 
/// 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.

using System;
using System.Messaging;
using System.Threading;
using log4net;
using MassTransit.ServiceBus.Exceptions;

namespace MassTransit.ServiceBus
{
	/// <summary>
	/// Receives envelopes from a message queue
	/// </summary>
	public class MessageQueueReceiver :
		IMessageReceiver
	{
		private static readonly object _locker = new object();
		private static readonly ILog _log = LogManager.GetLogger(typeof (MessageQueueReceiver));
		private readonly IMessageQueueEndpoint _endpoint;

		private IEnvelopeConsumer _consumer;

		private MessageQueue _queue;

		/// <summary>
		/// Initializes a MessageQueueReceiver
		/// </summary>
		/// <param name="endpoint">The endpoint where the receiver should be attached</param>
		public MessageQueueReceiver(IMessageQueueEndpoint endpoint)
		{
			_endpoint = endpoint;
		}

		#region IMessageReceiver Members

		///<summary>
		///Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
		///</summary>
		///<filterpriority>2</filterpriority>
		public void Dispose()
		{
			Stop();
		}

		/// <summary>
		/// Adds a consumer to the message receiver
		/// </summary>
		/// <param name="consumer">The consumer to add</param>
		public void Subscribe(IEnvelopeConsumer consumer)
		{
			lock (_locker)
			{
				if (_consumer == null)
				{
					_consumer = consumer;

					Restart();
				}
				else if (_consumer != consumer)
				{
					throw new EndpointException(_endpoint, "Only one consumer can be registered for a message receiver");
				}
			}
		}

		#endregion

		private void Restart()
		{
			lock (this)
			{
				Stop();

				Start();
			}
		}

		private void Start()
		{
			_queue = _endpoint.Open(QueueAccessMode.SendAndReceive);

			ThreadPool.QueueUserWorkItem(MonitorQueue, null);
		}

		private void Stop()
		{
			if (_queue != null)
			{
				_queue.Close();
				_queue.Dispose();
				_queue = null;
			}
		}

		/// <summary>
		/// Called by the thread pool to process a message that has been seen on the queue (via Peek)
		/// 
		/// The message is checked to see if it will handled by a consumer and if so, the message
		/// is received from the queue and send to the consumer(s) that will handle it.
		/// </summary>
		/// <param name="obj">An instance of <c ref="Message" /> that was seen on the queue</param>
		public void ProcessMessage(object obj)
		{
			IEnvelope e = obj as IEnvelope;
			if (e == null)
				return;
			try
			{
				_consumer.Deliver(e);
			}
			catch (Exception ex)
			{
				if (_log.IsErrorEnabled)
					_log.Error(string.Format("An exception occured delivering envelope {0}", e.Id), ex);
			}
		}

		private void MonitorQueue(object obj)
		{
			MessageQueue queue = _queue;
			if (queue == null)
				return;

			try
			{
				Cursor cursor = _queue.CreateCursor();

				Message msg = queue.Peek(TimeSpan.FromSeconds(1), cursor, PeekAction.Current);

				while (msg != null)
				{
					if (_log.IsDebugEnabled)
						_log.DebugFormat("Queue: {0} Received Message Id {1}", _queue.Path, msg.Id);

					if (_consumer != null)
					{
						IEnvelope e = EnvelopeMessageMapper.MapFrom(msg);

						if (_consumer.IsHandled(e))
						{
							Message received = queue.Receive(TimeSpan.FromSeconds(3), cursor);

							if (received.Id == msg.Id)
							{
								ThreadPool.QueueUserWorkItem(ProcessMessage, e);
								break;
							}
						}
					}

					msg = queue.Peek(TimeSpan.FromSeconds(1), cursor, PeekAction.Next);
				}
			}
			catch (MessageQueueException ex)
			{
				if (ex.MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout)
				{
				}
				else
				{
					_log.Error("An error occured while communicating with the queue", ex);
				}
			}
			catch(Exception ex)
			{
				_log.Error("An unknown exception occured", ex);
			}

			ThreadPool.QueueUserWorkItem(MonitorQueue, obj);
		}
	}
}<commit_msg>trying to expose the exceptions better<commit_after>/// Copyright 2007-2008 The Apache Software Foundation.
/// 
/// 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.

using System;
using System.Messaging;
using System.Threading;
using log4net;
using MassTransit.ServiceBus.Exceptions;

namespace MassTransit.ServiceBus
{
	/// <summary>
	/// Receives envelopes from a message queue
	/// </summary>
	public class MessageQueueReceiver :
		IMessageReceiver
	{
		private static readonly object _locker = new object();
		private static readonly ILog _log = LogManager.GetLogger(typeof (MessageQueueReceiver));
		private readonly IMessageQueueEndpoint _endpoint;

		private IEnvelopeConsumer _consumer;

		private MessageQueue _queue;

		/// <summary>
		/// Initializes a MessageQueueReceiver
		/// </summary>
		/// <param name="endpoint">The endpoint where the receiver should be attached</param>
		public MessageQueueReceiver(IMessageQueueEndpoint endpoint)
		{
			_endpoint = endpoint;
		}

		#region IMessageReceiver Members

		///<summary>
		///Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
		///</summary>
		///<filterpriority>2</filterpriority>
		public void Dispose()
		{
			Stop();
		}

		/// <summary>
		/// Adds a consumer to the message receiver
		/// </summary>
		/// <param name="consumer">The consumer to add</param>
		public void Subscribe(IEnvelopeConsumer consumer)
		{
			lock (_locker)
			{
				if (_consumer == null)
				{
					_consumer = consumer;

					Restart();
				}
				else if (_consumer != consumer)
				{
					throw new EndpointException(_endpoint, "Only one consumer can be registered for a message receiver");
				}
			}
		}

		#endregion

		private void Restart()
		{
			lock (this)
			{
				Stop();

				Start();
			}
		}

		private void Start()
		{
			_queue = _endpoint.Open(QueueAccessMode.SendAndReceive);

			ThreadPool.QueueUserWorkItem(MonitorQueue, null);
		}

		private void Stop()
		{
			if (_queue != null)
			{
				_queue.Close();
				_queue.Dispose();
				_queue = null;
			}
		}

		/// <summary>
		/// Called by the thread pool to process a message that has been seen on the queue (via Peek)
		/// 
		/// The message is checked to see if it will handled by a consumer and if so, the message
		/// is received from the queue and send to the consumer(s) that will handle it.
		/// </summary>
		/// <param name="obj">An instance of <c ref="Message" /> that was seen on the queue</param>
		public void ProcessMessage(object obj)
		{
			IEnvelope e = obj as IEnvelope;
			if (e == null)
				return;
			try
			{
				_consumer.Deliver(e);
			}
			catch (Exception ex)
			{
				if (_log.IsErrorEnabled)
					_log.Error(string.Format("An exception occured delivering envelope {0}", e.Id), ex);
			}
		}

		private void MonitorQueue(object obj)
		{
			MessageQueue queue = _queue;
			if (queue == null)
				return;

			try
			{
				Cursor cursor = _queue.CreateCursor();

				Message msg = queue.Peek(TimeSpan.FromSeconds(1), cursor, PeekAction.Current);

				while (msg != null)
				{
					if (_log.IsDebugEnabled)
						_log.DebugFormat("Queue: {0} Received Message Id {1}", _queue.Path, msg.Id);

					if (_consumer != null)
					{
						IEnvelope e = EnvelopeMessageMapper.MapFrom(msg);

						if (_consumer.IsHandled(e))
						{
							Message received = queue.Receive(TimeSpan.FromSeconds(3), cursor);

							if (received.Id == msg.Id)
							{
								ThreadPool.QueueUserWorkItem(ProcessMessage, e);
								break;
							}
						}
					}

					msg = queue.Peek(TimeSpan.FromSeconds(1), cursor, PeekAction.Next);
				}
			}
			catch (MessageQueueException ex)
			{
				if (ex.MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout)
				{
                    if (_log.IsErrorEnabled)
                        _log.Error("IO Timeout", ex);
				}
				else
				{
                    if (_log.IsErrorEnabled)
					    _log.Error("An error occured while communicating with the queue", ex);
				}

                throw new EndpointException(_endpoint, "message", ex);
			}
			catch(Exception ex)
			{
                if(_log.IsErrorEnabled)
				    _log.Error("An unknown exception occured", ex);

                throw new EndpointException(_endpoint, "message", ex);
			}

			ThreadPool.QueueUserWorkItem(MonitorQueue, obj);
		}
	}
}
{'fraction_non_alphanumeric': 0.1114564355374558, 'fraction_numerical': 0.004286785982209838, 'mean_word_length': 2.7720291026677444, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2970933', 'n_tokens_mistral': 3406, 'n_tokens_neox': 2712, 'n_words': 849}