-
Notifications
You must be signed in to change notification settings - Fork 0
/
nats_conn.go
110 lines (104 loc) · 2.9 KB
/
nats_conn.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
//go:build !cache_nop
package main
import (
"context"
"errors"
"os"
"time"
"github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
)
// SetupNatsConnection connects the service to NATS.
// Depending on the config an embedded NATS server is started.
func SetupNatsConnection(conf TesConfig) (*nats.Conn, jetstream.JetStream) {
var js jetstream.JetStream
var nc *nats.Conn
var err error
var attempts int = 0
if conf.NatsUrl != "" {
logger.Info("Try connecting to NATS", "url", conf.NatsUrl, "timeoutSecs", conf.NatsTimeout.Seconds(), "count", attempts)
for nc == nil {
attempts++
nc, err = nats.Connect(conf.NatsUrl, nats.Name("TES"), nats.Timeout(conf.NatsTimeout))
if err != nil {
logger.Error("Connecting to NATS failed",
"url", conf.NatsUrl,
"timeoutSecs", conf.NatsTimeout.Seconds(),
"err", err,
"count", attempts,
"maxRetries", conf.NatsConnectRetries)
if attempts > conf.NatsConnectRetries {
logger.Error("Connecting to NATS failed. Retry count exceeded", "err", err, "maxRetries", conf.NatsConnectRetries)
if conf.FailWithoutJetstream {
logger.Error("FATAL: terminating")
os.Exit(2)
} else {
logger.Warn("Resuming initialization without NATS. Cache disabled.")
return nil, nil
}
}
time.Sleep(time.Second)
} else {
logger.Info("NATS connected")
}
}
} else {
ns, err := server.NewServer(
&server.Options{
JetStream: true,
MaxPayload: conf.NatsMaxPayload,
TLS: false,
DontListen: !conf.ExposeNats,
Host: conf.NatsHost,
Port: conf.NatsPort,
StoreDir: conf.NatsStoreDir,
})
if err != nil {
panic(err)
}
ns.ConfigureLogger()
ns.Start()
if !ns.ReadyForConnections(5 * time.Second) {
panic("NATS not ready!")
}
nc, err = nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
if err != nil {
panic(err)
}
}
js, err = jetstream.New(nc)
if err != nil {
logger.Error("FATAL: Error when initializing NATS JetStream", "err", err.Error())
os.Exit(1)
}
// test if JetStream is available
// we re-use the retry attempt counter from above
for {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
_, err = js.AccountInfo(ctx)
if err != nil {
if errors.Is(err, jetstream.ErrJetStreamNotEnabled) || errors.Is(err, jetstream.ErrJetStreamNotEnabledForAccount) {
logger.Error("FATAL: JetStream not enabled or not enabled for this account.")
os.Exit(2)
}
logger.Error("NATS JetStream check failed. Is JetStream enabled in external NATS server(s)?",
"err", err,
"count", attempts,
"maxRetries", conf.NatsConnectRetries)
if attempts >= conf.NatsConnectRetries {
if conf.FailWithoutJetstream {
os.Exit(2)
} else {
break
}
}
attempts++
time.Sleep(time.Second)
} else {
break
}
}
return nc, js
}