Redis streams input allows for automatic creation of streams (redpand…

…a-data#855)

* Redis streams allows for automatic creation of stream (i.e. Redis MKSTREAM option).

Also fixes the error messages where group and stream names were flipped.

* redis_streams auto-create streams by default.

config/redis_streams.yaml

@@ -25,6 +25,7 @@ input:
  limit: 10
  client_id: benthos_consumer
  consumer_group: benthos_group
+ create_streams: true
  start_from_oldest: true
  commit_period: 1s
  timeout: 1s

lib/input/docs.go

@@ -54,6 +54,7 @@ input:
  url: tcp:https://localhost:6379
  streams:
  - benthos_stream
+ create_streams: false
  body_key: body
  consumer_group: benthos_group
 

lib/input/reader/redis_streams.go

@@ -25,6 +25,7 @@ type RedisStreamsConfig struct {
  bredis.Config `json:",inline" yaml:",inline"`
  BodyKey string `json:"body_key" yaml:"body_key"`
  Streams []string `json:"streams" yaml:"streams"`
+ CreateStreams bool `json:"create_streams" yaml:"create_streams"`
  ConsumerGroup string `json:"consumer_group" yaml:"consumer_group"`
  ClientID string `json:"client_id" yaml:"client_id"`
  Limit int64 `json:"limit" yaml:"limit"`
@@ -42,6 +43,7 @@ func NewRedisStreamsConfig() RedisStreamsConfig {
  Config: bredis.NewConfig(),
  BodyKey: "body",
  Streams: []string{"benthos_stream"},
+ CreateStreams: true,
  ConsumerGroup: "benthos_group",
  ClientID: "benthos_consumer",
  Limit: 10,
@@ -219,10 +221,14 @@ func (r *RedisStreams) ConnectWithContext(ctx context.Context) error {
  if r.conf.StartFromOldest {
  offset = "0"
  }
- if err := client.XGroupCreate(s, r.conf.ConsumerGroup, offset).Err(); err != nil {
- if err.Error() != "BUSYGROUP Consumer Group name already exists" {
- return fmt.Errorf("failed to create group %v for stream %v: %v", s, r.conf.ConsumerGroup, err)
- }
+ var err error
+ if r.conf.CreateStreams {
+ err = client.XGroupCreateMkStream(s, r.conf.ConsumerGroup, offset).Err()
+ } else {
+ err = client.XGroupCreate(s, r.conf.ConsumerGroup, offset).Err()
+ }
+ if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
+ return fmt.Errorf("failed to create group %v for stream %v: %v", r.conf.ConsumerGroup, s, err)
  }
  }
 

lib/input/redis_streams.go

@@ -33,6 +33,7 @@ as metadata fields.`,
  docs.FieldCommon("limit", "The maximum number of messages to consume from a single request."),
  docs.FieldCommon("client_id", "An identifier for the client connection."),
  docs.FieldCommon("consumer_group", "An identifier for the consumer group of the stream."),
+ docs.FieldAdvanced("create_streams", "Create subscribed streams if they do not exist (MKSTREAM option)."),
  docs.FieldAdvanced("start_from_oldest", "If an offset is not found for a stream, determines whether to consume from the oldest available offset, otherwise messages are consumed from the latest offset."),
  docs.FieldAdvanced("commit_period", "The period of time between each commit of the current offset. Offsets are always committed during shutdown."),
  docs.FieldAdvanced("timeout", "The length of time to poll for new messages before reattempting."),

website/docs/components/inputs/redis_streams.md

@@ -65,6 +65,7 @@ input:
  limit: 10
  client_id: benthos_consumer
  consumer_group: benthos_group
+ create_streams: true
  start_from_oldest: true
  commit_period: 1s
  timeout: 1s
@@ -294,6 +295,14 @@ An identifier for the consumer group of the stream.
 Type: `string` 
 Default: `"benthos_group"` 
 
+### `create_streams`
+
+Create subscribed streams if they do not exist (MKSTREAM option).
+
+
+Type: `bool` 
+Default: `true` 
+
 ### `start_from_oldest`
 
 If an offset is not found for a stream, determines whether to consume from the oldest available offset, otherwise messages are consumed from the latest offset.