mirror of
https://git.mills.io/prologic/msgbus.git
synced 2024-06-28 09:41:43 +00:00
7b71102aa8
Closes #20 Co-authored-by: James Mills <prologic@shortcircuit.net.au> Co-authored-by: xuu <xuu@noreply@mills.io> Reviewed-on: https://git.mills.io/prologic/msgbus/pulls/26
384 lines
7.6 KiB
Go
384 lines
7.6 KiB
Go
package msgbus
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"flag"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"testing"
|
|
|
|
log "github.com/sirupsen/logrus"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"nhooyr.io/websocket"
|
|
"nhooyr.io/websocket/wsjson"
|
|
)
|
|
|
|
var (
|
|
debug = flag.Bool("d", false, "enable debug logging")
|
|
)
|
|
|
|
func TestMessageBusLen(t *testing.T) {
|
|
mb := New(nil)
|
|
assert.Equal(t, mb.Len(), 0)
|
|
}
|
|
|
|
func TestMessage(t *testing.T) {
|
|
mb := New(nil)
|
|
assert.Equal(t, mb.Len(), 0)
|
|
|
|
topic := mb.NewTopic("foo")
|
|
expected := mb.NewMessage(topic, []byte("bar"))
|
|
mb.Put(expected)
|
|
|
|
actual, ok := mb.Get(topic)
|
|
assert.True(t, ok)
|
|
assert.Equal(t, actual, expected)
|
|
}
|
|
|
|
func TestMessageIds(t *testing.T) {
|
|
assert := assert.New(t)
|
|
require := require.New(t)
|
|
|
|
mb := New(nil)
|
|
assert.Equal(0, mb.Len())
|
|
|
|
topic := mb.NewTopic("foo")
|
|
expected := mb.NewMessage(topic, []byte("bar"))
|
|
mb.Put(expected)
|
|
|
|
actual, ok := mb.Get(topic)
|
|
require.True(ok)
|
|
assert.Equal(expected, actual)
|
|
|
|
mb.Put(mb.NewMessage(topic, []byte("bar")))
|
|
msg, ok := mb.Get(topic)
|
|
require.True(ok)
|
|
assert.Equal(msg.ID, 1)
|
|
|
|
}
|
|
|
|
func TestMessageGetEmpty(t *testing.T) {
|
|
mb := New(nil)
|
|
assert.Equal(t, mb.Len(), 0)
|
|
|
|
topic := mb.NewTopic("foo")
|
|
msg, ok := mb.Get(topic)
|
|
assert.False(t, ok)
|
|
assert.Equal(t, msg, Message{})
|
|
}
|
|
|
|
func TestMessageBusPutGet(t *testing.T) {
|
|
mb := New(nil)
|
|
topic := mb.NewTopic("foo")
|
|
expected := mb.NewMessage(topic, []byte("foo"))
|
|
mb.Put(expected)
|
|
|
|
actual, ok := mb.Get(topic)
|
|
assert.True(t, ok)
|
|
assert.Equal(t, actual, expected)
|
|
}
|
|
|
|
func TestMessageBusSubscribe(t *testing.T) {
|
|
mb := New(nil)
|
|
|
|
msgs := mb.Subscribe("id1", "foo")
|
|
|
|
topic := mb.NewTopic("foo")
|
|
expected := mb.NewMessage(topic, []byte("foo"))
|
|
mb.Put(expected)
|
|
|
|
actual := <-msgs
|
|
assert.Equal(t, actual, expected)
|
|
}
|
|
|
|
func TestMessageBusSubscribeWithIndex(t *testing.T) {
|
|
assert := assert.New(t)
|
|
|
|
mb := New(nil)
|
|
|
|
msgs := mb.Subscribe("id1", "foo")
|
|
|
|
topic := mb.NewTopic("foo")
|
|
expected := mb.NewMessage(topic, []byte("foo"))
|
|
mb.Put(expected)
|
|
|
|
actual := <-msgs
|
|
assert.Equal(expected, actual)
|
|
assert.Equal(0, actual.ID)
|
|
|
|
mb.Unsubscribe("id1", "foo")
|
|
|
|
mb.Put(mb.NewMessage(topic, []byte("bar"))) // ID == 1
|
|
mb.Put(mb.NewMessage(topic, []byte("baz"))) // ID == 2
|
|
|
|
msgs = mb.Subscribe("id1", "foo", WithIndex(1))
|
|
|
|
assert.Equal([]byte("bar"), (<-msgs).Payload)
|
|
assert.Equal([]byte("baz"), (<-msgs).Payload)
|
|
}
|
|
|
|
func TestServeHTTPGETIndexEmpty(t *testing.T) {
|
|
assert := assert.New(t)
|
|
|
|
mb := New(nil)
|
|
w := httptest.NewRecorder()
|
|
r, _ := http.NewRequest("GET", "/", nil)
|
|
|
|
mb.ServeHTTP(w, r)
|
|
assert.Equal(w.Code, http.StatusOK)
|
|
assert.Equal(w.Body.String(), "{}")
|
|
}
|
|
|
|
func TestServeHTTPGETTopics(t *testing.T) {
|
|
assert := assert.New(t)
|
|
|
|
mb := New(nil)
|
|
|
|
mb.Put(mb.NewMessage(mb.NewTopic("foo"), []byte("foo")))
|
|
mb.Put(mb.NewMessage(mb.NewTopic("hello"), []byte("hello world")))
|
|
|
|
w := httptest.NewRecorder()
|
|
r, _ := http.NewRequest("GET", "/", nil)
|
|
|
|
mb.ServeHTTP(w, r)
|
|
assert.Equal(w.Code, http.StatusOK)
|
|
assert.Contains(w.Body.String(), "foo")
|
|
assert.Contains(w.Body.String(), "hello")
|
|
}
|
|
|
|
func TestServeHTTPGETEmptyQueue(t *testing.T) {
|
|
assert := assert.New(t)
|
|
|
|
mb := New(nil)
|
|
w := httptest.NewRecorder()
|
|
r, _ := http.NewRequest("GET", "/hello", nil)
|
|
|
|
mb.ServeHTTP(w, r)
|
|
assert.Equal(w.Code, http.StatusNoContent)
|
|
}
|
|
|
|
func TestServeHTTPPOST(t *testing.T) {
|
|
assert := assert.New(t)
|
|
|
|
mb := New(nil)
|
|
w := httptest.NewRecorder()
|
|
b := bytes.NewBufferString("hello world")
|
|
r, _ := http.NewRequest("POST", "/hello", b)
|
|
|
|
mb.ServeHTTP(w, r)
|
|
assert.Equal(w.Code, http.StatusAccepted)
|
|
}
|
|
|
|
func TestServeHTTPMaxPayloadSize(t *testing.T) {
|
|
assert := assert.New(t)
|
|
|
|
mb := New(nil)
|
|
w := httptest.NewRecorder()
|
|
b := bytes.NewBuffer(bytes.Repeat([]byte{'X'}, (DefaultMaxPayloadSize * 2)))
|
|
r, _ := http.NewRequest("POST", "/hello", b)
|
|
|
|
mb.ServeHTTP(w, r)
|
|
assert.Equal(http.StatusRequestEntityTooLarge, w.Code)
|
|
assert.Regexp(`payload exceeds max-payload-size`, w.Body.String())
|
|
}
|
|
|
|
func TestServeHTTPSimple(t *testing.T) {
|
|
assert := assert.New(t)
|
|
|
|
mb := New(nil)
|
|
|
|
w := httptest.NewRecorder()
|
|
b := bytes.NewBufferString("hello world")
|
|
r, _ := http.NewRequest("POST", "/hello", b)
|
|
|
|
mb.ServeHTTP(w, r)
|
|
assert.Equal(w.Code, http.StatusAccepted)
|
|
|
|
w = httptest.NewRecorder()
|
|
r, _ = http.NewRequest("GET", "/hello", nil)
|
|
|
|
mb.ServeHTTP(w, r)
|
|
assert.Equal(w.Code, http.StatusOK)
|
|
|
|
var msg *Message
|
|
json.Unmarshal(w.Body.Bytes(), &msg)
|
|
assert.Equal(msg.ID, 0)
|
|
assert.Equal(msg.Topic.Name, "hello")
|
|
assert.Equal(msg.Payload, []byte("hello world"))
|
|
}
|
|
|
|
func BenchmarkServeHTTPPOST(b *testing.B) {
|
|
mb := New(nil)
|
|
|
|
b.ResetTimer()
|
|
for i := 0; i < b.N; i++ {
|
|
w := httptest.NewRecorder()
|
|
b := bytes.NewBufferString("hello world")
|
|
r, _ := http.NewRequest("POST", "/hello", b)
|
|
|
|
mb.ServeHTTP(w, r)
|
|
}
|
|
}
|
|
|
|
func TestServeHTTPSubscriber(t *testing.T) {
|
|
assert := assert.New(t)
|
|
require := require.New(t)
|
|
|
|
mb := New(nil)
|
|
|
|
s := httptest.NewServer(mb)
|
|
defer s.Close()
|
|
|
|
msgs := make(chan Message, 10)
|
|
ready := make(chan bool, 1)
|
|
|
|
consumer := func() {
|
|
var msg Message
|
|
|
|
// u := fmt.Sprintf("ws%s/hello", strings.TrimPrefix(s.URL, "http"))
|
|
ws, _, err := websocket.Dial(context.Background(), s.URL+"/hello", nil)
|
|
require.NoError(err)
|
|
defer ws.Close(websocket.StatusNormalClosure, "")
|
|
|
|
ready <- true
|
|
|
|
err = wsjson.Read(context.Background(), ws, &msg)
|
|
require.NoError(err)
|
|
|
|
msgs <- msg
|
|
}
|
|
|
|
go consumer()
|
|
|
|
<-ready
|
|
|
|
c := s.Client()
|
|
b := bytes.NewBufferString("hello world")
|
|
r, err := c.Post(s.URL+"/hello", "text/plain", b)
|
|
require.NoError(err)
|
|
defer r.Body.Close()
|
|
|
|
msg := <-msgs
|
|
assert.Equal(msg.ID, 0)
|
|
assert.Equal(msg.Topic.Name, "hello")
|
|
assert.Equal(msg.Payload, []byte("hello world"))
|
|
}
|
|
|
|
func TestServeHTTPSubscriberReconnect(t *testing.T) {
|
|
assert := assert.New(t)
|
|
require := require.New(t)
|
|
|
|
mb := New(nil)
|
|
|
|
s := httptest.NewServer(mb)
|
|
|
|
msgs := make(chan Message, 10)
|
|
ready := make(chan bool, 1)
|
|
|
|
consumer := func() {
|
|
var msg Message
|
|
|
|
ws, _, err := websocket.Dial(context.Background(), s.URL+"/hello", nil)
|
|
require.NoError(err)
|
|
defer ws.Close(websocket.StatusNormalClosure, "")
|
|
|
|
ready <- true
|
|
|
|
err = wsjson.Read(context.Background(), ws, &msg)
|
|
require.NoError(err)
|
|
|
|
msgs <- msg
|
|
}
|
|
|
|
go consumer()
|
|
|
|
<-ready
|
|
|
|
s.Close()
|
|
s = httptest.NewServer(mb)
|
|
defer s.Close()
|
|
|
|
c := s.Client()
|
|
b := bytes.NewBufferString("hello world")
|
|
r, err := c.Post(s.URL+"/hello", "text/plain", b)
|
|
require.NoError(err)
|
|
defer r.Body.Close()
|
|
|
|
msg := <-msgs
|
|
assert.Equal(msg.ID, 0)
|
|
assert.Equal(msg.Topic.Name, "hello")
|
|
assert.Equal(msg.Payload, []byte("hello world"))
|
|
}
|
|
|
|
func TestMsgBusMetrics(t *testing.T) {
|
|
assert := assert.New(t)
|
|
|
|
opts := Options{
|
|
WithMetrics: true,
|
|
}
|
|
mb := New(&opts)
|
|
|
|
assert.IsType(&Metrics{}, mb.Metrics())
|
|
}
|
|
|
|
func BenchmarkMessageBusPut(b *testing.B) {
|
|
mb := New(nil)
|
|
topic := mb.NewTopic("foo")
|
|
msg := mb.NewMessage(topic, []byte("foo"))
|
|
b.ResetTimer()
|
|
for i := 0; i < b.N; i++ {
|
|
mb.Put(msg)
|
|
}
|
|
}
|
|
|
|
func BenchmarkMessageBusGet(b *testing.B) {
|
|
mb := New(nil)
|
|
topic := mb.NewTopic("foo")
|
|
msg := mb.NewMessage(topic, []byte("foo"))
|
|
for i := 0; i < b.N; i++ {
|
|
mb.Put(msg)
|
|
}
|
|
b.ResetTimer()
|
|
for i := 0; i < b.N; i++ {
|
|
mb.Get(topic)
|
|
}
|
|
}
|
|
|
|
func BenchmarkMessageBusGetEmpty(b *testing.B) {
|
|
mb := New(nil)
|
|
topic := mb.NewTopic("foo")
|
|
b.ResetTimer()
|
|
for i := 0; i < b.N; i++ {
|
|
mb.Get(topic)
|
|
}
|
|
}
|
|
|
|
func BenchmarkMessageBusPutGet(b *testing.B) {
|
|
mb := New(nil)
|
|
topic := mb.NewTopic("foo")
|
|
msg := mb.NewMessage(topic, []byte("foo"))
|
|
b.ResetTimer()
|
|
for i := 0; i < b.N; i++ {
|
|
mb.Put(msg)
|
|
mb.Get(topic)
|
|
}
|
|
}
|
|
|
|
func TestMain(m *testing.M) {
|
|
flag.Parse()
|
|
|
|
if *debug {
|
|
log.SetLevel(log.DebugLevel)
|
|
} else {
|
|
log.SetLevel(log.WarnLevel)
|
|
}
|
|
|
|
result := m.Run()
|
|
|
|
os.Exit(result)
|
|
}
|