<commit_before>// Copyright 2012 <NAME>
//  
// 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.IO;
using System.Threading;
using Magnum.Extensions;
using MassTransit.AzureServiceBus;
using MassTransit.AzureServiceBus.Util;
using MassTransit.Logging;
using MassTransit.Transports.AzureServiceBus.Internal;
using Microsoft.ServiceBus.Messaging;

#pragma warning disable 1591

namespace MassTransit.Transports.AzureServiceBus
{
	/// <summary>
	/// 	Outbound transport targeting the azure service bus.
	/// </summary>
	public class OutboundTransportImpl
		: IOutboundTransport
	{
		const string NumberOfRetries = "busy-retries";
		static readonly ILog _logger = Logger.Get(typeof (OutboundTransportImpl));

		bool _disposed;

		int _messagesInFlight;
		int _sleeping;

		readonly AzureServiceBusEndpointAddress _address;
		readonly ConnectionHandler<ConnectionImpl> _connectionHandler;
		readonly SenderSettings _settings;

		/// <summary>
		/// 	c'tor
		/// </summary>
		public OutboundTransportImpl(
			[NotNull] AzureServiceBusEndpointAddress address,
			[NotNull] ConnectionHandler<ConnectionImpl> connectionHandler, 
			[NotNull] SenderSettings settings)
		{
			if (address == null) throw new ArgumentNullException("address");
			if (connectionHandler == null) throw new ArgumentNullException("connectionHandler");
			if (settings == null) throw new ArgumentNullException("settings");

			_connectionHandler = connectionHandler;
			_settings = settings;
			_address = address;

			_logger.DebugFormat("created outbound transport for address '{0}'", address);
		}

		public void Dispose()
		{
			if (_disposed) return;
			try
			{
				_address.Dispose();
				_connectionHandler.Dispose();
			}
			finally
			{
				_disposed = true;
			}
		}

		/// <summary>
		/// 	Gets the endpoint address this transport sends to.
		/// </summary>
		public IEndpointAddress Address
		{
			get { return _address; }
		}

		// service bus best practices for performance:
		// http://msdn.microsoft.com/en-us/library/windowsazure/hh528527.aspx
		public void Send(ISendContext context)
		{
			_connectionHandler
				.Use(connection =>
					{
						// don't have too many outstanding at same time
						SpinWait.SpinUntil(() => _messagesInFlight < _settings.MaxOutstanding);

						using (var body = new MemoryStream())
						{
							context.SerializeTo(body);
							
							// the envelope is re-usable, so let's capture it in the below closure
							// as a value
							var envelope = new MessageEnvelope(body.ToArray());

							TrySendMessage(connection, () =>
								{
									var brokeredMessage = new BrokeredMessage(envelope);

									if (!string.IsNullOrWhiteSpace(context.CorrelationId))
										brokeredMessage.CorrelationId = context.CorrelationId;

									if (!string.IsNullOrWhiteSpace(context.MessageId))
										brokeredMessage.MessageId = context.MessageId;
									
									return brokeredMessage;
								}, 1);
						}
					});
		}


		void TrySendMessage(ConnectionImpl connection, Func<BrokeredMessage> createMessage, int sendNumber)
		{
			var msg = createMessage();
			var messageId = msg.MessageId;
			var sender = connection.MessageSender;

			msg.Properties[NumberOfRetries] = sendNumber - 1;

			Address.LogBeginSend(msg.MessageId);

			Interlocked.Increment(ref _messagesInFlight);

			connection.MessageSender.BeginSend(msg, ar =>
				{
					Exception caught = null;

					// if the queue is deleted in the middle of things here, then I can't recover
					// at the moment; I have to extend the connection handler with an asynchronous
					// API to let it re-initialize the queue and hence maybe even the full transport...

					// So if I get MessagingEntityNotFoundException, I'm toast with this code: 
					// don't delete queues in use.

					Interlocked.Decrement(ref _messagesInFlight);

					try
					{
						sender.EndSend(ar);
						Address.LogEndSend(msg.MessageId);
					}
					catch (ServerBusyException ex)
					{
						_logger.Warn(string.Format("server busy, retrying for msg #{0}", messageId), ex);
						caught = ex;
					}
					catch (MessagingCommunicationException ex)
					{
						_logger.Warn(string.Format("server sad, retrying for msg #{0}", messageId), ex);
						caught = ex;
					}
					catch (Exception ex)
					{
						// this is really bad, we are now losing a message, but we can't recover in any way
						_logger.Fatal(string.Format("other exception for msg #{0} - giving up", messageId), ex);
						// intentionally not setting 'caught' variable
					}
<commit_msg>catching timeout exception<commit_after>// Copyright 2012 <NAME>
//  
// 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.IO;
using System.Threading;
using Magnum.Extensions;
using MassTransit.AzureServiceBus;
using MassTransit.AzureServiceBus.Util;
using MassTransit.Logging;
using MassTransit.Transports.AzureServiceBus.Internal;
using Microsoft.ServiceBus.Messaging;

#pragma warning disable 1591

namespace MassTransit.Transports.AzureServiceBus
{
	/// <summary>
	/// 	Outbound transport targeting the azure service bus.
	/// </summary>
	public class OutboundTransportImpl
		: IOutboundTransport
	{
		const string NumberOfRetries = "busy-retries";
		static readonly ILog _logger = Logger.Get(typeof (OutboundTransportImpl));

		bool _disposed;

		int _messagesInFlight;
		int _sleeping;

		readonly AzureServiceBusEndpointAddress _address;
		readonly ConnectionHandler<ConnectionImpl> _connectionHandler;
		readonly SenderSettings _settings;

		/// <summary>
		/// 	c'tor
		/// </summary>
		public OutboundTransportImpl(
			[NotNull] AzureServiceBusEndpointAddress address,
			[NotNull] ConnectionHandler<ConnectionImpl> connectionHandler, 
			[NotNull] SenderSettings settings)
		{
			if (address == null) throw new ArgumentNullException("address");
			if (connectionHandler == null) throw new ArgumentNullException("connectionHandler");
			if (settings == null) throw new ArgumentNullException("settings");

			_connectionHandler = connectionHandler;
			_settings = settings;
			_address = address;

			_logger.DebugFormat("created outbound transport for address '{0}'", address);
		}

		public void Dispose()
		{
			if (_disposed) return;
			try
			{
				_address.Dispose();
				_connectionHandler.Dispose();
			}
			finally
			{
				_disposed = true;
			}
		}

		/// <summary>
		/// 	Gets the endpoint address this transport sends to.
		/// </summary>
		public IEndpointAddress Address
		{
			get { return _address; }
		}

		// service bus best practices for performance:
		// http://msdn.microsoft.com/en-us/library/windowsazure/hh528527.aspx
		public void Send(ISendContext context)
		{
			_connectionHandler
				.Use(connection =>
					{
						// don't have too many outstanding at same time
						SpinWait.SpinUntil(() => _messagesInFlight < _settings.MaxOutstanding);

						using (var body = new MemoryStream())
						{
							context.SerializeTo(body);
							
							// the envelope is re-usable, so let's capture it in the below closure
							// as a value
							var envelope = new MessageEnvelope(body.ToArray());

							TrySendMessage(connection, () =>
								{
									var brokeredMessage = new BrokeredMessage(envelope);

									if (!string.IsNullOrWhiteSpace(context.CorrelationId))
										brokeredMessage.CorrelationId = context.CorrelationId;

									if (!string.IsNullOrWhiteSpace(context.MessageId))
										brokeredMessage.MessageId = context.MessageId;
									
									return brokeredMessage;
								}, 1);
						}
					});
		}


		void TrySendMessage(ConnectionImpl connection, Func<BrokeredMessage> createMessage, int sendNumber)
		{
			var msg = createMessage();
			var messageId = msg.MessageId;
			var sender = connection.MessageSender;

			msg.Properties[NumberOfRetries] = sendNumber - 1;

			Address.LogBeginSend(msg.MessageId);

			Interlocked.Increment(ref _messagesInFlight);

			connection.MessageSender.BeginSend(msg, ar =>
				{
					Exception caught = null;

					// if the queue is deleted in the middle of things here, then I can't recover
					// at the moment; I have to extend the connection handler with an asynchronous
					// API to let it re-initialize the queue and hence maybe even the full transport...

					// So if I get MessagingEntityNotFoundException, I'm toast with this code: 
					// don't delete queues in use.

					Interlocked.Decrement(ref _messagesInFlight);

					try
					{
						sender.EndSend(ar);
						Address.LogEndSend(msg.MessageId);
					}
					catch (ServerBusyException ex)
					{
						_logger.Warn(string.Format("server busy, retrying for msg #{0}", messageId), ex);
						caught = ex;
					}
					catch (MessagingCommunicationException ex)
					{
						_logger.Warn(string.Format("server sad, retrying for msg #{0}", messageId), ex);
						caught = ex;
					}
					catch(TimeoutException ex)
					{
						_logger.Info(string.Format("server timed out for msg #{0}", messageId), ex);
						caught = ex;
					}
					catch (Exception ex)
					{
						// this is really bad, we are now losing a message, but we can't recover in any way
						_logger.Fatal(string.Format("other exception for msg #{0} - giving up", messageId), ex);
						// intentionally not setting 'caught' variable
					}

{'fraction_non_alphanumeric': 0.09544173389056274, 'fraction_numerical': 0.004762367577024006, 'mean_word_length': 3.171057965139846, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 31, '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': '5012611', 'n_tokens_mistral': 3694, 'n_tokens_neox': 2846, 'n_words': 876}