Fix nats_jetstream input stream without subject

redpanda-data · May 10, 2024 · 44f3071 · 44f3071
1 parent 88c01d1
commit 44f3071
Show file tree

Hide file tree

Showing 2 changed files with 43 additions and 2 deletions.
diff --git a/internal/impl/nats/input_jetstream.go b/internal/impl/nats/input_jetstream.go
@@ -180,8 +180,8 @@ func newJetStreamReaderFromConfig(conf *service.ParsedConfig, mgr *service.Resou
 			return nil, errors.New("stream or durable is required, when bind is true")
 		}
 	} else {
-		if j.subject == "" {
-			return nil, errors.New("subject is empty")
+		if j.subject == "" && j.stream == "" {
+			return nil, errors.New("subject and stream is empty")
 		}
 	}
 

diff --git a/internal/impl/nats/input_jetstream_test.go b/internal/impl/nats/input_jetstream_test.go
@@ -110,4 +110,45 @@ bind: true
 		_, err = newJetStreamReaderFromConfig(conf, service.MockResources())
 		require.NoError(t, err)
 	})
+
+	t.Run("Stream set without subject", func(t *testing.T) {
+		inputConfig := `
+urls: [ url1 ]
+stream: foostream
+bind: false
+`
+
+		conf, err := spec.ParseYAML(inputConfig, env)
+		require.NoError(t, err)
+
+		_, err = newJetStreamReaderFromConfig(conf, service.MockResources())
+		require.NoError(t, err)
+	})
+
+	t.Run("Subject set without stream", func(t *testing.T) {
+		inputConfig := `
+urls: [ url1 ]
+subject: testsubject
+bind: false
+`
+
+		conf, err := spec.ParseYAML(inputConfig, env)
+		require.NoError(t, err)
+
+		_, err = newJetStreamReaderFromConfig(conf, service.MockResources())
+		require.NoError(t, err)
+	})
+
+	t.Run("Stream and subject empty", func(t *testing.T) {
+		inputConfig := `
+urls: [ url1 ]
+bind: false
+`
+
+		conf, err := spec.ParseYAML(inputConfig, env)
+		require.NoError(t, err)
+
+		_, err = newJetStreamReaderFromConfig(conf, service.MockResources())
+		require.Error(t, err)
+	})
 }