2020-08-18 16:19:53 +01:00
|
|
|
package nats
|
|
|
|
|
|
|
|
import (
|
|
|
|
"net"
|
|
|
|
"os/exec"
|
|
|
|
"sync"
|
|
|
|
"testing"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/google/uuid"
|
2020-08-19 17:47:17 +03:00
|
|
|
"github.com/unistack-org/micro/v3/events"
|
2020-08-18 16:19:53 +01:00
|
|
|
"github.com/stretchr/testify/assert"
|
|
|
|
)
|
|
|
|
|
|
|
|
type testPayload struct {
|
|
|
|
Message string
|
|
|
|
}
|
|
|
|
|
|
|
|
func TestStream(t *testing.T) {
|
|
|
|
_, err := exec.LookPath("nats-streaming-server")
|
|
|
|
if err != nil {
|
|
|
|
t.Skipf("Skipping nats test, nats-streaming-server binary is not detected")
|
|
|
|
}
|
|
|
|
|
|
|
|
conn, err := net.DialTimeout("tcp", ":4222", time.Millisecond*100)
|
|
|
|
if err != nil {
|
|
|
|
t.Skipf("Skipping nats test, could not connect to cluster on port 4222: %v", err)
|
|
|
|
}
|
|
|
|
if err := conn.Close(); err != nil {
|
|
|
|
t.Fatalf("Error closing test tcp connection to nats cluster")
|
|
|
|
}
|
|
|
|
|
|
|
|
stream, err := NewStream(ClusterID("test-cluster"))
|
|
|
|
assert.Nilf(t, err, "NewStream should not return an error")
|
|
|
|
assert.NotNilf(t, stream, "NewStream should return a stream object")
|
|
|
|
|
|
|
|
// TestMissingTopic will test the topic validation on publish
|
|
|
|
t.Run("TestMissingTopic", func(t *testing.T) {
|
2020-08-20 09:29:29 +01:00
|
|
|
err := stream.Publish("", nil)
|
2020-08-18 16:19:53 +01:00
|
|
|
assert.Equalf(t, err, events.ErrMissingTopic, "Publishing to a blank topic should return an error")
|
|
|
|
})
|
|
|
|
|
|
|
|
// TestSubscribeTopic will publish a message to the test topic. The subscriber will subscribe to the
|
|
|
|
// same test topic.
|
|
|
|
t.Run("TestSubscribeTopic", func(t *testing.T) {
|
|
|
|
payload := &testPayload{Message: "HelloWorld"}
|
|
|
|
metadata := map[string]string{"foo": "bar"}
|
|
|
|
|
|
|
|
// create the subscriber
|
2020-08-20 09:29:29 +01:00
|
|
|
evChan, err := stream.Subscribe("test")
|
2020-08-18 16:19:53 +01:00
|
|
|
assert.Nilf(t, err, "Subscribe should not return an error")
|
|
|
|
|
|
|
|
// setup the subscriber async
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
timeout := time.NewTimer(time.Millisecond * 250)
|
|
|
|
|
|
|
|
select {
|
|
|
|
case event, _ := <-evChan:
|
|
|
|
assert.NotNilf(t, event, "The message was nil")
|
|
|
|
assert.Equal(t, event.Metadata, metadata, "Metadata didn't match")
|
|
|
|
|
|
|
|
var result testPayload
|
|
|
|
err = event.Unmarshal(&result)
|
|
|
|
assert.Nil(t, err, "Error decoding result")
|
|
|
|
assert.Equal(t, result, *payload, "Payload didn't match")
|
|
|
|
|
|
|
|
wg.Done()
|
|
|
|
case <-timeout.C:
|
|
|
|
t.Fatalf("Event was not recieved")
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
2020-08-20 09:29:29 +01:00
|
|
|
err = stream.Publish("test", payload, events.WithMetadata(metadata))
|
2020-08-18 16:19:53 +01:00
|
|
|
assert.Nil(t, err, "Publishing a valid message should not return an error")
|
|
|
|
wg.Add(1)
|
|
|
|
|
|
|
|
// wait for the subscriber to recieve the message or timeout
|
|
|
|
wg.Wait()
|
|
|
|
})
|
|
|
|
|
|
|
|
// TestSubscribeQueue will publish a message to a random topic. Two subscribers will then consume
|
|
|
|
// the message from the firehose topic with different queues. The second subscriber will be registered
|
|
|
|
// after the message is published to test durability.
|
|
|
|
t.Run("TestSubscribeQueue", func(t *testing.T) {
|
|
|
|
topic := uuid.New().String()
|
|
|
|
payload := &testPayload{Message: "HelloWorld"}
|
|
|
|
metadata := map[string]string{"foo": "bar"}
|
|
|
|
|
|
|
|
// create the first subscriber
|
2020-08-20 09:29:29 +01:00
|
|
|
evChan1, err := stream.Subscribe(topic)
|
2020-08-18 16:19:53 +01:00
|
|
|
assert.Nilf(t, err, "Subscribe should not return an error")
|
|
|
|
|
|
|
|
// setup the subscriber async
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
timeout := time.NewTimer(time.Millisecond * 250)
|
|
|
|
|
|
|
|
select {
|
|
|
|
case event, _ := <-evChan1:
|
|
|
|
assert.NotNilf(t, event, "The message was nil")
|
|
|
|
assert.Equal(t, event.Metadata, metadata, "Metadata didn't match")
|
|
|
|
|
|
|
|
var result testPayload
|
|
|
|
err = event.Unmarshal(&result)
|
|
|
|
assert.Nil(t, err, "Error decoding result")
|
|
|
|
assert.Equal(t, result, *payload, "Payload didn't match")
|
|
|
|
|
|
|
|
wg.Done()
|
|
|
|
case <-timeout.C:
|
|
|
|
t.Fatalf("Event was not recieved")
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
2020-08-20 09:29:29 +01:00
|
|
|
err = stream.Publish(topic, payload, events.WithMetadata(metadata))
|
2020-08-18 16:19:53 +01:00
|
|
|
assert.Nil(t, err, "Publishing a valid message should not return an error")
|
|
|
|
wg.Add(2)
|
|
|
|
|
|
|
|
// create the second subscriber
|
2020-08-20 09:29:29 +01:00
|
|
|
evChan2, err := stream.Subscribe(topic,
|
2020-08-18 16:19:53 +01:00
|
|
|
events.WithQueue("second_queue"),
|
|
|
|
events.WithStartAtTime(time.Now().Add(time.Minute*-1)),
|
|
|
|
)
|
|
|
|
assert.Nilf(t, err, "Subscribe should not return an error")
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
timeout := time.NewTimer(time.Millisecond * 250)
|
|
|
|
|
|
|
|
select {
|
|
|
|
case event, _ := <-evChan2:
|
|
|
|
assert.NotNilf(t, event, "The message was nil")
|
|
|
|
assert.Equal(t, event.Metadata, metadata, "Metadata didn't match")
|
|
|
|
|
|
|
|
var result testPayload
|
|
|
|
err = event.Unmarshal(&result)
|
|
|
|
assert.Nil(t, err, "Error decoding result")
|
|
|
|
assert.Equal(t, result, *payload, "Payload didn't match")
|
|
|
|
|
|
|
|
wg.Done()
|
|
|
|
case <-timeout.C:
|
|
|
|
t.Fatalf("Event was not recieved")
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
// wait for the subscriber to recieve the message or timeout
|
|
|
|
wg.Wait()
|
|
|
|
})
|
|
|
|
}
|