(#301) Adjust channel scaling + recovery

This commit fixes a potential DevidedByZeroException when channel
workload is calculated in the auto scaling channel pool.

It also passes a cancellation token to the autoscaling, making it so
that it the channel is not created within a reasonable time span, the
task is cancelled.

Author: pardahlman

src/RawRabbit/Channel/AutoScalingChannelPool.cs

@@ -65,20 +65,25 @@ public void SetupScaling()
 
 			_timer = new Timer(state =>
 			{
-				var workPerChannel = ChannelRequestQueue.Count / Pool.Count;
+				var workPerChannel = Pool.Count == 0 ? int.MaxValue : ChannelRequestQueue.Count / Pool.Count;
 				var scaleUp = Pool.Count < _options.MaximumPoolSize;
 				var scaleDown = _options.MinimunPoolSize < Pool.Count;
 
 				_logger.Debug("Channel pool currently has {channelCount} channels open and a total workload of {totalWorkload}", Pool.Count, ChannelRequestQueue.Count);
 				if (scaleUp && _options.DesiredAverageWorkload < workPerChannel)
 				{
 					_logger.Debug("The estimated workload is {averageWorkload} operations/channel, which is higher than the desired workload ({desiredAverageWorkload}). Creating channel.", workPerChannel, _options.DesiredAverageWorkload);
+
+					var channelCancellation = new CancellationTokenSource(_options.RefreshInterval);
 					_factory
-						.CreateChannelAsync()
+						.CreateChannelAsync(channelCancellation.Token)
 						.ContinueWith(tChannel =>
 						{
-							Add(tChannel.Result);
-						});
+							if (tChannel.Status == TaskStatus.RanToCompletion)
+							{
+								Add(tChannel.Result);
+							}
+						}, CancellationToken.None);
 					return;
 				}
 

src/RawRabbit/Channel/ChannelFactory.cs

@@ -32,6 +32,8 @@ public ChannelFactory(IConnectionFactory connectionFactory, RawRabbitConfigurati
 			{
 				_logger.Debug("Creating a new connection for {hostNameCount} hosts.", ClientConfig.Hostnames.Count);
 				Connection = ConnectionFactory.CreateConnection(ClientConfig.Hostnames);
+				Connection.ConnectionShutdown += (sender, args) =>
+					_logger.Warn("Connection was shutdown by {Initiator}. ReplyText {ReplyText}", args.Initiator, args.ReplyText);
 			}
 			catch (BrokerUnreachableException e)
 			{
@@ -43,8 +45,8 @@ public ChannelFactory(IConnectionFactory connectionFactory, RawRabbitConfigurati
 
 		public virtual async Task<IModel> CreateChannelAsync(CancellationToken token = default(CancellationToken))
 		{
-			token.ThrowIfCancellationRequested();
 			var connection = await GetConnectionAsync(token);
+			token.ThrowIfCancellationRequested();
 			var channel = connection.CreateModel();
 			Channels.Add(channel);
 			return channel;
@@ -80,11 +82,16 @@ public ChannelFactory(IConnectionFactory connectionFactory, RawRabbitConfigurati
 
 			_logger.Debug("Connection is recoverable. Waiting for 'Recovery' event to be triggered. ");
 			var recoverTcs = new TaskCompletionSource<IConnection>();
+			token.Register(() => recoverTcs.SetCanceled());
 
 			EventHandler<EventArgs> completeTask = null;
 			completeTask = (sender, args) =>
 			{
-				_logger.Debug("Connection has been recovered!");
+				if (recoverTcs.Task.IsCanceled)
+				{
+					return;
+				}
+				_logger.Info("Connection has been recovered!");
 				recoverTcs.TrySetResult(recoverable as IConnection);
 				recoverable.Recovery -= completeTask;
 			};