JetStream instead NATS stan

Author: ihippik

README.md

@@ -23,7 +23,7 @@ Then we filter out only the events we need and publish them in the queue
 
 ### Event publishing
 
-NATS Streaming is used as a message broker.
+NATS JetStream is used as a message broker.
 Service publishes the following structure.
 The name of the topic for subscription to receive messages is formed from the prefix of the topic, 
 the name of the database and the name of the table `prefix + schema_table`.
@@ -39,7 +39,7 @@ the name of the database and the name of the table `prefix + schema_table`.
 }
 ```
 
-Messages are published to Nats-Streaming at least once!
+Messages are published to NATS (JetStream) at least once!
 
 ### Filter configuration example
 
@@ -56,7 +56,7 @@ This filter means that we only process events occurring with the `users` table,
 and in particular `insert` and `update` data.
 
 ### Topic mapping
-By default,  output NATS topic name consist of prefix, DB schema, and DB table name, 
+By default, output NATS topic name consist of prefix, DB schema, and DB table name, 
 but if you want to send all update in one topic you should be configured the topic map:
 ```yaml
   topicsMap:
@@ -114,5 +114,5 @@ monitoring:
 You can start the container from the project folder (configuration file is required)
 
 ```
-docker run -v $(pwd)/config.yml:/app/config.yml ihippik/wal-listener:v2.0.0
+docker run -v $(pwd)/config.yml:/app/config.yml ihippik/wal-listener:v2.1.0
 ```

cmd/wal-listener/init.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/evalphobia/logrus_sentry"
 	"github.com/jackc/pgx"
+	"github.com/nats-io/nats.go"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 
@@ -82,6 +83,29 @@ func initLogger(cfg config.LoggerCfg, version string) *logrus.Entry {
 	return logger.WithField("version", version)
 }
 
+// createStream creates a stream by using JetStreamContext. We can do it manually.
+func createStream(logger *logrus.Entry, js nats.JetStreamContext, streamName string) error {
+	stream, err := js.StreamInfo(streamName)
+	if err != nil {
+		logger.WithError(err).Warnln("stream info")
+	}
+
+	if stream == nil {
+		var streamSubjects = streamName + ".*"
+
+		if _, err = js.AddStream(&nats.StreamConfig{
+			Name:     streamName,
+			Subjects: []string{streamSubjects},
+		}); err != nil {
+			return err
+		}
+
+		logger.WithField("subjects", streamSubjects).Infoln("stream not exists, created..")
+	}
+
+	return nil
+}
+
 func initSentry(dsn string, logger *logrus.Entry) {
 	if len(dsn) == 0 {
 		logger.Warnln("empty Sentry DSN")

cmd/wal-listener/main.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"os"
 
-	"github.com/nats-io/stan.go"
+	"github.com/nats-io/nats.go"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli/v2"
 
@@ -47,10 +47,20 @@ func main() {
 
 			initSentry(cfg.Monitoring.SentryDSN, logger)
 
-			natsConn, err := stan.Connect(cfg.Nats.ClusterID, cfg.Nats.ClientID, stan.NatsURL(cfg.Nats.Address))
+			natsConn, err := nats.Connect(cfg.Nats.Address)
 			if err != nil {
 				return fmt.Errorf("nats connection: %w", err)
 			}
+			defer natsConn.Close()
+
+			js, err := natsConn.JetStream()
+			if err != nil {
+				return fmt.Errorf("jet stream: %w", err)
+			}
+
+			if err := createStream(logger, js, cfg.Nats.StreamName); err != nil {
+				return fmt.Errorf("create Nats stream: %w", err)
+			}
 
 			conn, rConn, err := initPgxConnections(cfg.Database)
 			if err != nil {
@@ -62,7 +72,7 @@ func main() {
 				logger,
 				listener.NewRepository(conn),
 				rConn,
-				listener.NewNatsPublisher(natsConn),
+				listener.NewNatsPublisher(js),
 				listener.NewBinaryParser(binary.BigEndian),
 			)
 

config/config.go

@@ -28,8 +28,7 @@ type ListenerCfg struct {
 // NatsCfg path of the NATS config.
 type NatsCfg struct {
 	Address     string `valid:"required"`
-	ClusterID   string `valid:"required"`
-	ClientID    string `valid:"required"`
+	StreamName  string `valid:"required"`
 	TopicPrefix string
 }
 

config/config_test.go

@@ -36,8 +36,7 @@ func TestConfig_Validate(t *testing.T) {
 				},
 				Nats: NatsCfg{
 					Address:     "addr",
-					ClusterID:   "cluster",
-					ClientID:    "client",
+					StreamName:  "stream",
 					TopicPrefix: "prefix",
 				},
 			},
@@ -59,8 +58,7 @@ func TestConfig_Validate(t *testing.T) {
 				},
 				Nats: NatsCfg{
 					Address:     "addr",
-					ClusterID:   "cluster",
-					ClientID:    "client",
+					StreamName:  "stream",
 					TopicPrefix: "prefix",
 				},
 			},
@@ -82,8 +80,7 @@ func TestConfig_Validate(t *testing.T) {
 				},
 				Nats: NatsCfg{
 					Address:     "addr",
-					ClusterID:   "cluster",
-					ClientID:    "client",
+					StreamName:  "stream",
 					TopicPrefix: "prefix",
 				},
 			},
@@ -106,13 +103,35 @@ func TestConfig_Validate(t *testing.T) {
 					Password: "pass",
 				},
 				Nats: NatsCfg{
-					ClusterID:   "cluster",
-					ClientID:    "client",
+					StreamName:  "stream",
 					TopicPrefix: "prefix",
 				},
 			},
 			wantErr: errors.New("Nats.Address: non zero value required"),
 		},
+		{
+			name: "empty nats stream cfg",
+			fields: fields{
+				Listener: ListenerCfg{
+					SlotName:          "slot",
+					AckTimeout:        10,
+					RefreshConnection: 10,
+					HeartbeatInterval: 10,
+				},
+				Database: DatabaseCfg{
+					Host:     "host",
+					Port:     10,
+					Name:     "db",
+					User:     "usr",
+					Password: "pass",
+				},
+				Nats: NatsCfg{
+					Address:     "addr",
+					TopicPrefix: "prefix",
+				},
+			},
+			wantErr: errors.New("Nats.StreamName: non zero value required"),
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/google/uuid v1.3.0
 	github.com/jackc/pgx v3.6.2+incompatible
 	github.com/magiconair/properties v1.8.6
-	github.com/nats-io/stan.go v0.10.3
+	github.com/nats-io/nats.go v1.16.0
 	github.com/sirupsen/logrus v1.9.0
 	github.com/spf13/viper v1.12.0
 	github.com/stretchr/testify v1.8.0
@@ -25,15 +25,12 @@ require (
 	github.com/fsnotify/fsnotify v1.5.4 // indirect
 	github.com/getsentry/raven-go v0.2.0 // indirect
 	github.com/gofrs/uuid v3.2.0+incompatible // indirect
-	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/jackc/fake v0.0.0-20150926172116-812a484cc733 // indirect
 	github.com/lib/pq v1.2.0 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/nats-io/jwt v1.2.2 // indirect
 	github.com/nats-io/nats-server/v2 v2.1.2 // indirect
-	github.com/nats-io/nats-streaming-server v0.16.2 // indirect
-	github.com/nats-io/nats.go v1.16.0 // indirect
 	github.com/nats-io/nkeys v0.3.0 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect