subscription before consumer (#597)

nats-io · May 4, 2022 · 6b8c68e · 6b8c68e
1 parent 7788937
commit 6b8c68e
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 6 deletions.
diff --git a/src/NATS.Client/JetStream/JetStream.cs b/src/NATS.Client/JetStream/JetStream.cs
@@ -264,7 +264,8 @@ Subscription CreateSubscription(string subject, string queueName,
                 inboxDeliver = Conn.NewInbox();
             }
 
-            // 5. If consumer does not exist, create
+            // 5. If consumer does not exist, create and settle on the config. Name will have to wait
+            //    If the consumer exists, I know what the settled info is
             if (serverCC == null) {
                 ConsumerConfiguration.ConsumerConfigurationBuilder ccBuilder = ConsumerConfiguration.Builder(userCC);
 
@@ -284,9 +285,8 @@ Subscription CreateSubscription(string subject, string queueName,
                 }
 
                 // createOrUpdateConsumer can fail for security reasons, maybe other reasons?
-                ConsumerInfo ci = AddOrUpdateConsumerInternal(stream, ccBuilder.Build());
-                consumerName = ci.Name;
-                serverCC = ci.ConsumerConfiguration;
+                serverCC = ccBuilder.Build();
+                consumerName = null;
             }
 
             // 6. create the subscription
@@ -345,6 +345,30 @@ AsyncSubscription CreateAsyncSubDelegate(Connection lConn, string lSubject, stri
             }
 
             asm.SetSub(sub);
+
+            // 7. The consumer might need to be created, do it here
+            if (consumerName == null)
+            {
+                try
+                {
+                    ConsumerInfo ci = AddOrUpdateConsumerInternal(stream, serverCC);
+                    if (sub is JetStreamAbstractSyncSubscription syncSub)
+                    {
+                        syncSub.Consumer = ci.Name;
+                    }
+                    else if (sub is JetStreamPushAsyncSubscription asyncSub)
+                    {
+                        asyncSub.Consumer = ci.Name;
+                    }
+                }
+                catch
+                {
+                    // create consumer can fail, unsubscribe and then throw the exception to the user
+                    sub.Unsubscribe();
+                    throw;
+                }
+            }
+
             return sub;
         }
 

diff --git a/src/NATS.Client/JetStream/JetStreamAbstractSyncSubscription.cs b/src/NATS.Client/JetStream/JetStreamAbstractSyncSubscription.cs
@@ -20,7 +20,7 @@ public class JetStreamAbstractSyncSubscription : SyncSubscription
         internal IAutoStatusManager _asm;
         public JetStream Context { get; }
         public string Stream { get; }
-        public string Consumer { get; }
+        public string Consumer { get; internal set; }
         public string DeliverSubject { get; }
 
         internal JetStreamAbstractSyncSubscription(Connection conn, string subject, string queue,

diff --git a/src/NATS.Client/JetStream/JetStreamPushAsyncSubscription.cs b/src/NATS.Client/JetStream/JetStreamPushAsyncSubscription.cs
@@ -18,7 +18,7 @@ public class JetStreamPushAsyncSubscription : AsyncSubscription, IJetStreamPushA
         private IAutoStatusManager _asm;
         public JetStream Context { get; }
         public string Stream { get; }
-        public string Consumer { get; }
+        public string Consumer { get; internal set; }
         public string DeliverSubject { get; }
 
         internal JetStreamPushAsyncSubscription(Connection conn, string subject, string queue,