Added NATS option to disable dynamic subject creation for consumers.

src/DotNetCore.CAP.NATS/CAP.NATSOptions.cs

@@ -25,6 +25,11 @@ public class NATSOptions
         /// </summary>
         public int ConnectionPoolSize { get; set; } = 10;
 
+        /// <summary>
+        /// Allows a nats consumer client to dynamically create a stream and configure the expected subjects on the stream. Defaults to true.
+        /// </summary>
+        public bool EnableSubscriberClientStreamAndSubjectCreation { get; set; } = true;
+
         /// <summary>
         /// Used to setup all NATs client options
         /// </summary>

src/DotNetCore.CAP.NATS/NATSConsumerClient.cs

@@ -43,37 +43,40 @@ public NATSConsumerClient(string groupName, byte groupConcurrent, IOptions<NATSO
 
         public ICollection<string> FetchTopics(IEnumerable<string> topicNames)
         {
-            Connect();
-
-            var jsm = _consumerClient!.CreateJetStreamManagementContext();
-
-            var streamGroup = topicNames.GroupBy(x => _natsOptions.NormalizeStreamName(x));
-
-            foreach (var subjectStream in streamGroup)
+            if (_natsOptions.EnableSubscriberClientStreamAndSubjectCreation)
             {
-                var builder = StreamConfiguration.Builder()
-                    .WithName(subjectStream.Key)
-                    .WithNoAck(false)
-                    .WithStorageType(StorageType.Memory)
-                    .WithSubjects(subjectStream.ToList());
+                Connect();
 
-                _natsOptions.StreamOptions?.Invoke(builder);
+                var jsm = _consumerClient!.CreateJetStreamManagementContext();
 
-                try
-                {
-                    jsm.GetStreamInfo(subjectStream.Key); // this throws if the stream does not exist
+                var streamSubjectsGroups = topicNames.GroupBy(x => _natsOptions.NormalizeStreamName(x));
 
-                    jsm.UpdateStream(builder.Build());
-                }
-                catch (NATSJetStreamException)
+                foreach (var streamSubjectsGroup in streamSubjectsGroups)
                 {
+                    var builder = StreamConfiguration.Builder()
+                        .WithName(streamSubjectsGroup.Key)
+                        .WithNoAck(false)
+                        .WithStorageType(StorageType.Memory)
+                        .WithSubjects(streamSubjectsGroup.ToList());
+
+                    _natsOptions.StreamOptions?.Invoke(builder);
+
                     try
                     {
-                        jsm.AddStream(builder.Build());
+                        jsm.GetStreamInfo(streamSubjectsGroup.Key); // this throws if the stream does not exist
+
+                        jsm.UpdateStream(builder.Build());
                     }
-                    catch
+                    catch (NATSJetStreamException)
                     {
-                        // ignored
+                        try
+                        {
+                            jsm.AddStream(builder.Build());
+                        }
+                        catch
+                        {
+                            // ignored
+                        }
                     }
                 }
             }