this repo has no description
wip: setup nats jetstream
+95
-3
+1
Justfile
+1
Justfile
+6
docker-compose.yaml
+6
docker-compose.yaml
+6
-1
go.mod
+6
-1
go.mod
···
6
6
github.com/caarlos0/env/v11 v11.2.2
7
7
github.com/ewen-lbh/label-logger-go v0.1.1
8
8
github.com/joho/godotenv v1.5.1
9
+
github.com/nats-io/nats.go v1.37.0
9
10
github.com/orcaman/concurrent-map/v2 v2.0.1
10
11
github.com/shopspring/decimal v1.4.0
11
12
github.com/steebchen/prisma-client-go v0.42.0
···
13
14
14
15
require (
15
16
github.com/golang-jwt/jwt v3.2.2+incompatible // indirect
16
-
golang.org/x/crypto v0.9.0 // indirect
17
+
github.com/klauspost/compress v1.17.2 // indirect
18
+
github.com/nats-io/nkeys v0.4.7 // indirect
19
+
github.com/nats-io/nuid v1.0.1 // indirect
20
+
golang.org/x/crypto v0.18.0 // indirect
21
+
golang.org/x/text v0.14.0 // indirect
17
22
)
18
23
19
24
require (
+12
-1
go.sum
+12
-1
go.sum
···
16
16
github.com/gosuri/uiprogress v0.0.1/go.mod h1:C1RTYn4Sc7iEyf6j8ft5dyoZ4212h8G1ol9QQluh5+0=
17
17
github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
18
18
github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
19
+
github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
20
+
github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
19
21
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
20
22
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
21
23
github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db h1:62I3jR2EmQ4l5rM/4FEfDWcRD+abF5XlKShorW5LRoQ=
22
24
github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db/go.mod h1:l0dey0ia/Uv7NcFFVbCLtqEBQbrT4OCwCSKTEv6enCw=
25
+
github.com/nats-io/nats.go v1.37.0 h1:07rauXbVnnJvv1gfIyghFEo6lUcYRY0WXc3x7x0vUxE=
26
+
github.com/nats-io/nats.go v1.37.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
27
+
github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
28
+
github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
29
+
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
30
+
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
23
31
github.com/orcaman/concurrent-map/v2 v2.0.1 h1:jOJ5Pg2w1oeB6PeDurIYf6k9PQ+aTITr/6lP/L/zp6c=
24
32
github.com/orcaman/concurrent-map/v2 v2.0.1/go.mod h1:9Eq3TG2oBe5FirmYWQfYO5iH1q0Jv47PLaNK++uCdOM=
25
33
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
···
33
41
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
34
42
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
35
43
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
36
-
golang.org/x/crypto v0.9.0 h1:LF6fAI+IutBocDJ2OT0Q1g8plpYljMZ4+lty+dsqw3g=
37
44
golang.org/x/crypto v0.9.0/go.mod h1:yrmDGqONDYtNj3tH8X9dzUun2m2lzPa9ngI6/RUPGR0=
45
+
golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
46
+
golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
38
47
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
39
48
golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
40
49
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
···
64
73
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
65
74
golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
66
75
golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
76
+
golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
77
+
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
67
78
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
68
79
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
69
80
golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+9
receiver.go
+9
receiver.go
+61
-1
server/main.go
+61
-1
server/main.go
···
3
3
package main
4
4
5
5
import (
6
+
"context"
6
7
"fmt"
7
-
"net/http"
8
8
"os"
9
+
"os/signal"
10
+
"syscall"
11
+
"time"
9
12
10
13
"git.inpt.fr/churros/notella"
11
14
"github.com/caarlos0/env/v11"
12
15
"github.com/common-nighthawk/go-figure"
13
16
ll "github.com/ewen-lbh/label-logger-go"
14
17
"github.com/joho/godotenv"
18
+
"github.com/nats-io/nats.go"
19
+
"github.com/nats-io/nats.go/jetstream"
15
20
)
16
21
17
22
type Configuration struct {
···
54
59
ll.Info("starting scheduler")
55
60
go notella.StartScheduler()
56
61
62
+
ll.Log("Connecting", "cyan", "to NATS server at [bold]%s[reset]", nats.DefaultURL)
63
+
nc, err := nats.Connect(nats.DefaultURL)
64
+
if err != nil {
65
+
ll.ErrorDisplay("could not connect to NATS at %s", err, nats.DefaultURL)
66
+
return
67
+
}
68
+
69
+
ctx, cancel := context.WithTimeout(context.Background(), 30 * time.Second)
70
+
defer cancel()
71
+
72
+
js, err := jetstream.New(nc)
73
+
if err != nil {
74
+
ll.ErrorDisplay("could not connect to Jetstream", err)
75
+
return
76
+
}
77
+
78
+
ll.Log("Initializing", "cyan", "a Jetstream stream [bold]notella:stream[reset], listening for [bold]notella:*[reset] subjects")
79
+
80
+
s, err := js.CreateStream(ctx, jetstream.StreamConfig{
81
+
Name: "notella:stream",
82
+
Subjects: []string{"notella:*"},
83
+
})
84
+
if err != nil {
85
+
ll.ErrorDisplay("could not create stream", err)
86
+
return
87
+
}
88
+
89
+
ll.Log("Initializing", "cyan", "Jetstream consumer [bold]NotellaConsumer[reset] with [bold]AckExplicitPolicy[reset]")
90
+
91
+
cons, err := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
92
+
Durable: "NotellaConsumer",
93
+
AckPolicy: jetstream.AckExplicitPolicy,
94
+
})
95
+
if err != nil {
96
+
ll.ErrorDisplay("could not create consumer", err)
97
+
return
98
+
}
99
+
100
+
ll.Log("Starting", "cyan", "consumer [bold]NotellaConsumer[reset]")
101
+
102
+
cc, err := cons.Consume(func(msg jetstream.Msg) {
103
+
fmt.Println(string(msg.Data()))
104
+
msg.Ack()
105
+
}, jetstream.ConsumeErrHandler(func(consumeCtx jetstream.ConsumeContext, err error) {
106
+
fmt.Println(err)
107
+
}))
108
+
if err != nil {
109
+
ll.ErrorDisplay("could not start consumer", err)
110
+
return
111
+
}
112
+
defer cc.Stop()
113
+
114
+
sig := make(chan os.Signal, 1)
115
+
signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
116
+
<-sig
57
117
}