(#35) Return 'ISubscription' for Subscribe and Respond

pardahlman · Mar 19, 2016 · 495dd70 · 495dd70
1 parent 6c73ed5
commit 495dd70
Show file tree

Hide file tree

Showing 8 changed files with 91 additions and 13 deletions.
diff --git a/src/RawRabbit/Common/BaseBusClient.cs b/src/RawRabbit/Common/BaseBusClient.cs
@@ -34,11 +34,11 @@ public BaseBusClient(
 			_logger.LogInformation("BusClient initialized.");
 		}
 
-		public void SubscribeAsync<T>(Func<T, TMessageContext, Task> subscribeMethod, Action<ISubscriptionConfigurationBuilder> configuration = null)
+		public ISubscription SubscribeAsync<T>(Func<T, TMessageContext, Task> subscribeMethod, Action<ISubscriptionConfigurationBuilder> configuration = null)
 		{
 			var config = _configEval.GetConfiguration<T>(configuration);
 			_logger.LogInformation($"Subscribing to message '{typeof(T).Name}' on exchange '{config.Exchange.ExchangeName}' with routing key {config.RoutingKey}.");
-			_subscriber.SubscribeAsync(subscribeMethod, config);
+			return _subscriber.SubscribeAsync(subscribeMethod, config);
 		}
 
 		public Task PublishAsync<T>(T message = default(T), Guid globalMessageId = new Guid(), Action<IPublishConfigurationBuilder> configuration = null)
@@ -48,11 +48,11 @@ public void SubscribeAsync<T>(Func<T, TMessageContext, Task> subscribeMethod, Ac
 			return _publisher.PublishAsync(message, globalMessageId, config);
 		}
 
-		public void RespondAsync<TRequest, TResponse>(Func<TRequest, TMessageContext, Task<TResponse>> onMessage, Action<IResponderConfigurationBuilder> configuration = null)
+		public ISubscription RespondAsync<TRequest, TResponse>(Func<TRequest, TMessageContext, Task<TResponse>> onMessage, Action<IResponderConfigurationBuilder> configuration = null)
 		{
 			var config = _configEval.GetConfiguration<TRequest, TResponse>(configuration);
 			_logger.LogInformation($"Responding to to requests '{typeof(TRequest).Name}' with '{typeof(TResponse).Name}'.");
-			_responder.RespondAsync(onMessage, config);
+			return _responder.RespondAsync(onMessage, config);
 		}
 
 		public Task<TResponse> RequestAsync<TRequest, TResponse>(TRequest message = default(TRequest), Guid globalMessageId = new Guid(), Action<IRequestConfigurationBuilder> configuration = null)

diff --git a/src/RawRabbit/Common/Subscription.cs b/src/RawRabbit/Common/Subscription.cs
@@ -0,0 +1,37 @@
+using System;
+using RabbitMQ.Client;
+using RawRabbit.Consumer.Abstraction;
+
+namespace RawRabbit.Common
+{
+	public interface ISubscription : IDisposable
+	{
+		string QueueName { get; }
+		string ConsumerTag { get; }
+	}
+
+	public class Subscription : ISubscription
+	{
+		public string QueueName { get; }
+		public string ConsumerTag { get; }
+
+		private readonly IRawConsumer _consumer;
+
+		public Subscription(IRawConsumer consumer, string queueName)
+		{
+			_consumer = consumer;
+			var basicConsumer = consumer as DefaultBasicConsumer;
+			if (basicConsumer == null)
+			{
+				return;
+			}
+			QueueName = queueName;
+			ConsumerTag = basicConsumer.ConsumerTag;
+		}
+
+		public void Dispose()
+		{
+			_consumer.Disconnect();
+		}
+	}
+}
diff --git a/src/RawRabbit/IBusClient.cs b/src/RawRabbit/IBusClient.cs
@@ -1,5 +1,6 @@
 using System;
 using System.Threading.Tasks;
+using RawRabbit.Common;
 using RawRabbit.Configuration.Publish;
 using RawRabbit.Configuration.Request;
 using RawRabbit.Configuration.Respond;
@@ -10,11 +11,11 @@ namespace RawRabbit
 {
 	public interface IBusClient<out TMessageContext> : IDisposable where TMessageContext : IMessageContext
 	{
-		void SubscribeAsync<T>(Func<T, TMessageContext, Task> subscribeMethod, Action<ISubscriptionConfigurationBuilder> configuration = null);
+		ISubscription SubscribeAsync<T>(Func<T, TMessageContext, Task> subscribeMethod, Action<ISubscriptionConfigurationBuilder> configuration = null);
 
 		Task PublishAsync<T>(T message = default(T), Guid globalMessageId = new Guid(), Action<IPublishConfigurationBuilder> configuration = null);
 
-		void RespondAsync<TRequest, TResponse>(Func<TRequest, TMessageContext, Task<TResponse>> onMessage, Action<IResponderConfigurationBuilder> configuration = null);
+		ISubscription RespondAsync<TRequest, TResponse>(Func<TRequest, TMessageContext, Task<TResponse>> onMessage, Action<IResponderConfigurationBuilder> configuration = null);
 
 		Task<TResponse> RequestAsync<TRequest, TResponse>(TRequest message = default(TRequest), Guid globalMessageId = new Guid(), Action<IRequestConfigurationBuilder> configuration = null);
 	}

diff --git a/src/RawRabbit/Operations/Abstraction/IResponder.cs b/src/RawRabbit/Operations/Abstraction/IResponder.cs
@@ -1,12 +1,13 @@
 using System;
 using System.Threading.Tasks;
+using RawRabbit.Common;
 using RawRabbit.Configuration.Respond;
 using RawRabbit.Context;
 
 namespace RawRabbit.Operations.Abstraction
 {
 	public interface IResponder<out TMessageContext> where TMessageContext : IMessageContext
 	{
-		void RespondAsync<TRequest, TResponse>(Func<TRequest, TMessageContext, Task<TResponse>> onMessage, ResponderConfiguration cfg);
+		ISubscription RespondAsync<TRequest, TResponse>(Func<TRequest, TMessageContext, Task<TResponse>> onMessage, ResponderConfiguration cfg);
 	}
 }
diff --git a/src/RawRabbit/Operations/Abstraction/ISubscriber.cs b/src/RawRabbit/Operations/Abstraction/ISubscriber.cs
@@ -1,12 +1,13 @@
 using System;
 using System.Threading.Tasks;
+using RawRabbit.Common;
 using RawRabbit.Configuration.Subscribe;
 using RawRabbit.Context;
 
 namespace RawRabbit.Operations.Abstraction
 {
 	public interface ISubscriber<out TMessageContext> where TMessageContext : IMessageContext
 	{
-		void SubscribeAsync<T>(Func<T, TMessageContext, Task> subscribeMethod, SubscriptionConfiguration config);
+		ISubscription SubscribeAsync<T>(Func<T, TMessageContext, Task> subscribeMethod, SubscriptionConfiguration config);
 	}
 }
diff --git a/src/RawRabbit/Operations/Responder.cs b/src/RawRabbit/Operations/Responder.cs
@@ -46,7 +46,7 @@ public Responder(
 			_consumers = new List<IRawConsumer>();
 		}
 
-		public void RespondAsync<TRequest, TResponse>(Func<TRequest, TMessageContext, Task<TResponse>> onMessage, ResponderConfiguration cfg)
+		public ISubscription RespondAsync<TRequest, TResponse>(Func<TRequest, TMessageContext, Task<TResponse>> onMessage, ResponderConfiguration cfg)
 		{
 			var topologyTask = _topologyProvider.BindQueueAsync(cfg.Queue, cfg.Exchange, cfg.RoutingKey);
 			var channelTask = _channelFactory.CreateChannelAsync();
@@ -87,9 +87,10 @@ public void RespondAsync<TRequest, TResponse>(Func<TRequest, TMessageContext, Ta
 							});
 					};
 					consumer.Model.BasicConsume(cfg.Queue.QueueName, cfg.NoAck, consumer);
+					return new Subscription(consumer, cfg.Queue.QueueName);
 				});
