update broker test

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

broker/broker_test.go

@@ -1,19 +1,141 @@
-package broker
+package broker_test
 
 import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 
+	kg "github.com/twmb/franz-go/pkg/kgo"
+	kgo "github.com/unistack-org/micro-broker-kgo/v3"
 	jsoncodec "github.com/unistack-org/micro-codec-json/v3"
 	"github.com/unistack-org/micro/v3/broker"
+	"github.com/unistack-org/micro/v3/logger"
+	"github.com/unistack-org/micro/v3/metadata"
 )
 
-func TestBrokerMessage(t *testing.T) {
-	c := jsoncodec.NewCodec()
+var (
+	msgcnt    = int64(60000)
+	group     = "33"
+	prefill   = true
+	subtopic  = "subtest"
+	pubtopic  = "pubtest"
+	rateRecs  int64
+	rateBytes int64
+)
 
-	buf := []byte(`{"Header":{"Content-Type":"application\/json","Micro-Topic":"notify"},"Body":{"Class":{}}}`)
-	msg := &broker.Message{}
+var subbm = &broker.Message{
+	Header: map[string]string{"hkey": "hval", metadata.HeaderTopic: subtopic},
+	Body:   []byte(`"bodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybody"`),
+}
 
-	if err := c.Unmarshal(buf, msg); err != nil {
+var pubbm = &broker.Message{
+	Header: map[string]string{"hkey": "hval", metadata.HeaderTopic: pubtopic},
+	Body:   []byte(`"bodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybodybody"`),
+}
+
+func printRate() {
+	recs := atomic.SwapInt64(&rateRecs, 0)
+	bytes := atomic.SwapInt64(&rateBytes, 0)
+	fmt.Printf("%0.2f MiB/s; %0.2fk records/s\n", float64(bytes)/(1024*1024), float64(recs)/1000)
+}
+
+func TestKgo(t *testing.T) {
+	if tr := os.Getenv("INTEGRATION_TESTS"); len(tr) > 0 {
+		t.Skip()
+	}
+
+	logger.DefaultLogger.Init(logger.WithLevel(logger.TraceLevel), logger.WithCallerSkipCount(3))
+	ctx := context.Background()
+
+	var addrs []string
+	if addr := os.Getenv("BROKER_ADDRS"); len(addr) == 0 {
+		addrs = []string{"172.18.0.201:29091", "172.18.0.202:29092", "172.18.0.203:29093"}
+	} else {
+		addrs = strings.Split(addr, ",")
+	}
+
+	b := kgo.NewBroker(
+		broker.Codec(jsoncodec.NewCodec()),
+		broker.Addrs(addrs...),
+		kgo.CommitInterval(5*time.Second),
+		kgo.Options(kg.ClientID("test"), kg.FetchMaxBytes(1*1024*1024)),
+	)
+	if err := b.Init(); err != nil {
 		t.Fatal(err)
 	}
+
+	if err := b.Connect(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	defer func() {
+		if err := b.Disconnect(ctx); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	if prefill {
+		msgs := make([]*broker.Message, 0, msgcnt)
+		for i := int64(0); i < msgcnt; i++ {
+			msgs = append(msgs, subbm)
+		}
+
+		if err := b.BatchPublish(ctx, msgs); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	done := make(chan bool, 1)
+	idx := int64(0)
+
+	fn := func(m broker.Event) error {
+		atomic.AddInt64(&idx, 1)
+		time.Sleep(20 * time.Millisecond)
+		if err := b.BatchPublish(ctx, []*broker.Message{pubbm}); err != nil {
+			return err
+		}
+		atomic.AddInt64(&rateRecs, 1)
+		atomic.AddInt64(&rateBytes, int64(len(m.Message().Body)))
+		return m.Ack()
+	}
+
+	sub, err := b.Subscribe(ctx, subtopic, fn,
+		broker.SubscribeAutoAck(true),
+		broker.SubscribeGroup(group),
+		broker.SubscribeBodyOnly(true))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	defer func() {
+		if err := sub.Unsubscribe(ctx); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	// ticker := time.NewTicker(10 * time.Second)
+	// defer ticker.Stop()
+
+	pticker := time.NewTicker(1 * time.Second)
+	defer pticker.Stop()
+
+	go func() {
+		for {
+			select {
+			case <-pticker.C:
+				printRate()
+				if prc := atomic.LoadInt64(&idx); prc == msgcnt {
+					close(done)
+				}
+				//	case <-ticker.C:
+				//		close(done)
+			}
+		}
+	}()
+
+	<-done
 }