Add Message.Respond implementing #281

NATS.Client/AsyncSub.cs

@@ -131,13 +131,16 @@ internal void enableAsyncProcessing()
 
         internal void disableAsyncProcessing()
         {
-            if (msgFeeder != null)
+            lock (mu)
             {
-                mch.close();               
-                msgFeeder = null;
+                if (msgFeeder != null)
+                {
+                    mch.close();
+                    msgFeeder = null;
+                }
+                MessageHandler = null;
+                started = false;
             }
-            MessageHandler = null;
-            started = false;
         }
 
         /// <summary>
@@ -155,11 +158,14 @@ public void Start()
             if (started)
                 return;
 
-            if (conn == null)
-                throw new NATSBadSubscriptionException();
+            lock (mu)
+            {
+                if (conn == null)
+                    throw new NATSBadSubscriptionException();
 
-            conn.sendSubscriptionMessage(this);
-            enableAsyncProcessing();
+                conn.sendSubscriptionMessage(this);
+                enableAsyncProcessing();
+            }
         }
 
         /// <summary>

NATS.Client/Conn.cs

@@ -3485,7 +3485,6 @@ internal virtual void removeSub(Subscription s)
                 s.mch = null;
             }
 
-            s.conn = null;
             s.closed = true;
         }
 

NATS.Client/Msg.cs

@@ -167,6 +167,31 @@ public ISubscription ArrivalSubcription
             get { return sub; }
         }
 
+        /// <summary>
+        /// Send a response to the message on the arrival subscription.
+        /// </summary>
+        /// <param name="data">The response payload to send.</param>
+        /// <exception cref="NATSException">
+        /// <para><see cref="Reply"/> is null or empty.</para>
+        /// <para>-or-</para>
+        /// <para><see cref="ArrivalSubcription"/> is null.</para>
+        /// </exception>
+        public void Respond(byte[] data)
+        {
+            if (String.IsNullOrEmpty(Reply))
+            {
+                throw new NATSException("No Reply subject");
+            }
+
+            Connection conn = ArrivalSubcription?.Connection;
+            if (conn == null)
+            {
+                throw new NATSException("Message is not bound to a subscription");
+            }
+
+            conn.Publish(this.Reply, data);
+        }
+
         /// <summary>
         /// Generates a string representation of the messages.
         /// </summary>

NATS.Client/Subscription.cs

@@ -222,20 +222,22 @@ public bool IsValid
             {
                 lock (mu)
                 {
-                    return (conn != null);
+                    return (conn != null) && !closed;
                 }
             }
         }
 
         internal void unsubscribe(bool throwEx)
         {
             Connection c;
+            bool isClosed;
             lock (mu)
             {
                 c = this.conn;
+                isClosed = this.closed;
             }
 
-            if (c == null)
+            if (c == null || closed)
             {
                 if (throwEx)
                     throw new NATSBadSubscriptionException();
@@ -281,7 +283,7 @@ public virtual void AutoUnsubscribe(int max)
 
             lock (mu)
             {
-                if (conn == null)
+                if (conn == null || closed)
                     throw new NATSBadSubscriptionException();
 
                 c = conn;
@@ -301,7 +303,7 @@ public int QueuedMessageCount
             {
                 lock (mu)
                 {
-                    if (conn == null)
+                    if (conn == null || closed)
                         throw new NATSBadSubscriptionException();
 
                     return mch.Count;
@@ -332,6 +334,9 @@ protected virtual void Dispose(bool disposing)
                     // auto unsubscribing, so ignore.
                 }
 
+                conn = null;
+                closed = true;
+
                 disposedValue = true;
             }
         }
@@ -370,9 +375,8 @@ public override string ToString()
 
         private void checkState()
         {
-            if (conn == null)
+            if (conn == null || closed)
                 throw new NATSBadSubscriptionException();
-
         }
 
         /// <summary>
@@ -573,7 +577,7 @@ internal Task InternalDrain(int timeout)
 
             lock (mu)
             {
-                if (conn == null)
+                if (conn == null || closed)
                     throw new NATSBadSubscriptionException();
 
                 c = conn;
@@ -595,7 +599,6 @@ public Task DrainAsync(int timeout)
             return InternalDrain(timeout);
         }
 
-
         public void Drain()
         {
            Drain(Defaults.DefaultDrainTimeout);

NATSUnitTests/UnitTestSub.cs

@@ -532,7 +532,6 @@ public void TestAsyncSubscriptionPending()
             }
         }
 
-
         [Fact]
         public void TestAsyncPendingSubscriptionBatchSizeExactlyOne()
         {
@@ -1040,6 +1039,70 @@ public void TestInvalidSubjects()
                 }
             }
         }
+
+        [Fact]
+        public void TestRespond()
+        {
+            using (new NATSServer())
+            {
+                using (IConnection c = utils.DefaultTestConnection)
+                {
+                    ISyncSubscription s = c.SubscribeSync("foo");
+
+                    string replyTo = c.NewInbox();
+                    ISyncSubscription r = c.SubscribeSync(replyTo);
+
+                    c.Publish("foo", replyTo, Encoding.UTF8.GetBytes("message"));
+
+                    Msg m = s.NextMessage(1000);
+                    Assert.NotNull(m);
+                    Assert.Equal(replyTo, m.Reply);
+
+                    byte[] reply = Encoding.UTF8.GetBytes("reply");
+                    m.Respond(reply);
+
+                    m = r.NextMessage(1000);
+                    Assert.NotNull(m);
+                    Assert.Equal(replyTo, m.Subject);
+                    Assert.Equal(reply, m.Data);
+
+                    s.Unsubscribe();
+                    r.Unsubscribe();
+                }
+            }
+        }
+
+        [Fact]
+        public void TestRespondWithAutoUnsubscribe()
+        {
+            using (new NATSServer())
+            {
+                using (IConnection c = utils.DefaultTestConnection)
+                {
+                    ISyncSubscription s = c.SubscribeSync("foo");
+                    s.AutoUnsubscribe(1);
+
+                    string replyTo = c.NewInbox();
+                    ISyncSubscription r = c.SubscribeSync(replyTo);
+
+                    c.Publish("foo", replyTo, Encoding.UTF8.GetBytes("message"));
+
+                    Msg m = s.NextMessage(1000);
+                    Assert.NotNull(m);
+                    Assert.Equal(replyTo, m.Reply);
+
+                    byte[] reply = Encoding.UTF8.GetBytes("reply");
+                    m.Respond(reply);
+
+                    m = r.NextMessage(1000);
+                    Assert.NotNull(m);
+                    Assert.Equal(replyTo, m.Subject);
+                    Assert.Equal(reply, m.Data);
+
+                    r.Unsubscribe();
+                }
+            }
+        }
     }
 }