sqlx_test.go

package outbox

import (
	"context"
	"fmt"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/jmoiron/sqlx"
	"github.com/nilorg/eventbus"
)

func newTestSqlxEngine(t *testing.T) (engine Engine) {
	var (
		err error
		db  *sqlx.DB
	)
	db, err = sqlx.Connect("mysql", "root:test123@tcp(127.0.0.1:3306)/outbox?charset=utf8&parseTime=True&loc=Local")
	if err != nil {
		t.Fatalf("sqlx connect error: %v", err)
	}

	// 注意: AutoMigrate 会在 New 函数内部自动调用，无需手动创建表

	bus := newTestEventBus(t)
	opts := &EngineOptions{
		FailedRetryInterval:        time.Minute,
		FailedRetryCount:           50,
		DataCleanInterval:          time.Hour,
		SucceedMessageExpiredAfter: 24 * time.Hour,
		SnowflakeNode:              1,
		Logger:                     &eventbus.StdLogger{},
	}
	engine, err = New(EngineTypeSqlx, db, bus, opts)
	if err != nil {
		t.Fatalf("new sqlx engine error: %s", err)
	}
	return
}

func TestSqlx(t *testing.T) {
	engine := newTestSqlxEngine(t)
	ctx := context.Background()
	var err error
	topic := "order.create.success.sync"
	ctxGroup1 := eventbus.NewGroupIDContext(ctx, "nilorg.events.sync.group1")
	go func() {
		err = engine.Subscribe(ctxGroup1, topic, func(ctx context.Context, msg *Message) error {
			fmt.Printf("group1 %s: %+v is timeout: %v\n", topic, msg, msg.IsTimeout(3*time.Minute))
			return nil
		})
		if err != nil {
			t.Error(err)
			return
		}
	}()
	go func() {
		err = engine.Subscribe(ctxGroup1, topic, func(ctx context.Context, msg *Message) error {
			fmt.Printf("group1(copy) %s: %+v is timeout: %v\n", topic, msg, msg.IsTimeout(3*time.Minute))
			return nil
		})
		if err != nil {
			t.Error(err)
			return
		}
	}()

	ctxGroup2 := eventbus.NewGroupIDContext(ctx, "nilorg.events.sync.group2")
	go func() {
		err = engine.Subscribe(ctxGroup2, topic, func(ctx context.Context, msg *Message) error {
			fmt.Printf("group2 %s: %+v is timeout: %v\n", topic, msg, msg.IsTimeout(3*time.Minute))
			return nil
		})
		if err != nil {
			t.Error(err)
			return
		}
	}()
	time.Sleep(1 * time.Second)
	err = engine.Publish(ctx, topic, "publish 测试")
	if err != nil {
		t.Error(err)
		return
	}
	time.Sleep(2 * time.Second)
}