From 93a61d8ef3364ce442ae2e298392e1127096a9aa Mon Sep 17 00:00:00 2001 From: postmannen Date: Tue, 21 Jun 2022 10:40:32 +0200 Subject: [PATCH] tested flushing nats.Conn --- process.go | 10 +++++++++- server.go | 1 + 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/process.go b/process.go index ba9c9ef..af000c7 100644 --- a/process.go +++ b/process.go @@ -272,8 +272,16 @@ func (p process) messageDeliverNats(natsMsgPayload []byte, natsMsgHeader nats.He } p.metrics.promNatsDeliveredTotal.Inc() - // The reaming logic is for handling ACK messages, so we return here + //err = natsConn.Flush() + //if err != nil { + // er := fmt.Errorf("error: nats publish flush failed: %v", err) + // log.Printf("%v\n", er) + // return + //} + + // The remaining logic is for handling ACK messages, so we return here // since it was a NACK message, and all or now done. + return } diff --git a/server.go b/server.go index 089e757..907225e 100644 --- a/server.go +++ b/server.go @@ -107,6 +107,7 @@ func NewServer(configuration *Configuration, version string) (*server, error) { // Setting MaxReconnects to -1 which equals unlimited. conn, err = nats.Connect(configuration.BrokerAddress, opt, + //nats.FlusherTimeout(time.Second*10), nats.MaxReconnects(-1), nats.ReconnectJitter(time.Duration(configuration.NatsReconnectJitter)*time.Millisecond, time.Duration(configuration.NatsReconnectJitterTLS)*time.Second), nats.Timeout(time.Second*time.Duration(configuration.NatsConnOptTimeout)),