Decouple: Add a output queue

Without any output queue, decouple drop a packet, before sink.http have
time to increase the number of workers, to handle more requests.

Author: cpaillet

README.md

@@ -204,7 +204,9 @@ In the following example we decouple the sink from the source to avoid slowing d
   },
   {
     "type": "control.decouple",
-    "config": {}
+    "config": {
+      "queue_size": 500
+    }
   },
   {
     "type": "sink.http",
@@ -218,7 +220,8 @@ In the following example we decouple the sink from the source to avoid slowing d
 
 | Param   | Value                                                 |
 | ------- | ----------------------------------------------------- |
-| `quiet` | Do not log dropped messages summary. Default: `false` |
+| `quiet`      | Do not log dropped messages summary. Default: `false` |
+| `queue_size` | Size of the output queue. Default: `100` |
 
 #### control.rate_limit
 

mirror/modules/control/decouple.go

@@ -30,6 +30,7 @@ func init() {
 
 type DecoupleConfig struct {
 	Quiet bool `json:"log"`
+	QueueSize int `json:"queue_size"`
 }
 
 type Decouple struct {
@@ -45,9 +46,14 @@ func NewDecouple(ctx *mirror.ModuleContext, cfg []byte) (mirror.Module, error) {
 		return nil, err
 	}
 
+	queueSize := 100
+	if c.QueueSize > 0 {
+		queueSize = c.QueueSize
+	}
+
 	mod := &Decouple{
 		ctx:   ctx,
-		out:   make(chan mirror.Request),
+		out:   make(chan mirror.Request, queueSize),
 		quiet: c.Quiet,
 	}