Merge pull request #549 from wendall-robinson/Add_BufferSize_to_JetStream_Output

Added BufferSize to Jetsream and NATS outputs

Author: karimra

docs/user_guide/outputs/jetstream_output.md

@@ -122,6 +122,10 @@ outputs:
     write-timeout: 5s 
     # boolean, enables extra logging for the nats output
     debug: false
+    # integer, sets the size of the local buffer where received
+    # NATS messages are stored before being sent to outputs.
+    # This value is set per worker. Defaults to 0 messages
+    buffer-size: 0
     # boolean, enables the collection and export (via prometheus) of output specific metrics
     enable-metrics: false 
     # list of processors to apply to the message before writing

docs/user_guide/outputs/nats_output.md

@@ -72,6 +72,10 @@ outputs:
     write-timeout: 5s 
     # boolean, enables extra logging for the nats output
     debug: false
+    # integer, sets the size of the local buffer where received
+    # NATS messages are stored before being sent to outputs.
+    # This value is set per worker. Defaults to 0 messages
+    buffer-size: 0
     # boolean, enables the collection and export (via prometheus) of output specific metrics
     enable-metrics: false 
     # list of processors to apply on the message before writing

pkg/outputs/nats_outputs/jetstream/jetstream_output.go

@@ -87,6 +87,7 @@ type config struct {
 	NumWorkers         int                 `mapstructure:"num-workers,omitempty" json:"num-workers,omitempty"`
 	WriteTimeout       time.Duration       `mapstructure:"write-timeout,omitempty" json:"write-timeout,omitempty"`
 	Debug              bool                `mapstructure:"debug,omitempty" json:"debug,omitempty"`
+	BufferSize         uint                `mapstructure:"buffer-size,omitempty"`
 	EnableMetrics      bool                `mapstructure:"enable-metrics,omitempty" json:"enable-metrics,omitempty"`
 	EventProcessors    []string            `mapstructure:"event-processors,omitempty" json:"event-processors,omitempty"`
 }
@@ -136,7 +137,7 @@ func (n *jetstreamOutput) Init(ctx context.Context, name string, cfg map[string]
 		return err
 	}
 
-	n.msgChan = make(chan *outputs.ProtoMsg)
+	n.msgChan = make(chan *outputs.ProtoMsg, n.Cfg.BufferSize)
 	initMetrics()
 	n.mo = &formatters.MarshalOptions{
 		Format:     n.Cfg.Format,

pkg/outputs/nats_outputs/nats/nats_output.go

@@ -87,6 +87,7 @@ type Config struct {
 	NumWorkers         int              `mapstructure:"num-workers,omitempty"`
 	WriteTimeout       time.Duration    `mapstructure:"write-timeout,omitempty"`
 	Debug              bool             `mapstructure:"debug,omitempty"`
+	BufferSize         uint             `mapstructure:"buffer-size,omitempty"`
 	EnableMetrics      bool             `mapstructure:"enable-metrics,omitempty"`
 	EventProcessors    []string         `mapstructure:"event-processors,omitempty"`
 }
@@ -145,7 +146,7 @@ func (n *NatsOutput) Init(ctx context.Context, name string, cfg map[string]inter
 		return err
 	}
 
-	n.msgChan = make(chan *outputs.ProtoMsg)
+	n.msgChan = make(chan *outputs.ProtoMsg, n.Cfg.BufferSize)
 	initMetrics()
 	n.mo = &formatters.MarshalOptions{
 		Format:     n.Cfg.Format,