forwarder New tests

Author: nikolayk812

errors.go

@@ -9,4 +9,7 @@ var (
 	ErrTableEmpty = errors.New("table is empty")
 
 	ErrPoolNil = errors.New("pool is nil")
+
+	ErrReaderNil    = errors.New("reader is nil")
+	ErrPublisherNil = errors.New("publisher is nil")
 )

forwarder.go

@@ -2,7 +2,6 @@ package outbox
 
 import (
 	"context"
-	"errors"
 	"fmt"
 
 	"github.com/jackc/pgx/v5/pgxpool"
@@ -24,10 +23,10 @@ type forwarder struct {
 
 func NewForwarder(reader Reader, publisher Publisher, opts ...ForwardOption) (Forwarder, error) {
 	if reader == nil {
-		return nil, errors.New("reader is nil")
+		return nil, ErrReaderNil
 	}
 	if publisher == nil {
-		return nil, errors.New("publisher is nil")
+		return nil, ErrPublisherNil
 	}
 
 	f := &forwarder{

forwarder_test.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"testing"
 
+	"github.com/jackc/pgx/v5/pgxpool"
 	outbox "github.com/nikolayk812/pgx-outbox"
 	"github.com/nikolayk812/pgx-outbox/internal/fakes"
 	"github.com/nikolayk812/pgx-outbox/internal/mocks"
@@ -175,3 +176,113 @@ func TestForwarder_Forward(t *testing.T) {
 		})
 	}
 }
+
+func TestForwarder_New(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name      string
+		reader    outbox.Reader
+		publisher outbox.Publisher
+		options   []outbox.ForwardOption
+		wantErr   error
+	}{
+		{
+			name:    "nil reader",
+			reader:  nil,
+			wantErr: outbox.ErrReaderNil,
+		},
+		{
+			name:      "nil publisher",
+			reader:    new(mocks.Reader),
+			publisher: nil,
+			wantErr:   outbox.ErrPublisherNil,
+		},
+		{
+			name:      "with options",
+			reader:    new(mocks.Reader),
+			publisher: new(mocks.Publisher),
+			options:   []outbox.ForwardOption{outbox.WithForwardFilter(types.MessageFilter{Brokers: []string{"broker1"}})},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			forwarder, err := outbox.NewForwarder(tt.reader, tt.publisher, tt.options...)
+			if tt.wantErr != nil {
+				require.Error(t, err)
+				assert.ErrorIs(t, err, tt.wantErr)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.NotNil(t, forwarder)
+		})
+	}
+}
+
+func TestForwarder_NewFromPool(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name      string
+		table     string
+		pool      *pgxpool.Pool
+		publisher outbox.Publisher
+		options   []outbox.ForwardOption
+		wantErr   error
+	}{
+		{
+			name:    "empty table",
+			table:   "",
+			pool:    new(pgxpool.Pool),
+			wantErr: outbox.ErrTableEmpty,
+		},
+		{
+			name:    "nil pool",
+			table:   "outbox_messages",
+			pool:    nil,
+			wantErr: outbox.ErrPoolNil,
+		},
+		{
+			name:      "nil publisher",
+			table:     "outbox_messages",
+			pool:      new(pgxpool.Pool),
+			publisher: nil,
+			wantErr:   outbox.ErrPublisherNil,
+		},
+		{
+			name:      "valid inputs",
+			table:     "outbox_messages",
+			pool:      new(pgxpool.Pool),
+			publisher: new(mocks.Publisher),
+			wantErr:   nil,
+		},
+		{
+			name:      "with option",
+			table:     "outbox_messages",
+			pool:      new(pgxpool.Pool),
+			publisher: new(mocks.Publisher),
+			options:   []outbox.ForwardOption{outbox.WithForwardFilter(types.MessageFilter{Brokers: []string{"broker1"}})},
+			wantErr:   nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			forwarder, err := outbox.NewForwarderFromPool(tt.table, tt.pool, tt.publisher, tt.options...)
+			if tt.wantErr != nil {
+				require.ErrorIs(t, err, tt.wantErr)
+				assert.Nil(t, forwarder)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.NotNil(t, forwarder)
+		})
+	}
+}

writer.go

@@ -3,7 +3,6 @@ package outbox
 import (
 	"context"
 	"database/sql"
-	"errors"
 	"fmt"
 	"time"
 
@@ -161,10 +160,6 @@ func (w *writer) WriteBatch(ctx context.Context, tx pgx.Tx, messages []types.Mes
 type Tx interface{}
 
 func queryRow(ctx context.Context, tx Tx, q string, args ...interface{}) (pgx.Row, error) {
-	if tx == nil {
-		return nil, errors.New("tx is nil")
-	}
-
 	var row pgx.Row
 
 	switch t := tx.(type) {

writer_reader_test.go

@@ -363,7 +363,7 @@ func (suite *WriterReaderTestSuite) TestReader_ReadMessage() {
 	}
 }
 
-func (suite *WriterReaderTestSuite) TestWriter_MarkMessage() {
+func (suite *WriterReaderTestSuite) TestWriter_AckMessage() {
 	msg1 := fakes.FakeMessage()
 	msg2 := fakes.FakeMessage()
 	msg3 := fakes.FakeMessage()
@@ -726,7 +726,7 @@ func (suite *WriterReaderTestSuite) TestReader_New() {
 		name    string
 		table   string
 		pool    *pgxpool.Pool
-		option  outbox.ReadOption
+		options []outbox.ReadOption
 		wantErr error
 	}{
 		{
@@ -742,18 +742,18 @@ func (suite *WriterReaderTestSuite) TestReader_New() {
 			wantErr: outbox.ErrPoolNil,
 		},
 		{
-			name:   "with option",
-			table:  "outbox_messages",
-			pool:   suite.pool,
-			option: outbox.WithReadFilter(types.MessageFilter{Brokers: []string{"broker1"}}),
+			name:    "with options",
+			table:   "outbox_messages",
+			pool:    suite.pool,
+			options: []outbox.ReadOption{outbox.WithReadFilter(types.MessageFilter{Brokers: []string{"broker1"}})},
 		},
 	}
 
 	for _, tt := range tests {
 		suite.Run(tt.name, func() {
 			t := suite.T()
 
-			reader, err := outbox.NewReader(tt.table, tt.pool, tt.option)
+			reader, err := outbox.NewReader(tt.table, tt.pool, tt.options...)
 			if tt.wantErr != nil {
 				require.ErrorIs(t, err, tt.wantErr)
 				return