Wait for nats server master edition (#2573)

cs3org · Mar 7, 2022 · ee9653f · ee9653f
1 parent cb124db
commit ee9653f
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 1 deletion.
diff --git a/changelog/unreleased/waitForNats.md b/changelog/unreleased/waitForNats.md
@@ -0,0 +1,5 @@
+Bugfix: wait for nats server on middleware start
+
+Use a retry mechanism to connect to the nats server when it is not ready yet
+
+https://github.com/cs3org/reva/pull/2572
diff --git a/go.mod b/go.mod
@@ -14,6 +14,7 @@ require (
 	github.com/beevik/etree v1.1.0
 	github.com/bluele/gcache v0.0.2
 	github.com/c-bata/go-prompt v0.2.5
+	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/ceph/go-ceph v0.14.0
 	github.com/cheggaaa/pb v1.0.29
 	github.com/coreos/go-oidc v2.2.1+incompatible

diff --git a/go.sum b/go.sum
@@ -163,6 +163,8 @@ github.com/bmizerany/pat v0.0.0-20170815010413-6226ea591a40/go.mod h1:8rLXio+Wji
 github.com/boombuler/barcode v1.0.1-0.20190219062509-6c824513bacc/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
 github.com/c-bata/go-prompt v0.2.5 h1:3zg6PecEywxNn0xiqcXHD96fkbxghD+gdB2tbsYfl+Y=
 github.com/c-bata/go-prompt v0.2.5/go.mod h1:vFnjEGDIIA/Lib7giyE4E9c50Lvl8j0S+7FVlAwDAVw=
+github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=
+github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
 github.com/cenkalti/backoff/v4 v4.1.0/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/cenkalti/backoff/v4 v4.1.1/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/census-instrumentation/opencensus-proto v0.2.0/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=

diff --git a/pkg/events/server/nats.go b/pkg/events/server/nats.go
@@ -19,7 +19,10 @@
 package server
 
 import (
+	"fmt"
+
 	"github.com/asim/go-micro/plugins/events/nats/v4"
+	"github.com/cenkalti/backoff"
 	"go-micro.dev/v4/events"
 
 	stanServer "github.com/nats-io/nats-streaming-server/server"
@@ -38,6 +41,20 @@ func RunNatsServer(opts ...Option) error {
 }
 
 // NewNatsStream returns a streaming client used by `Consume` and `Publish` methods
+// retries exponentially to connect to a nats server
 func NewNatsStream(opts ...nats.Option) (events.Stream, error) {
-	return nats.NewStream(opts...)
+	b := backoff.NewExponentialBackOff()
+	var stream events.Stream
+	o := func() error {
+		s, err := nats.NewStream(opts...)
+		if err != nil {
+			// TODO: should we get the standard logger here? if yes: How?
+			fmt.Printf("can't connect to nats (stan) server, retrying in %s\n", b.NextBackOff())
+		}
+		stream = s
+		return err
+	}
+
+	err := backoff.Retry(o, b)
+	return stream, err
 }