micro/broker/http_broker_test.go

344 lines
6.3 KiB
Go
Raw Normal View History

2016-03-14 22:37:51 +03:00
package broker
import (
"sync"
2016-03-14 22:37:51 +03:00
"testing"
"time"
2016-03-14 22:37:51 +03:00
"github.com/google/uuid"
2019-01-14 18:27:25 +03:00
"github.com/micro/go-micro/registry/memory"
2016-03-14 22:37:51 +03:00
)
2019-01-14 18:27:25 +03:00
func newTestRegistry() *memory.Registry {
r := memory.NewRegistry()
m := r.(*memory.Registry)
m.Setup()
return m
}
func sub(be *testing.B, c int) {
be.StopTimer()
2019-01-14 18:27:25 +03:00
m := newTestRegistry()
b := NewBroker(Registry(m))
topic := uuid.New().String()
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++ {
sub, err := b.Subscribe(topic, func(p Publication) error {
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) {
be.StopTimer()
2019-01-14 18:27:25 +03:00
m := newTestRegistry()
b := NewBroker(Registry(m))
topic := uuid.New().String()
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)
sub, err := b.Subscribe(topic, func(p Publication) error {
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() {
for _ = range ch {
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 {
t.Fatalf("Unexpected init error: %v", err)
2016-03-14 22:37:51 +03:00
}
if err := b.Connect(); err != nil {
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)
sub, err := b.Subscribe("test", func(p Publication) error {
m := p.Message()
if string(m.Body) != string(msg.Body) {
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 {
t.Fatalf("Unexpected subscribe error: %v", err)
2016-03-14 22:37:51 +03:00
}
if err := b.Publish("test", msg); err != nil {
t.Fatalf("Unexpected publish error: %v", err)
2016-03-14 22:37:51 +03:00
}
<-done
sub.Unsubscribe()
if err := b.Disconnect(); err != nil {
t.Fatalf("Unexpected disconnect error: %v", err)
2016-03-14 22:37:51 +03:00
}
}
func TestConcurrentSubBroker(t *testing.T) {
2019-01-14 18:27:25 +03:00
m := newTestRegistry()
b := NewBroker(Registry(m))
if err := b.Init(); err != nil {
t.Fatalf("Unexpected init error: %v", err)
}
if err := b.Connect(); err != nil {
t.Fatalf("Unexpected connect error: %v", err)
}
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++ {
sub, err := b.Subscribe("test", func(p Publication) error {
defer wg.Done()
m := p.Message()
if string(m.Body) != string(msg.Body) {
t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
}
return nil
})
if err != nil {
t.Fatalf("Unexpected subscribe error: %v", err)
}
wg.Add(1)
subs = append(subs, sub)
}
if err := b.Publish("test", msg); err != nil {
t.Fatalf("Unexpected publish error: %v", err)
}
wg.Wait()
for _, sub := range subs {
sub.Unsubscribe()
}
if err := b.Disconnect(); err != nil {
t.Fatalf("Unexpected disconnect error: %v", err)
}
}
func TestConcurrentPubBroker(t *testing.T) {
2019-01-14 18:27:25 +03:00
m := newTestRegistry()
b := NewBroker(Registry(m))
if err := b.Init(); err != nil {
t.Fatalf("Unexpected init error: %v", err)
}
if err := b.Connect(); err != nil {
t.Fatalf("Unexpected connect error: %v", err)
}
msg := &Message{
Header: map[string]string{
"Content-Type": "application/json",
},
Body: []byte(`{"message": "Hello World"}`),
}
var wg sync.WaitGroup
sub, err := b.Subscribe("test", func(p Publication) error {
defer wg.Done()
m := p.Message()
if string(m.Body) != string(msg.Body) {
t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
}
return nil
})
if err != nil {
t.Fatalf("Unexpected subscribe error: %v", err)
}
for i := 0; i < 10; i++ {
wg.Add(1)
if err := b.Publish("test", msg); err != nil {
t.Fatalf("Unexpected publish error: %v", err)
}
}
wg.Wait()
sub.Unsubscribe()
if err := b.Disconnect(); err != nil {
t.Fatalf("Unexpected disconnect error: %v", err)
}
}
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)
}