2016-03-14 22:37:51 +03:00
|
|
|
package broker
|
|
|
|
|
|
|
|
import (
|
2016-04-28 02:16:11 +03:00
|
|
|
"sync"
|
2016-03-14 22:37:51 +03:00
|
|
|
"testing"
|
2016-06-25 05:15:56 +03:00
|
|
|
"time"
|
2016-03-14 22:37:51 +03:00
|
|
|
|
2018-11-21 13:13:03 +03:00
|
|
|
"github.com/google/uuid"
|
2019-12-18 19:12:25 +03:00
|
|
|
"github.com/micro/go-micro/debug/log/noop"
|
2019-11-16 13:55:11 +03:00
|
|
|
"github.com/micro/go-micro/registry"
|
2019-01-14 18:27:25 +03:00
|
|
|
"github.com/micro/go-micro/registry/memory"
|
2019-05-31 02:38:05 +03:00
|
|
|
"github.com/micro/go-micro/util/log"
|
2016-03-14 22:37:51 +03:00
|
|
|
)
|
|
|
|
|
2019-11-16 13:55:11 +03:00
|
|
|
func newTestRegistry() registry.Registry {
|
|
|
|
return memory.NewRegistry(memory.Services(testData))
|
2019-01-14 18:27:25 +03:00
|
|
|
}
|
|
|
|
|
2016-06-25 05:15:56 +03:00
|
|
|
func sub(be *testing.B, c int) {
|
2019-01-16 21:54:04 +03:00
|
|
|
// set no op logger
|
2019-12-18 19:12:25 +03:00
|
|
|
log.SetLogger(noop.NewLog())
|
2019-01-16 21:54:04 +03:00
|
|
|
|
2016-06-25 05:15:56 +03:00
|
|
|
be.StopTimer()
|
2019-01-14 18:27:25 +03:00
|
|
|
m := newTestRegistry()
|
|
|
|
|
2016-06-25 05:15:56 +03:00
|
|
|
b := NewBroker(Registry(m))
|
2018-11-21 13:13:03 +03:00
|
|
|
topic := uuid.New().String()
|
2016-06-25 05:15:56 +03:00
|
|
|
|
|
|
|
if err := b.Init(); err != nil {
|
|
|
|
be.Fatalf("Unexpected init error: %v", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := b.Connect(); err != nil {
|
|
|
|
be.Fatalf("Unexpected connect error: %v", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
msg := &Message{
|
|
|
|
Header: map[string]string{
|
|
|
|
"Content-Type": "application/json",
|
|
|
|
},
|
|
|
|
Body: []byte(`{"message": "Hello World"}`),
|
|
|
|
}
|
|
|
|
|
|
|
|
var subs []Subscriber
|
|
|
|
done := make(chan bool, c)
|
|
|
|
|
|
|
|
for i := 0; i < c; i++ {
|
2019-07-07 14:44:09 +03:00
|
|
|
sub, err := b.Subscribe(topic, func(p Event) error {
|
2016-06-25 05:15:56 +03:00
|
|
|
done <- true
|
|
|
|
m := p.Message()
|
|
|
|
|
|
|
|
if string(m.Body) != string(msg.Body) {
|
|
|
|
be.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}, Queue("shared"))
|
|
|
|
if err != nil {
|
|
|
|
be.Fatalf("Unexpected subscribe error: %v", err)
|
|
|
|
}
|
|
|
|
subs = append(subs, sub)
|
|
|
|
}
|
|
|
|
|
|
|
|
for i := 0; i < be.N; i++ {
|
|
|
|
be.StartTimer()
|
|
|
|
if err := b.Publish(topic, msg); err != nil {
|
|
|
|
be.Fatalf("Unexpected publish error: %v", err)
|
|
|
|
}
|
|
|
|
<-done
|
|
|
|
be.StopTimer()
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, sub := range subs {
|
|
|
|
sub.Unsubscribe()
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := b.Disconnect(); err != nil {
|
|
|
|
be.Fatalf("Unexpected disconnect error: %v", err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func pub(be *testing.B, c int) {
|
2019-01-16 21:54:04 +03:00
|
|
|
// set no op logger
|
2019-12-18 19:12:25 +03:00
|
|
|
log.SetLogger(noop.NewLog())
|
2019-01-16 21:54:04 +03:00
|
|
|
|
2016-06-25 05:15:56 +03:00
|
|
|
be.StopTimer()
|
2019-01-14 18:27:25 +03:00
|
|
|
m := newTestRegistry()
|
2016-06-25 05:15:56 +03:00
|
|
|
b := NewBroker(Registry(m))
|
2018-11-21 13:13:03 +03:00
|
|
|
topic := uuid.New().String()
|
2016-06-25 05:15:56 +03:00
|
|
|
|
|
|
|
if err := b.Init(); err != nil {
|
|
|
|
be.Fatalf("Unexpected init error: %v", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := b.Connect(); err != nil {
|
|
|
|
be.Fatalf("Unexpected connect error: %v", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
msg := &Message{
|
|
|
|
Header: map[string]string{
|
|
|
|
"Content-Type": "application/json",
|
|
|
|
},
|
|
|
|
Body: []byte(`{"message": "Hello World"}`),
|
|
|
|
}
|
|
|
|
|
|
|
|
done := make(chan bool, c*4)
|
|
|
|
|
2019-07-07 14:44:09 +03:00
|
|
|
sub, err := b.Subscribe(topic, func(p Event) error {
|
2016-06-25 05:15:56 +03:00
|
|
|
done <- true
|
|
|
|
m := p.Message()
|
|
|
|
if string(m.Body) != string(msg.Body) {
|
|
|
|
be.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}, Queue("shared"))
|
|
|
|
if err != nil {
|
|
|
|
be.Fatalf("Unexpected subscribe error: %v", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
ch := make(chan int, c*4)
|
|
|
|
be.StartTimer()
|
|
|
|
|
|
|
|
for i := 0; i < c; i++ {
|
|
|
|
go func() {
|
2019-11-01 18:07:53 +03:00
|
|
|
for range ch {
|
2016-06-25 05:15:56 +03:00
|
|
|
if err := b.Publish(topic, msg); err != nil {
|
|
|
|
be.Fatalf("Unexpected publish error: %v", err)
|
|
|
|
}
|
|
|
|
select {
|
|
|
|
case <-done:
|
|
|
|
case <-time.After(time.Second):
|
|
|
|
}
|
|
|
|
wg.Done()
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
}
|
|
|
|
|
|
|
|
for i := 0; i < be.N; i++ {
|
|
|
|
wg.Add(1)
|
|
|
|
ch <- i
|
|
|
|
}
|
|
|
|
|
|
|
|
wg.Wait()
|
|
|
|
be.StopTimer()
|
|
|
|
sub.Unsubscribe()
|
|
|
|
close(ch)
|
|
|
|
close(done)
|
|
|
|
|
|
|
|
if err := b.Disconnect(); err != nil {
|
|
|
|
be.Fatalf("Unexpected disconnect error: %v", err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-03-14 22:37:51 +03:00
|
|
|
func TestBroker(t *testing.T) {
|
2019-01-14 18:27:25 +03:00
|
|
|
m := newTestRegistry()
|
2016-03-16 01:12:28 +03:00
|
|
|
b := NewBroker(Registry(m))
|
2016-03-14 22:37:51 +03:00
|
|
|
|
|
|
|
if err := b.Init(); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected init error: %v", err)
|
2016-03-14 22:37:51 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
if err := b.Connect(); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected connect error: %v", err)
|
2016-03-14 22:37:51 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
msg := &Message{
|
|
|
|
Header: map[string]string{
|
|
|
|
"Content-Type": "application/json",
|
|
|
|
},
|
|
|
|
Body: []byte(`{"message": "Hello World"}`),
|
|
|
|
}
|
|
|
|
|
|
|
|
done := make(chan bool)
|
|
|
|
|
2019-07-07 14:44:09 +03:00
|
|
|
sub, err := b.Subscribe("test", func(p Event) error {
|
2016-03-14 22:37:51 +03:00
|
|
|
m := p.Message()
|
|
|
|
|
|
|
|
if string(m.Body) != string(msg.Body) {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
|
2016-03-14 22:37:51 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
close(done)
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
if err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected subscribe error: %v", err)
|
2016-03-14 22:37:51 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
if err := b.Publish("test", msg); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected publish error: %v", err)
|
2016-03-14 22:37:51 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
<-done
|
|
|
|
sub.Unsubscribe()
|
|
|
|
|
|
|
|
if err := b.Disconnect(); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected disconnect error: %v", err)
|
2016-03-14 22:37:51 +03:00
|
|
|
}
|
|
|
|
}
|
2016-04-28 02:16:11 +03:00
|
|
|
|
|
|
|
func TestConcurrentSubBroker(t *testing.T) {
|
2019-01-14 18:27:25 +03:00
|
|
|
m := newTestRegistry()
|
2016-04-28 02:16:11 +03:00
|
|
|
b := NewBroker(Registry(m))
|
|
|
|
|
|
|
|
if err := b.Init(); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected init error: %v", err)
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
if err := b.Connect(); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected connect error: %v", err)
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
msg := &Message{
|
|
|
|
Header: map[string]string{
|
|
|
|
"Content-Type": "application/json",
|
|
|
|
},
|
|
|
|
Body: []byte(`{"message": "Hello World"}`),
|
|
|
|
}
|
|
|
|
|
|
|
|
var subs []Subscriber
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
|
|
|
|
for i := 0; i < 10; i++ {
|
2019-07-07 14:44:09 +03:00
|
|
|
sub, err := b.Subscribe("test", func(p Event) error {
|
2016-04-28 02:16:11 +03:00
|
|
|
defer wg.Done()
|
|
|
|
|
|
|
|
m := p.Message()
|
|
|
|
|
|
|
|
if string(m.Body) != string(msg.Body) {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
if err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected subscribe error: %v", err)
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
wg.Add(1)
|
|
|
|
subs = append(subs, sub)
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := b.Publish("test", msg); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected publish error: %v", err)
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
wg.Wait()
|
|
|
|
|
|
|
|
for _, sub := range subs {
|
|
|
|
sub.Unsubscribe()
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := b.Disconnect(); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected disconnect error: %v", err)
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func TestConcurrentPubBroker(t *testing.T) {
|
2019-01-14 18:27:25 +03:00
|
|
|
m := newTestRegistry()
|
2016-04-28 02:16:11 +03:00
|
|
|
b := NewBroker(Registry(m))
|
|
|
|
|
|
|
|
if err := b.Init(); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected init error: %v", err)
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
if err := b.Connect(); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected connect error: %v", err)
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
msg := &Message{
|
|
|
|
Header: map[string]string{
|
|
|
|
"Content-Type": "application/json",
|
|
|
|
},
|
|
|
|
Body: []byte(`{"message": "Hello World"}`),
|
|
|
|
}
|
|
|
|
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
|
2019-07-07 14:44:09 +03:00
|
|
|
sub, err := b.Subscribe("test", func(p Event) error {
|
2016-04-28 02:16:11 +03:00
|
|
|
defer wg.Done()
|
|
|
|
|
|
|
|
m := p.Message()
|
|
|
|
|
|
|
|
if string(m.Body) != string(msg.Body) {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
if err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected subscribe error: %v", err)
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
for i := 0; i < 10; i++ {
|
|
|
|
wg.Add(1)
|
|
|
|
|
|
|
|
if err := b.Publish("test", msg); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected publish error: %v", err)
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
wg.Wait()
|
|
|
|
|
|
|
|
sub.Unsubscribe()
|
|
|
|
|
|
|
|
if err := b.Disconnect(); err != nil {
|
2016-06-25 05:15:56 +03:00
|
|
|
t.Fatalf("Unexpected disconnect error: %v", err)
|
2016-04-28 02:16:11 +03:00
|
|
|
}
|
|
|
|
}
|
2016-06-25 05:15:56 +03:00
|
|
|
|
|
|
|
func BenchmarkSub1(b *testing.B) {
|
|
|
|
sub(b, 1)
|
|
|
|
}
|
|
|
|
func BenchmarkSub8(b *testing.B) {
|
|
|
|
sub(b, 8)
|
|
|
|
}
|
|
|
|
|
|
|
|
func BenchmarkSub32(b *testing.B) {
|
|
|
|
sub(b, 32)
|
|
|
|
}
|
|
|
|
|
|
|
|
func BenchmarkSub64(b *testing.B) {
|
|
|
|
sub(b, 64)
|
|
|
|
}
|
|
|
|
|
|
|
|
func BenchmarkSub128(b *testing.B) {
|
|
|
|
sub(b, 128)
|
|
|
|
}
|
|
|
|
|
|
|
|
func BenchmarkPub1(b *testing.B) {
|
|
|
|
pub(b, 1)
|
|
|
|
}
|
|
|
|
|
|
|
|
func BenchmarkPub8(b *testing.B) {
|
|
|
|
pub(b, 8)
|
|
|
|
}
|
|
|
|
|
|
|
|
func BenchmarkPub32(b *testing.B) {
|
|
|
|
pub(b, 32)
|
|
|
|
}
|
|
|
|
|
|
|
|
func BenchmarkPub64(b *testing.B) {
|
|
|
|
pub(b, 64)
|
|
|
|
}
|
|
|
|
|
|
|
|
func BenchmarkPub128(b *testing.B) {
|
|
|
|
pub(b, 128)
|
|
|
|
}
|