Bugfix/gh 220 slow consumer error not handled

README.md

@@ -156,6 +156,9 @@ Configuration is a JSON encoded file. If no config file is found at the given pa
     // Timeout in milliseconds for NATS requests.
     "requestTimeout": 3000,
 
+    // Size of message buffer for incoming NATS requests.
+    "bufferSize": 8192,
+
     // Header authentication resource method for web resources.
     // Prior to accessing the resource, this resource method will be
     // called, allowing an auth service to set a token using

main.go

@@ -78,6 +78,7 @@ type Config struct {
 	NatsTLSKey     string   `json:"natsKey"`
 	NatsRootCAs    []string `json:"natsRootCAs"`
 	RequestTimeout int      `json:"requestTimeout"`
+	BufferSize     int      `json:"bufferSize"`
 	Debug          bool     `json:"debug"`
 	Trace          bool     `json:"trace"`
 	server.Config
@@ -110,6 +111,9 @@ func (c *Config) SetDefault() {
 	if c.NatsRootCAs == nil {
 		c.NatsRootCAs = []string{}
 	}
+	if c.BufferSize == 0 {
+		c.BufferSize = 8192
+	}
 	c.Config.SetDefault()
 }
 
@@ -303,6 +307,7 @@ func main() {
 		ClientKey:      cfg.NatsTLSKey,
 		RootCAs:        cfg.NatsRootCAs,
 		RequestTimeout: time.Duration(cfg.RequestTimeout) * time.Millisecond,
+		BufferSize:     cfg.BufferSize,
 		Logger:         l,
 	}, cfg.Config)
 	if err != nil {

nats/nats.go

@@ -13,10 +13,6 @@ import (
 	"github.com/resgateio/resgate/server/mq"
 )
 
-const (
-	natsChannelSize = 256
-)
-
 // Client holds a client connection to a nats server.
 type Client struct {
 	RequestTimeout time.Duration
@@ -26,6 +22,7 @@ type Client struct {
 	ClientKey      string
 	RootCAs        []string
 	Logger         logger.Logger
+	BufferSize     int
 
 	mq           *nats.Conn
 	mqCh         chan *nats.Msg
@@ -75,7 +72,11 @@ func (c *Client) Connect() error {
 	c.Logf("Connecting to NATS at %s", c.URL)
 
 	// Create connection options
-	opts := []nats.Option{nats.NoReconnect(), nats.ClosedHandler(c.onClose)}
+	opts := []nats.Option{
+		nats.NoReconnect(),
+		nats.ClosedHandler(c.onClose),
+		nats.ErrorHandler(c.onError),
+	}
 	if c.Creds != "" {
 		opts = append(opts, nats.UserCredentials(c.Creds))
 	}
@@ -98,7 +99,7 @@ func (c *Client) Connect() error {
 	}
 
 	c.mq = nc
-	c.mqCh = make(chan *nats.Msg, natsChannelSize)
+	c.mqCh = make(chan *nats.Msg, c.BufferSize)
 	c.mqReqs = make(map[*nats.Subscription]*responseCont)
 	c.tq = timerqueue.New(c.onTimeout, c.RequestTimeout)
 	c.stopped = make(chan struct{})
@@ -173,6 +174,13 @@ func (c *Client) onClose(conn *nats.Conn) {
 	}
 }
 
+func (c *Client) onError(conn *nats.Conn, sub *nats.Subscription, err error) {
+	c.Logger.Error(err.Error())
+	if err == nats.ErrSlowConsumer {
+		c.Close()
+	}
+}
+
 // SendRequest sends a request to the MQ.
 func (c *Client) SendRequest(subj string, payload []byte, cb mq.Response) {
 	inbox := nats.NewInbox()