Repository: redpanda-data/connect (language: go)

Issue:
Add Interpolation/dynamic routing support for redis streams
For Kafka, benthos supports dynamic routing, e.g:
```
output:
  kafka:
    addresses: [ TODO ]
    topic: ${! meta("output_topic") }
    key: ${! meta("key") }
```

It would be great if similar support could be added for redis streams, if possible.


Discussion / hints:
Hey @agrawalravi90, this is definitely worthwhile, I'm surprised we don't already support it.
Great, thanks. Just FYI, currently it seems to treat the value as a string literal and I get stream names like so:
127.0.0.1:6379> SCAN 0 TYPE stream
1) "15"
2) 1) "${! meta(\"topic\") }"
   2) "${! json(\"class\") }"

Fix patch:
diff --git a/internal/impl/redis/output_streams.go b/internal/impl/redis/output_streams.go
index 0e30995f12..ebfc0aec69 100644
--- a/internal/impl/redis/output_streams.go
+++ b/internal/impl/redis/output_streams.go
@@ -10,6 +10,7 @@ import (
 
 	ibatch "github.com/benthosdev/benthos/v4/internal/batch"
 	"github.com/benthosdev/benthos/v4/internal/batch/policy"
+	"github.com/benthosdev/benthos/v4/internal/bloblang/field"
 	"github.com/benthosdev/benthos/v4/internal/bundle"
 	"github.com/benthosdev/benthos/v4/internal/component"
 	"github.com/benthosdev/benthos/v4/internal/component/output"
@@ -38,7 +39,7 @@ key to be set to the body of the message. All metadata fields of the message
 will also be set as key/value pairs, if there is a key collision between
 a metadata item and the body then the body takes precedence.`),
 		Config: docs.FieldComponent().WithChildren(old.ConfigDocs()...).WithChildren(
-			docs.FieldString("stream", "The stream to add messages to."),
+			docs.FieldString("stream", "The stream to add messages to.").IsInterpolated(),
 			docs.FieldString("body_key", "A key to set the raw body of the message to."),
 			docs.FieldInt("max_length", "When greater than zero enforces a rough cap on the length of the target stream."),
 			docs.FieldInt("max_in_flight", "The maximum number of messages to have in flight at a given time. Increase this to improve throughput."),
@@ -55,7 +56,7 @@ a metadata item and the body then the body takes precedence.`),
 }
 
 func newRedisStreamsOutput(conf output.Config, mgr bundle.NewManagement) (output.Streamed, error) {
-	w, err := newRedisStreamsWriter(conf.RedisStreams, mgr.Logger())
+	w, err := newRedisStreamsWriter(conf.RedisStreams, mgr)
 	if err != nil {
 		return nil, err
 	}
@@ -70,20 +71,23 @@ type redisStreamsWriter struct {
 	log log.Modular
 
 	conf       output.RedisStreamsConfig
+	stream     *field.Expression
 	metaFilter *metadata.ExcludeFilter
 
 	client  redis.UniversalClient
 	connMut sync.RWMutex
 }
 
-func newRedisStreamsWriter(conf output.RedisStreamsConfig, log log.Modular) (*redisStreamsWriter, error) {
-
+func newRedisStreamsWriter(conf output.RedisStreamsConfig, mgr bundle.NewManagement) (*redisStreamsWriter, error) {
 	r := &redisStreamsWriter{
-		log:  log,
+    log:  mgr.Logger(),
 		conf: conf,
 	}
 
 	var err error
+	if r.stream, err = mgr.BloblEnvironment().NewField(conf.Stream); err != nil {
+		return nil, fmt.Errorf("failed to parse expression: %v", err)
+	}
 	if r.metaFilter, err = conf.Metadata.Filter(); err != nil {
 		return nil, fmt.Errorf("failed to construct metadata filter: %w", err)
 	}
@@ -134,7 +138,7 @@ func (r *redisStreamsWriter) WriteWithContext(ctx context.Context, msg *message.
 	if msg.Len() == 1 {
 		if err := client.XAdd(&redis.XAddArgs{
 			ID:           "*",
-			Stream:       r.conf.Stream,
+			Stream:       r.stream.String(0, msg),
 			MaxLenApprox: r.conf.MaxLenApprox,
 			Values:       partToMap(msg.Get(0)),
 		}).Err(); err != nil {
@@ -149,7 +153,7 @@ func (r *redisStreamsWriter) WriteWithContext(ctx context.Context, msg *message.
 	_ = msg.Iter(func(i int, p *message.Part) error {
 		_ = pipe.XAdd(&redis.XAddArgs{
 			ID:           "*",
-			Stream:       r.conf.Stream,
+			Stream:       r.stream.String(i, msg),
 			MaxLenApprox: r.conf.MaxLenApprox,
 			Values:       partToMap(p),
 		})
diff --git a/website/docs/components/outputs/redis_streams.md b/website/docs/components/outputs/redis_streams.md
index 8062a40eef..e50fc0127a 100644
--- a/website/docs/components/outputs/redis_streams.md
+++ b/website/docs/components/outputs/redis_streams.md
@@ -280,6 +280,7 @@ Default: `""`
 ### `stream`
 
 The stream to add messages to.
+This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries).
 
 
 Type: `string`  


Test patch:
diff --git a/internal/impl/redis/integration_test.go b/internal/impl/redis/integration_test.go
index 15de39c146..514669ea1e 100644
--- a/internal/impl/redis/integration_test.go
+++ b/internal/impl/redis/integration_test.go
@@ -14,6 +14,8 @@ import (
 	"github.com/benthosdev/benthos/v4/internal/component/output"
 	"github.com/benthosdev/benthos/v4/internal/integration"
 	"github.com/benthosdev/benthos/v4/internal/log"
+	"github.com/benthosdev/benthos/v4/internal/manager/mock"
+	_ "github.com/benthosdev/benthos/v4/public/components/pure"
 )
 
 func TestIntegrationRedis(t *testing.T) {
@@ -35,7 +37,7 @@ func TestIntegrationRedis(t *testing.T) {
 		conf := output.NewRedisStreamsConfig()
 		conf.URL = fmt.Sprintf("tcp://localhost:%v", resource.GetPort("6379/tcp"))
 
-		r, cErr := newRedisStreamsWriter(conf, log.Noop())
+		r, cErr := newRedisStreamsWriter(conf, mock.NewManager(), log.Noop())
 		if cErr != nil {
 			return cErr
 		}
@@ -52,7 +54,7 @@ func TestIntegrationRedis(t *testing.T) {
 output:
   redis_streams:
     url: tcp://localhost:$PORT
-    stream: stream-$ID
+    stream: ${! meta("routing_stream_prefix") }-stream-$ID
     body_key: body
     max_length: 0
     max_in_flight: $MAX_IN_FLIGHT
@@ -60,12 +62,14 @@ output:
       exclude_prefixes: [ $OUTPUT_META_EXCLUDE_PREFIX ]
     batching:
       count: $OUTPUT_BATCH_COUNT
+  processors:
+    - bloblang: meta routing_stream_prefix = "bar"
 
 input:
   redis_streams:
     url: tcp://localhost:$PORT
     body_key: body
-    streams: [ stream-$ID ]
+    streams: [ bar-stream-$ID ]
     limit: 10
     client_id: client-input-$ID
     consumer_group: group-$ID
@@ -241,7 +245,7 @@ func BenchmarkIntegrationRedis(b *testing.B) {
 		conf := output.NewRedisStreamsConfig()
 		conf.URL = fmt.Sprintf("tcp://localhost:%v", resource.GetPort("6379/tcp"))
 
-		r, cErr := newRedisStreamsWriter(conf, log.Noop())
+		r, cErr := newRedisStreamsWriter(conf, mock.NewManager(), log.Noop())
 		if cErr != nil {
 			return cErr
 		}