Always retry connections to NATS

filter/filter.go

@@ -101,7 +101,7 @@ func StartFilter(conf *config.Config) (_ *Filter, err error) {
 }
 
 func (f *Filter) natsConnect() (natsConn, error) {
-	nc, err := nats.Connect(f.c.NATSAddress)
+	nc, err := nats.Connect(f.c.NATSAddress, nats.MaxReconnects(-1))
 	if err != nil {
 		return nil, fmt.Errorf("NATS: failed to connect: %v", err)
 	}

listener/listener.go

@@ -139,12 +139,10 @@ func newListener(c *config.Config) (*Listener, error) {
 		batchSizeThreshold: c.ListenerBatchBytes - udpMaxDatagramSize,
 	}
 
-	nc, err := nats.Connect(l.c.NATSAddress)
+	nc, err := nats.Connect(l.c.NATSAddress, nats.MaxReconnects(-1))
 	if err != nil {
 		return nil, err
 	}
-	// If we disconnect, we want to try reconnecting as many times as we can.
-	nc.Opts.MaxReconnect = -1
 	l.nc = nc
 
 	return l, nil

monitor/monitor.go

@@ -99,7 +99,7 @@ func (m *Monitor) Stop() {
 }
 
 func (m *Monitor) natsConnect() (*nats.Conn, error) {
-	nc, err := nats.Connect(m.c.NATSAddress)
+	nc, err := nats.Connect(m.c.NATSAddress, nats.MaxReconnects(-1))
 	if err != nil {
 		return nil, fmt.Errorf("NATS: failed to connect: %v", err)
 	}

writer/writer.go

@@ -80,15 +80,11 @@ func StartWriter(c *config.Config) (_ *Writer, err error) {
 		return nil, err
 	}
 
-	w.nc, err = nats.Connect(c.NATSAddress)
+	w.nc, err = nats.Connect(c.NATSAddress, nats.MaxReconnects(-1))
 	if err != nil {
 		return nil, fmt.Errorf("NATS Error: can't connect: %v", err)
 	}
 
-	// if we disconnect, we want to try reconnecting as many times as
-	// we can
-	w.nc.Opts.MaxReconnect = -1
-
 	http.DefaultTransport.(*http.Transport).MaxIdleConnsPerHost = 100
 
 	jobs := make(chan *nats.Msg, 1024)