-
 			Task.WaitAll(respondTask);
+			return respondTask.Result;
 		}
 
 		public void Dispose()

diff --git a/src/RawRabbit/Operations/Subscriber.cs b/src/RawRabbit/Operations/Subscriber.cs
@@ -1,6 +1,5 @@
 using System;
 using System.Threading.Tasks;
-using RawRabbit.Channel;
 using RawRabbit.Channel.Abstraction;
 using RawRabbit.Common;
 using RawRabbit.Configuration.Subscribe;
@@ -40,7 +39,7 @@ public Subscriber(
 			_contextEnhancer = contextEnhancer;
 		}
 
-		public void SubscribeAsync<T>(Func<T, TMessageContext, Task> subscribeMethod, SubscriptionConfiguration config)
+		public ISubscription SubscribeAsync<T>(Func<T, TMessageContext, Task> subscribeMethod, SubscriptionConfiguration config)
 		{
 			var topologyTask = _topologyProvider.BindQueueAsync(config.Queue, config.Exchange, config.RoutingKey);
 			var channelTask = _channelFactory.CreateChannelAsync();
@@ -58,10 +57,11 @@ public void SubscribeAsync<T>(Func<T, TMessageContext, Task> subscribeMethod, Su
 						return subscribeMethod(body, context);
 					};
 					consumer.Model.BasicConsume(config.Queue.FullQueueName, config.NoAck, consumer);
-
 					_logger.LogDebug($"Setting up a consumer on channel '{channelTask.Result.ChannelNumber}' for queue {config.Queue.QueueName} with NoAck set to {config.NoAck}.");
+					return new Subscription(consumer, config.Queue.QueueName);
 				});
 			Task.WaitAll(subscriberTask);
+			return subscriberTask.Result;
 		}
 
 		public void Dispose()

diff --git a/test/RawRabbit.IntegrationTests/SimpleUse/PublishAndSubscribeTests.cs b/test/RawRabbit.IntegrationTests/SimpleUse/PublishAndSubscribeTests.cs
@@ -231,5 +231,42 @@ await subscriber.PublishAsync(new BasicMessage
 			/* Asset */
 			Assert.Equal(expected: "I am important!", actual: priorityTcs.Task.Result.Prop);
 		}
+
+		[Fact]
+		public async Task Should_Stop_Subscribe_When_Subscription_Is_Disposed()
+		{
+			/**/
+			var publisher = BusClientFactory.CreateDefault();
+			var subscriber = BusClientFactory.CreateDefault();
+			var firstMessage = new BasicMessage {Prop = "Value"};
+			var secondMessage = new BasicMessage {Prop = "AnotherValue"};
+			var  firstRecievedTcs = new TaskCompletionSource<BasicMessage>();
+			var  secondRecievedTcs = new TaskCompletionSource<BasicMessage>();
+			var recievedCount = 0;
+
+			var subscription = subscriber.SubscribeAsync<BasicMessage>((message, context) =>
+			{
+				recievedCount++;
+				firstRecievedTcs.SetResult(message);
+				return Task.FromResult(true);
+			});
+
+			await publisher.PublishAsync(firstMessage);
+			await firstRecievedTcs.Task;
+			subscription.Dispose();
+			await publisher.PublishAsync(secondMessage);
+			await Task.Delay(20);
+			publisher.SubscribeAsync<BasicMessage>((message, context) =>
+			{
+				secondRecievedTcs.SetResult(message);
+				return Task.FromResult(true);
+			});
+			await secondRecievedTcs.Task;
+
+			/* Assert */
+			Assert.Equal(1, recievedCount);
+			Assert.Equal(firstRecievedTcs.Task.Result.Prop, firstMessage.Prop);
+			Assert.Equal(secondRecievedTcs.Task.Result.Prop, secondMessage.Prop);
+		}
 	}
 }