unistack-org/micro-broker-kgo
4
0
Fork 2
Code Issues 3 Pull Requests 2 Actions Packages Projects Releases 73 Wiki Activity

new kgo version
All checks were successful
sync / sync (push) Has been skipped
Details
coverage / build (push) Successful in 4m56s
Details
test / test (push) Successful in 4m59s
Details

Browse Source 
update deps

fixup race conditions

add kfake usage

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов
2025-05-11 01:49:38 +03:00
parent ccd912adb2
commit e34f57a515
6 changed files with 104 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
kgo_test.go
View File

@@ -2,15 +2,15 @@ package kgo_test
import (
"context"
"os"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/twmb/franz-go/pkg/kfake"
kg "github.com/twmb/franz-go/pkg/kgo"
kgo "go.unistack.org/micro-broker-kgo/v4"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/logger/slog"
"go.unistack.org/micro/v4/metadata"
@@ -20,31 +20,33 @@ var (
msgcnt = int64(1200)
group = "38"
prefill = true
loglevel = logger.DebugLevel
loglevel = logger.ErrorLevel
cluster *kfake.Cluster
)
func TestFail(t *testing.T) {
if tr := os.Getenv("INTEGRATION_TESTS"); len(tr) > 0 {
t.Skip()
}
func TestMain(m *testing.M) {
cluster = kfake.MustCluster(
kfake.AllowAutoTopicCreation(),
)
defer cluster.Close()
m.Run()
}
func TestFail(t *testing.T) {
logger.DefaultLogger = slog.NewLogger()
if err := logger.DefaultLogger.Init(logger.WithLevel(loglevel)); err != nil {
t.Fatal(err)
}
ctx := context.Background()
var addrs []string
if addr := os.Getenv("BROKER_ADDRS"); len(addr) == 0 {
addrs = []string{"127.0.0.1:9092"}
} else {
addrs = strings.Split(addr, ",")
}
b := kgo.NewBroker(
broker.Addrs(addrs...),
broker.ContentType("application/octet-stream"),
broker.Codec("application/octet-stream", codec.NewCodec()),
broker.Addrs(cluster.ListenAddrs()...),
kgo.CommitInterval(5*time.Second),
kgo.Options(kg.ClientID("test"), kg.FetchMaxBytes(10*1024*1024),
kgo.Options(
kg.ClientID("test"),
kg.FetchMaxBytes(10*1024*1024),
kg.AllowAutoTopicCreation(),
),
)
@@ -69,23 +71,28 @@ func TestFail(t *testing.T) {
t.Logf("broker health %v", b.Health())
msgs := make([]broker.Message, 0, msgcnt)
for i := int64(0); i < msgcnt; i++ {
m, _ := b.NewMessage(ctx, metadata.Pairs("hkey", "hval"), []byte(`test`))
m, err := b.NewMessage(ctx, metadata.Pairs("hkey", "hval"), []byte(`test`))
if err != nil {
t.Fatal(err)
}
msgs = append(msgs, m)
}
for _, msg := range msgs {
t.Logf("broker publish")
if err := b.Publish(ctx, "test", msg); err != nil {
t.Fatal(err)
go func() {
for _, msg := range msgs {
// t.Logf("broker publish")
if err := b.Publish(ctx, "test", msg); err != nil {
t.Fatal(err)
}
}
}
}()
// t.Skip()
idx := int64(0)
fn := func(msg broker.Message) error {
atomic.AddInt64(&idx, 1)
time.Sleep(500 * time.Millisecond)
t.Logf("ack")
time.Sleep(100 * time.Millisecond)
// t.Logf("ack")
return msg.Ack()
}
@@ -96,6 +103,7 @@ func TestFail(t *testing.T) {
if err != nil {
t.Fatal(err)
}
defer func() {
if err := sub.Unsubscribe(ctx); err != nil {
t.Fatal(err)
@@ -108,17 +116,26 @@ func TestFail(t *testing.T) {
t.Logf("health works")
break
}
time.Sleep(1 * time.Second)
t.Logf("health sleep")
time.Sleep(100 * time.Millisecond)
if err := b.Disconnect(ctx); err != nil {
t.Fatal(err)
}
}
}
func TestConnect(t *testing.T) {
var addrs []string
ctx := context.TODO()
b := kgo.NewBroker(
broker.Addrs(addrs...),
broker.ContentType("application/octet-stream"),
broker.Codec("application/octet-stream", codec.NewCodec()),
broker.Addrs(cluster.ListenAddrs()...),
kgo.CommitInterval(5*time.Second),
kgo.Options(kg.ClientID("test"), kg.FetchMaxBytes(10*1024*1024)),
kgo.Options(
kg.ClientID("test"),
kg.FetchMaxBytes(10*1024*1024),
kg.AllowAutoTopicCreation(),
),
)
if err := b.Init(); err != nil {
t.Fatal(err)
@@ -130,26 +147,21 @@ func TestConnect(t *testing.T) {
}
func TestPubSub(t *testing.T) {
if tr := os.Getenv("INTEGRATION_TESTS"); len(tr) > 0 {
t.Skip()
}
if err := logger.DefaultLogger.Init(logger.WithLevel(loglevel)); err != nil {
t.Fatal(err)
}
ctx := context.Background()
var addrs []string
if addr := os.Getenv("BROKER_ADDRS"); len(addr) == 0 {
addrs = []string{"127.0.0.1:29091", "127.0.0.2:29092", "127.0.0.3:29093"}
} else {
addrs = strings.Split(addr, ",")
}
b := kgo.NewBroker(
broker.Addrs(addrs...),
broker.ContentType("application/octet-stream"),
broker.Codec("application/octet-stream", codec.NewCodec()),
broker.Addrs(cluster.ListenAddrs()...),
kgo.CommitInterval(5*time.Second),
kgo.Options(kg.ClientID("test"), kg.FetchMaxBytes(10*1024*1024)),
kgo.Options(
kg.ClientID("test"),
kg.FetchMaxBytes(10*1024*1024),
kg.AllowAutoTopicCreation(),
),
)
if err := b.Init(); err != nil {