Added Node.DetachAsync

- Added ability to detach as a future
- Minor internal fixes
- Moved to 0.1.9

Author: alandoherty

samples/Example.General/Program.cs

@@ -77,9 +77,16 @@ static async Task AsyncMain(string[] args) {
             });
 
             // attach
-            //Service service = await TestNode.AttachAsync("auth:login", RpcBehaviour.Bind<ITest001>(new Test001(Guid.NewGuid())));
-
-            //service.Dispose();
+            Service service = null;
+            
+            try {
+                service = await TestNode.AttachAsync("auth:login", ServiceType.Balanced, RpcBehaviour.Bind<ITest001>(new Test001(Guid.NewGuid())));
+            } catch(Exception) {
+                Console.WriteLine();
+            }
+
+            // detaches the service
+            await TestNode.DetachAsync(service).ConfigureAwait(false);
 
             await Task.Delay(50000);
         }

src/Holon/Holon.csproj

@@ -2,19 +2,19 @@
 
   <PropertyGroup>
     <TargetFramework>netstandard1.6</TargetFramework>
-    <Version>0.1.8</Version>
+    <Version>0.1.9</Version>
     <Authors>Alan Doherty</Authors>
     <Company>Alan Doherty</Company>
     <Description>A minimal service and event bus with additional support for RPC</Description>
     <Copyright>BattleCrate Ltd 2018</Copyright>
     <PackageProjectUrl>https://github.com/alandoherty/holon-net</PackageProjectUrl>
     <RepositoryUrl>https://github.com/alandoherty/holon-net</RepositoryUrl>
     <RepositoryType>git</RepositoryType>
-    <AssemblyVersion>0.1.8.0</AssemblyVersion>
+    <AssemblyVersion>0.1.9.0</AssemblyVersion>
     <PackageLicenseUrl>https://github.com/alandoherty/holon-net/blob/master/LICENSE</PackageLicenseUrl>
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
     <PackageIconUrl>https://s3-eu-west-1.amazonaws.com/assets.alandoherty.co.uk/github/holon-net-nuget.png</PackageIconUrl>
-    <FileVersion>0.1.8.0</FileVersion>
+    <FileVersion>0.1.9.0</FileVersion>
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|AnyCPU'">

src/Holon/Node.cs

@@ -565,6 +565,24 @@ public void Detach(Service service) {
             service.Dispose();
         }
 
+        /// <summary>
+        /// Detaches a service from the node, waiting for complete cancellation.
+        /// </summary>
+        /// <param name="service">The service.</param>
+        /// <returns></returns>
+        public async Task DetachAsync(Service service) {
+            // remove service
+            lock (_services) {
+                if (!_services.Contains(service))
+                    return;
+
+                // remove from list
+                _services.Remove(service);
+            }
+
+            await service.ShutdownAsync().ConfigureAwait(false);
+        }
+
         /// <summary>
         /// Disposes the node and underlying services.
         /// </summary>

src/Holon/Protocol/Broker.cs

@@ -174,7 +174,7 @@ public async Task<BrokerQueue> CreateQueueAsync(string name = "", bool durable =
             }).ConfigureAwait(false);
 
             // create consumer
-            ObservableConsumer consumer = new ObservableConsumer(_channel);
+            ObservableConsumer consumer = new ObservableConsumer(this);
 
             // consume queue
             string consumerTag = null;
@@ -234,11 +234,11 @@ internal Broker(BrokerContext ctx, IModel channel, string appId) {
                 OnReturned(new BrokerReturnedEventArgs(e.BasicProperties, e.Body, e.ReplyText));
             };
             _channel.ModelShutdown += delegate (object s, ShutdownEventArgs e) {
-                // dispose
-                Dispose();
-
                 // call event
                 OnShutdown(new BrokerShutdownEventArgs(e.ReplyText));
+
+                // dispose
+                Dispose();
             };
         }
         #endregion

src/Holon/Protocol/BrokerContext.cs

@@ -25,7 +25,14 @@ internal sealed class BrokerContext : IDisposable
         #endregion
 
         #region Properties
-
+        /// <summary>
+        /// Gets the connection.
+        /// </summary>
+        public IConnection Connection {
+            get {
+                return _connection;
+            }
+        }
         #endregion
 
         #region Methods

src/Holon/Protocol/BrokerQueue.cs

@@ -110,6 +110,18 @@ public void Unbind(string exchange, string routingKey) {
             });
         }
 
+        /// <summary>
+        /// Cancels the broker queue.
+        /// </summary>
+        /// <returns></returns>
+        public Task CancelAsync() {
+            _consumer = null;
+            return _broker.Context.AskWork(() => {
+                _broker.Channel.BasicCancel(_consumerTag);
+                return null;
+            });
+        }
+
         /// <summary>
         /// Disposes 
         /// </summary>

src/Holon/Protocol/ObservableConsumer.cs

@@ -15,17 +15,26 @@ namespace Holon.Protocol
     internal class ObservableConsumer : IBasicConsumer, IObservable<InboundMessage>
     {
         #region Fields
-        private IModel _channel;
+        private Broker _broker;
         private int _cancelled;
         private List<SubscribedObserver> _subscriptions = new List<SubscribedObserver>(1);
         #endregion
 
+        /// <summary>
+        /// Gets the underlying broker.
+        /// </summary>
+        public Broker Broker {
+            get {
+                return _broker;
+            }
+        }
+
         /// <summary>
         /// Gets the underlying channel.
         /// </summary>
         public IModel Model {
             get {
-                return _channel;
+                return _broker.Channel;
             }
         }
 
@@ -87,7 +96,7 @@ public void HandleBasicConsumeOk(string consumerTag) {
         public void HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, byte[] body) {
             lock(_subscriptions) {
                 foreach (SubscribedObserver observer in _subscriptions)
-                    observer.Observer.OnNext(new InboundMessage(_channel, deliveryTag, redelivered, exchange, routingKey, properties, body));
+                    observer.Observer.OnNext(new InboundMessage(Model, deliveryTag, redelivered, exchange, routingKey, properties, body));
             }
         }
 
@@ -170,9 +179,9 @@ struct SubscribedObserver
         /// <summary>
         /// Creates a new observable consumer.
         /// </summary>
-        /// <param name="channel">The channel.</param>
-        public ObservableConsumer(IModel channel) {
-            _channel = channel;
+        /// <param name="broker">The broker.</param>
+        public ObservableConsumer(Broker broker) {
+            _broker = broker;
         }
         #endregion
     }

src/Holon/Services/Service.cs

@@ -155,13 +155,24 @@ public void Dispose() {
                 return;
 
             // dispose of queue
-            _queue.Unbind(_addr.Namespace, _addr.RoutingKey);
             _queue.Dispose();
 
             // detach from node
             _namespace.Node.Detach(this);
         }
 
+        /// <summary>
+        /// Shutdown the service.
+        /// </summary>
+        /// <returns></returns>
+        internal async Task ShutdownAsync() {
+            // cancel
+            await _queue.CancelAsync().ConfigureAwait(false);
+
+            // dispose
+            Dispose();
+        }
+
         /// <summary>
         /// Creates the queue and internal consumer for this service.
         /// </summary>