Disable nats-server publishing

.gitignore

@@ -13,3 +13,4 @@
 
 # Dependency directories (remove the comment below to include it)
 # vendor/
+./config.yml

Dockerfile

@@ -28,6 +28,7 @@ COPY --from=builder /etc/passwd /etc/passwd
 COPY --from=builder /etc/group /etc/group
 
 COPY --from=builder /app .
+COPY /config.yml .
 
 USER goof:goof
 

cmd/wal-listener/main.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"os"
 
-	"github.com/nats-io/nats.go"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli/v2"
 
@@ -45,25 +44,6 @@ func main() {
 
 			logger := initLogger(cfg.Logger, version)
 
-			initSentry(cfg.Monitoring.SentryDSN, logger)
-
-			go initMetrics(cfg.Monitoring.PromAddr, logger)
-
-			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 {
 				return fmt.Errorf("pgx connection: %w", err)
@@ -74,7 +54,6 @@ func main() {
 				logger,
 				listener.NewRepository(conn),
 				rConn,
-				listener.NewNatsPublisher(js),
 				listener.NewBinaryParser(binary.BigEndian),
 			)
 

listener/listener.go

@@ -53,7 +53,6 @@ type Listener struct {
 	log        *logrus.Entry
 	mu         sync.RWMutex
 	slotName   string
-	publisher  publisher
 	replicator replication
 	repository repository
 	parser     parser
@@ -83,14 +82,12 @@ func NewWalListener(
 	log *logrus.Entry,
 	repo repository,
 	repl replication,
-	publ publisher,
 	parser parser,
 ) *Listener {
 	return &Listener{
 		log:        log,
 		slotName:   cfg.Listener.SlotName,
 		cfg:        cfg,
-		publisher:  publ,
 		repository: repo,
 		replicator: repl,
 		parser:     parser,
@@ -250,10 +247,10 @@ func (l *Listener) Stream(ctx context.Context) {
 					for _, event := range natsEvents {
 						subjectName := event.SubjectName(l.cfg)
 
-						if err = l.publisher.Publish(subjectName, event); err != nil {
-							l.errChannel <- fmt.Errorf("publish message: %w", err)
-							continue
-						}
+						// if err = l.publisher.Publish(subjectName, event); err != nil {
+						// 	l.errChannel <- fmt.Errorf("publish message: %w", err)
+						// 	continue
+						// }
 
 						publishedEvents.With(prometheus.Labels{"subject": subjectName, "table": event.Table}).Inc()
 

listener/nats_publisher.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/goccy/go-json"
+	// "github.com/goccy/go-json"
 	"github.com/google/uuid"
 	"github.com/nats-io/nats.go"
 
@@ -29,14 +29,14 @@ type Event struct {
 
 // Publish serializes the event and publishes it on the bus.
 func (n NatsPublisher) Publish(subject string, event Event) error {
-	msg, err := json.Marshal(event)
-	if err != nil {
-		return fmt.Errorf("marshal err: %w", err)
-	}
-
-	if _, err := n.js.Publish(subject, msg); err != nil {
-		return fmt.Errorf("failed to publish: %w", err)
-	}
+	// msg, err := json.Marshal(event)
+	// if err != nil {
+	// 	return fmt.Errorf("marshal err: %w", err)
+	// }
+
+	// if _, err := n.js.Publish(subject, msg); err != nil {
+	// 	return fmt.Errorf("failed to publish: %w", err)
+	// }
 
 	return nil
 }

listener/wal_transaction.go

@@ -1,14 +1,14 @@
 package listener
 
 import (
-	"strconv"
-	"strings"
-	"time"
-
+	"fmt"
 	"github.com/goccy/go-json"
 	"github.com/google/uuid"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
+	"strconv"
+	"strings"
+	"time"
 )
 
 // ActionKind kind of action on WAL message.
@@ -208,6 +208,8 @@ func (w *WalTransaction) CreateEventsWithFilter(tableMap map[string][]string) []
 		}
 
 		filterSkippedEvents.With(prometheus.Labels{"table": item.Table}).Inc()
+		bs, _ := json.Marshal(event.Data)
+		fmt.Println(string(bs))
 
 		logrus.WithFields(
 			logrus.Fields{