Added BufferSize to Jetsream and NATS outputs

  * follow-up to  Parallel processing with Jetstream #342
  * added option to provide a buffer size to go channel on output
  * updaed user guides for Jetstream and NATS outputs

Author: wendall-robinson

docs/user_guide/outputs/jetstream_output.md

@@ -123,6 +123,10 @@ outputs:
     # boolean, enables extra logging for the nats output
     debug: false
     # boolean, enables the collection and export (via prometheus) of output specific metrics
+    # 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 100 messages
+    buffer-size: 100
     enable-metrics: false 
     # list of processors to apply to the message before writing
     event-processors: 

docs/user_guide/outputs/nats_output.md

@@ -73,6 +73,10 @@ outputs:
     # boolean, enables extra logging for the nats output
     debug: false
     # boolean, enables the collection and export (via prometheus) of output specific metrics
+    # 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 100 messages
+    buffer-size: 100
     enable-metrics: false 
     # list of processors to apply on the message before writing
     event-processors: 

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         int                 `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, uint(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         int              `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, uint(n.Cfg.BufferSize))
 	initMetrics()
 	n.mo = &formatters.MarshalOptions{
 		Format:     n.Cfg.Format,