Merge pull request #95 from nats-io/support-bind-stream

Support bind stream JS option

Author: wallyqs

docker-bake.hcl

@@ -16,7 +16,7 @@ variable CI {
 }
 
 variable image_base {
-  default = "docker-image://alpine:3.17.3"
+  default = "docker-image://alpine:3.18.4"
 }
 
 variable image_goreleaser {

server/conf/conf.go

@@ -209,6 +209,7 @@ type ConnectorConfig struct {
 
 	Subject   string // Used for nats and jetstream connections
 	QueueName string // Optional, used for nats connections
+	Stream    string // Uses BindStream option for JetStream to consume from sourced streams
 
 	Brokers []string // list of brokers to use for creating a reader/writer
 	Topic   string   // kafka topic

server/core/connector.go

@@ -382,6 +382,9 @@ func (conn *BridgeConnector) subscribeToJetStream(subject string) (*nats.Subscri
 	if d := conn.bridge.config.JetStream.HeartbeatInterval; d > 0 {
 		options = append(options, nats.IdleHeartbeat(time.Duration(d)*time.Millisecond))
 	}
+	if len(conn.config.Stream) > 0 {
+		options = append(options, nats.BindStream(conn.config.Stream))
+	}
 
 	traceEnabled := conn.bridge.Logger().TraceEnabled()
 	ackSyncEnabled := conn.bridge.config.JetStream.EnableAckSync

server/core/jetstream2kafka.go

@@ -81,7 +81,7 @@ func (conn *JetStream2KafkaConnector) Shutdown() error {
 // CheckConnections ensures the nats/stan connection and report an error if it is down
 func (conn *JetStream2KafkaConnector) CheckConnections() error {
 	if !conn.bridge.CheckJetStream() {
-		return fmt.Errorf("%s connector requires nats streaming to be available", conn.String())
+		return fmt.Errorf("%s connector requires nats jetstream to be available", conn.String())
 	}
 	return nil
 }

server/core/jetstream2kafka_test.go

@@ -906,3 +906,55 @@ func TestJetStreamAlreadyConnected(t *testing.T) {
 
 	require.NoError(t, tbs.Bridge.connectToJetStream())
 }
+
+func TestJetStreamSourcesConsumedByKafka(t *testing.T) {
+	topic := nuid.Next()
+
+	connect := []conf.ConnectorConfig{
+		{
+			Type:        "JetStreamToKafka",
+			Subject:     "foo.*",
+			Topic:       topic,
+			DurableName: "KafkaBridgeConsumer",
+			Stream:      "FOO_GLOBAL",
+		},
+	}
+
+	tbs, err := StartTestEnvironmentWithSources(connect)
+	require.NoError(t, err)
+	defer tbs.Close()
+
+	tbs.Bridge.checkConnections()
+
+	_, err = tbs.JS.Publish("foo.one", []byte("one"))
+	require.NoError(t, err)
+	_, err = tbs.JS.Publish("foo.two", []byte("two"))
+	require.NoError(t, err)
+	_, err = tbs.JS.Publish("foo.one.1", []byte("another one"))
+	require.NoError(t, err)
+	_, err = tbs.JS.Publish("foo.three", []byte("three"))
+	require.NoError(t, err)
+
+	reader := tbs.CreateReader(topic, 5000)
+	defer reader.Close()
+
+	_, data, _, err := tbs.GetMessageFromKafka(reader, 5000)
+	require.NoError(t, err)
+	require.Equal(t, "one", string(data))
+
+	_, data, _, err = tbs.GetMessageFromKafka(reader, 5000)
+	require.NoError(t, err)
+	require.Equal(t, "two", string(data))
+
+	_, data, _, err = tbs.GetMessageFromKafka(reader, 5000)
+	require.NoError(t, err)
+	require.Equal(t, "three", string(data))
+
+	stats := tbs.Bridge.SafeStats()
+	connStats := stats.Connections[0]
+	require.Equal(t, int64(3), connStats.MessagesIn)
+	require.Equal(t, int64(3), connStats.MessagesOut)
+	require.Equal(t, int64(1), connStats.Connects)
+	require.Equal(t, int64(0), connStats.Disconnects)
+	require.True(t, connStats.Connected)
+}

server/core/kafka2jetstream.go

@@ -103,7 +103,7 @@ func (conn *Kafka2JetStreamConnector) Shutdown() error {
 // CheckConnections ensures the nats/stan connection and report an error if it is down
 func (conn *Kafka2JetStreamConnector) CheckConnections() error {
 	if !conn.bridge.CheckJetStream() {
-		return fmt.Errorf("%s connector requires nats streaming to be available", conn.String())
+		return fmt.Errorf("%s connector requires nats jetstream to be available", conn.String())
 	}
 	return nil
 }

server/core/server_test.go

@@ -172,6 +172,60 @@ func StartSASLTestEnvironment(connections []conf.ConnectorConfig) (*TestEnv, err
 	return tbs, err
 }
 
+func StartTestEnvironmentWithSources(connections []conf.ConnectorConfig) (*TestEnv, error) {
+	tbs, err := StartTestEnvironmentInfrastructure(false, false, collectTopics(connections))
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = tbs.JS.AddStream(&nats.StreamConfig{
+		Name:     "FOO_1",
+		Subjects: []string{"foo.one"},
+	})
+	if err != nil {
+		return nil, err
+	}
+	_, err = tbs.JS.AddStream(&nats.StreamConfig{
+		Name:     "FOO_2",
+		Subjects: []string{"foo.two"},
+	})
+	if err != nil {
+		return nil, err
+	}
+	_, err = tbs.JS.AddStream(&nats.StreamConfig{
+		Name:     "FOO_3",
+		Subjects: []string{"foo.three"},
+	})
+	if err != nil {
+		return nil, err
+	}
+	_, err = tbs.JS.AddStream(&nats.StreamConfig{
+		Name:     "SUB_FOO_1",
+		Subjects: []string{"foo.one.1"},
+	})
+	if err != nil {
+		return nil, err
+	}
+	_, err = tbs.JS.AddStream(&nats.StreamConfig{
+		Name: "FOO_GLOBAL",
+		Sources: []*nats.StreamSource{
+			{Name: "FOO_1"},
+			{Name: "FOO_2"},
+			{Name: "FOO_3"},
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	err = tbs.StartBridge(connections)
+	if err != nil {
+		tbs.Close()
+		return nil, err
+	}
+	return tbs, err
+}
+
 // StartTestEnvironmentInfrastructure creates the kafka server, Nats and streaming
 // but does not start a bridge, you can use StartBridge to start a bridge afterward
 func StartTestEnvironmentInfrastructure(useSASL, useTLS bool, topics []string) (*TestEnv, error) {