Some fixes

cmd/wal-listener/init.go

@@ -3,10 +3,10 @@ package main
 import (
 	"context"
 	"fmt"
+	"log/slog"
 
 	"github.com/jackc/pgx"
 	"github.com/nats-io/nats.go"
-	"log/slog"
 
 	"github.com/ihippik/wal-listener/v2/internal/config"
 	"github.com/ihippik/wal-listener/v2/internal/publisher"

internal/listener/listener.go

@@ -17,7 +17,7 @@ import (
 	"golang.org/x/sync/errgroup"
 
 	"github.com/ihippik/wal-listener/v2/internal/config"
-	"github.com/ihippik/wal-listener/v2/internal/listener/transaction"
+	tx "github.com/ihippik/wal-listener/v2/internal/listener/transaction"
 	"github.com/ihippik/wal-listener/v2/internal/publisher"
 )
 
@@ -29,7 +29,7 @@ type eventPublisher interface {
 }
 
 type parser interface {
-	ParseWalMessage([]byte, *transaction.WAL) error
+	ParseWalMessage([]byte, *tx.WAL) error
 }
 
 type replication interface {
@@ -71,7 +71,6 @@ type Listener struct {
 	isAlive    atomic.Bool
 }
 
-// Variable with connection errors.
 var (
 	errReplConnectionIsLost = errors.New("replication connection to postgres is lost")
 	errConnectionIsLost     = errors.New("db connection to postgres is lost")
@@ -319,7 +318,7 @@ func (l *Listener) Stream(ctx context.Context) error {
 		},
 	}
 
-	tx := transaction.NewWAL(l.log, pool, l.monitor)
+	txWAL := tx.NewWAL(l.log, pool, l.monitor)
 
 	for {
 		if err := ctx.Err(); err != nil {
@@ -337,29 +336,29 @@ func (l *Listener) Stream(ctx context.Context) error {
 			continue
 		}
 
-		if err = l.processMessage(ctx, msg, tx); err != nil {
+		if err = l.processMessage(ctx, msg, txWAL); err != nil {
 			return fmt.Errorf("process message: %w", err)
 		}
 
 		l.processHeartBeat(msg)
 	}
 }
 
-func (l *Listener) processMessage(ctx context.Context, msg *pgx.ReplicationMessage, tx *transaction.WAL) error {
+func (l *Listener) processMessage(ctx context.Context, msg *pgx.ReplicationMessage, txWAL *tx.WAL) error {
 	if msg.WalMessage == nil {
 		l.log.Debug("empty wal-message")
 		return nil
 	}
 
 	l.log.Debug("WAL message has been received", slog.Uint64("wal", msg.WalMessage.WalStart))
 
-	if err := l.parser.ParseWalMessage(msg.WalMessage.WalData, tx); err != nil {
+	if err := l.parser.ParseWalMessage(msg.WalMessage.WalData, txWAL); err != nil {
 		l.monitor.IncProblematicEvents(problemKindParse)
 		return fmt.Errorf("parse: %w", err)
 	}
 
-	if tx.CommitTime != nil {
-		for event := range tx.CreateEventsWithFilter(ctx, l.cfg.Listener.Filter.Tables) {
+	if txWAL.CommitTime != nil {
+		for event := range txWAL.CreateEventsWithFilter(ctx, l.cfg.Listener.Filter.Tables) {
 			subjectName := event.SubjectName(l.cfg)
 
 			if err := l.publisher.Publish(ctx, subjectName, event); err != nil {
@@ -377,10 +376,10 @@ func (l *Listener) processMessage(ctx context.Context, msg *pgx.ReplicationMessa
 				slog.Uint64("lsn", l.readLSN()),
 			)
 
-			tx.RetrieveEvent(event)
+			txWAL.RetrieveEvent(event)
 		}
 
-		tx.Clear()
+		txWAL.Clear()
 	}
 
 	if msg.WalMessage.WalStart > l.readLSN() {

internal/listener/publisher_mock_test.go

@@ -2,9 +2,10 @@ package listener
 
 import (
 	"context"
-	"github.com/ihippik/wal-listener/v2/internal/publisher"
 
 	"github.com/stretchr/testify/mock"
+
+	"github.com/ihippik/wal-listener/v2/internal/publisher"
 )
 
 type publisherMock struct {

internal/listener/transaction/wal.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+
 	"github.com/ihippik/wal-listener/v2/internal/publisher"
 )
 

internal/publisher/event.go

@@ -2,10 +2,11 @@ package publisher
 
 import (
 	"fmt"
-	"github.com/ihippik/wal-listener/v2/internal/config"
 	"time"
 
 	"github.com/google/uuid"
+
+	"github.com/ihippik/wal-listener/v2/internal/config"
 )
 
 // Event structure for publishing to the NATS server.

internal/publisher/kafka.go

@@ -5,11 +5,12 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
-	"github.com/ihippik/wal-listener/v2/internal/config"
 	"os"
 
 	"github.com/IBM/sarama"
 	"github.com/goccy/go-json"
+
+	"github.com/ihippik/wal-listener/v2/internal/config"
 )
 
 // KafkaPublisher represent event publisher with Kafka broker.

internal/publisher/nats_test.go

@@ -1,8 +1,9 @@
 package publisher
 
 import (
-	"github.com/ihippik/wal-listener/v2/internal/config"
 	"testing"
+
+	"github.com/ihippik/wal-listener/v2/internal/config"
 )
 
 func TestEvent_GetSubjectName(t *testing.T) {