mirror of https://github.com/tidwall/tile38.git
909 lines
23 KiB
Go
909 lines
23 KiB
Go
// Copyright 2012-2018 The NATS Authors
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package test
|
|
|
|
import (
|
|
"bytes"
|
|
"fmt"
|
|
"math"
|
|
"regexp"
|
|
"runtime"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/go-nats"
|
|
)
|
|
|
|
func TestCloseLeakingGoRoutines(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
|
|
// Give time for things to settle before capturing the number of
|
|
// go routines
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
base := runtime.NumGoroutine()
|
|
|
|
nc := NewDefaultConnection(t)
|
|
|
|
nc.Flush()
|
|
nc.Close()
|
|
|
|
// Give time for things to settle before capturing the number of
|
|
// go routines
|
|
waitFor(t, 2*time.Second, 100*time.Millisecond, func() error {
|
|
delta := (runtime.NumGoroutine() - base)
|
|
if delta > 0 {
|
|
return fmt.Errorf("%d Go routines still exist post Close()", delta)
|
|
}
|
|
return nil
|
|
})
|
|
|
|
// Make sure we can call Close() multiple times
|
|
nc.Close()
|
|
}
|
|
|
|
func TestLeakingGoRoutinesOnFailedConnect(t *testing.T) {
|
|
// Give time for things to settle before capturing the number of
|
|
// go routines
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
base := runtime.NumGoroutine()
|
|
|
|
nc, err := nats.Connect(nats.DefaultURL)
|
|
if err == nil {
|
|
nc.Close()
|
|
t.Fatalf("Expected failure to connect")
|
|
}
|
|
|
|
// Give time for things to settle before capturing the number of
|
|
// go routines
|
|
waitFor(t, 2*time.Second, 100*time.Millisecond, func() error {
|
|
delta := (runtime.NumGoroutine() - base)
|
|
if delta > 0 {
|
|
return fmt.Errorf("%d Go routines still exist post Close()", delta)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
func TestConnectedServer(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
u := nc.ConnectedUrl()
|
|
if u == "" || u != nats.DefaultURL {
|
|
t.Fatalf("Unexpected connected URL of %s\n", u)
|
|
}
|
|
srv := nc.ConnectedServerId()
|
|
if srv == "" {
|
|
t.Fatal("Expected a connected server id")
|
|
}
|
|
nc.Close()
|
|
u = nc.ConnectedUrl()
|
|
if u != "" {
|
|
t.Fatalf("Expected a nil connected URL, got %s\n", u)
|
|
}
|
|
srv = nc.ConnectedServerId()
|
|
if srv != "" {
|
|
t.Fatalf("Expected a nil connect server, got %s\n", srv)
|
|
}
|
|
}
|
|
|
|
func TestMultipleClose(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < 10; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
nc.Close()
|
|
wg.Done()
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
}
|
|
|
|
func TestBadOptionTimeoutConnect(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
|
|
opts := nats.GetDefaultOptions()
|
|
opts.Timeout = -1
|
|
opts.Url = "nats://localhost:4222"
|
|
|
|
_, err := opts.Connect()
|
|
if err == nil {
|
|
t.Fatal("Expected an error")
|
|
}
|
|
if err != nats.ErrNoServers {
|
|
t.Fatalf("Expected a ErrNoServers error: Got %v\n", err)
|
|
}
|
|
}
|
|
|
|
func TestSimplePublish(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
if err := nc.Publish("foo", []byte("Hello World")); err != nil {
|
|
t.Fatal("Failed to publish string message: ", err)
|
|
}
|
|
}
|
|
|
|
func TestSimplePublishNoData(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
if err := nc.Publish("foo", nil); err != nil {
|
|
t.Fatal("Failed to publish empty message: ", err)
|
|
}
|
|
}
|
|
|
|
func TestPublishDoesNotFailOnSlowConsumer(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
sub, err := nc.SubscribeSync("foo")
|
|
if err != nil {
|
|
t.Fatalf("Unable to create subscription: %v", err)
|
|
}
|
|
|
|
if err := sub.SetPendingLimits(1, 1000); err != nil {
|
|
t.Fatalf("Unable to set pending limits: %v", err)
|
|
}
|
|
|
|
var pubErr error
|
|
|
|
msg := []byte("Hello")
|
|
for i := 0; i < 10; i++ {
|
|
pubErr = nc.Publish("foo", msg)
|
|
if pubErr != nil {
|
|
break
|
|
}
|
|
nc.Flush()
|
|
}
|
|
|
|
if pubErr != nil {
|
|
t.Fatalf("Publish() should not fail because of slow consumer. Got '%v'", pubErr)
|
|
}
|
|
}
|
|
|
|
func TestAsyncSubscribe(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
omsg := []byte("Hello World")
|
|
ch := make(chan bool)
|
|
|
|
// Callback is mandatory
|
|
if _, err := nc.Subscribe("foo", nil); err == nil {
|
|
t.Fatal("Creating subscription without callback should have failed")
|
|
}
|
|
|
|
_, err := nc.Subscribe("foo", func(m *nats.Msg) {
|
|
if !bytes.Equal(m.Data, omsg) {
|
|
t.Fatal("Message received does not match")
|
|
}
|
|
if m.Sub == nil {
|
|
t.Fatal("Callback does not have a valid Subscription")
|
|
}
|
|
ch <- true
|
|
})
|
|
if err != nil {
|
|
t.Fatal("Failed to subscribe: ", err)
|
|
}
|
|
nc.Publish("foo", omsg)
|
|
if e := Wait(ch); e != nil {
|
|
t.Fatal("Message not received for subscription")
|
|
}
|
|
}
|
|
|
|
func TestAsyncSubscribeRoutineLeakOnUnsubscribe(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
ch := make(chan bool)
|
|
|
|
// Give time for things to settle before capturing the number of
|
|
// go routines
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
// Take the base once the connection is established, but before
|
|
// the subscriber is created.
|
|
base := runtime.NumGoroutine()
|
|
|
|
sub, err := nc.Subscribe("foo", func(m *nats.Msg) { ch <- true })
|
|
if err != nil {
|
|
t.Fatal("Failed to subscribe: ", err)
|
|
}
|
|
|
|
// Send to ourself
|
|
nc.Publish("foo", []byte("hello"))
|
|
|
|
// This ensures that the async delivery routine is up and running.
|
|
if err := Wait(ch); err != nil {
|
|
t.Fatal("Failed to receive message")
|
|
}
|
|
|
|
// Make sure to give it time to go back into wait
|
|
time.Sleep(200 * time.Millisecond)
|
|
|
|
// Explicit unsubscribe
|
|
sub.Unsubscribe()
|
|
|
|
// Give time for things to settle before capturing the number of
|
|
// go routines
|
|
waitFor(t, 2*time.Second, 100*time.Millisecond, func() error {
|
|
delta := (runtime.NumGoroutine() - base)
|
|
if delta > 0 {
|
|
return fmt.Errorf("%d Go routines still exist post Unsubscribe()", delta)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
func TestAsyncSubscribeRoutineLeakOnClose(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
|
|
ch := make(chan bool)
|
|
|
|
// Give time for things to settle before capturing the number of
|
|
// go routines
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
// Take the base before creating the connection, since we are going
|
|
// to close it before taking the delta.
|
|
base := runtime.NumGoroutine()
|
|
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
_, err := nc.Subscribe("foo", func(m *nats.Msg) { ch <- true })
|
|
if err != nil {
|
|
t.Fatal("Failed to subscribe: ", err)
|
|
}
|
|
|
|
// Send to ourself
|
|
nc.Publish("foo", []byte("hello"))
|
|
|
|
// This ensures that the async delivery routine is up and running.
|
|
if err := Wait(ch); err != nil {
|
|
t.Fatal("Failed to receive message")
|
|
}
|
|
|
|
// Make sure to give it time to go back into wait
|
|
time.Sleep(200 * time.Millisecond)
|
|
|
|
// Close connection without explicit unsubscribe
|
|
nc.Close()
|
|
|
|
// Give time for things to settle before capturing the number of
|
|
// go routines
|
|
waitFor(t, 2*time.Second, 100*time.Millisecond, func() error {
|
|
delta := (runtime.NumGoroutine() - base)
|
|
if delta > 0 {
|
|
return fmt.Errorf("%d Go routines still exist post Close()", delta)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
func TestSyncSubscribe(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
sub, err := nc.SubscribeSync("foo")
|
|
if err != nil {
|
|
t.Fatal("Failed to subscribe: ", err)
|
|
}
|
|
omsg := []byte("Hello World")
|
|
nc.Publish("foo", omsg)
|
|
msg, err := sub.NextMsg(1 * time.Second)
|
|
if err != nil || !bytes.Equal(msg.Data, omsg) {
|
|
t.Fatal("Message received does not match")
|
|
}
|
|
}
|
|
|
|
func TestPubSubWithReply(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
sub, err := nc.SubscribeSync("foo")
|
|
if err != nil {
|
|
t.Fatal("Failed to subscribe: ", err)
|
|
}
|
|
omsg := []byte("Hello World")
|
|
nc.PublishMsg(&nats.Msg{Subject: "foo", Reply: "bar", Data: omsg})
|
|
msg, err := sub.NextMsg(10 * time.Second)
|
|
if err != nil || !bytes.Equal(msg.Data, omsg) {
|
|
t.Fatal("Message received does not match")
|
|
}
|
|
}
|
|
|
|
func TestFlush(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
omsg := []byte("Hello World")
|
|
for i := 0; i < 10000; i++ {
|
|
nc.Publish("flush", omsg)
|
|
}
|
|
if err := nc.FlushTimeout(0); err == nil {
|
|
t.Fatal("Calling FlushTimeout() with invalid timeout should fail")
|
|
}
|
|
if err := nc.Flush(); err != nil {
|
|
t.Fatalf("Received error from flush: %s\n", err)
|
|
}
|
|
if nb, _ := nc.Buffered(); nb > 0 {
|
|
t.Fatalf("Outbound buffer not empty: %d bytes\n", nb)
|
|
}
|
|
|
|
nc.Close()
|
|
if _, err := nc.Buffered(); err == nil {
|
|
t.Fatal("Calling Buffered() on closed connection should fail")
|
|
}
|
|
}
|
|
|
|
func TestQueueSubscriber(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
s1, _ := nc.QueueSubscribeSync("foo", "bar")
|
|
s2, _ := nc.QueueSubscribeSync("foo", "bar")
|
|
omsg := []byte("Hello World")
|
|
nc.Publish("foo", omsg)
|
|
nc.Flush()
|
|
r1, _ := s1.QueuedMsgs()
|
|
r2, _ := s2.QueuedMsgs()
|
|
if (r1 + r2) != 1 {
|
|
t.Fatal("Received too many messages for multiple queue subscribers")
|
|
}
|
|
// Drain messages
|
|
s1.NextMsg(time.Second)
|
|
s2.NextMsg(time.Second)
|
|
|
|
total := 1000
|
|
for i := 0; i < total; i++ {
|
|
nc.Publish("foo", omsg)
|
|
}
|
|
nc.Flush()
|
|
v := uint(float32(total) * 0.15)
|
|
r1, _ = s1.QueuedMsgs()
|
|
r2, _ = s2.QueuedMsgs()
|
|
if r1+r2 != total {
|
|
t.Fatalf("Incorrect number of messages: %d vs %d", (r1 + r2), total)
|
|
}
|
|
expected := total / 2
|
|
d1 := uint(math.Abs(float64(expected - r1)))
|
|
d2 := uint(math.Abs(float64(expected - r2)))
|
|
if d1 > v || d2 > v {
|
|
t.Fatalf("Too much variance in totals: %d, %d > %d", d1, d2, v)
|
|
}
|
|
}
|
|
|
|
func TestReplyArg(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
ch := make(chan bool)
|
|
replyExpected := "bar"
|
|
|
|
nc.Subscribe("foo", func(m *nats.Msg) {
|
|
if m.Reply != replyExpected {
|
|
t.Fatalf("Did not receive correct reply arg in callback: "+
|
|
"('%s' vs '%s')", m.Reply, replyExpected)
|
|
}
|
|
ch <- true
|
|
})
|
|
nc.PublishMsg(&nats.Msg{Subject: "foo", Reply: replyExpected, Data: []byte("Hello")})
|
|
if e := Wait(ch); e != nil {
|
|
t.Fatal("Did not receive callback")
|
|
}
|
|
}
|
|
|
|
func TestSyncReplyArg(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
replyExpected := "bar"
|
|
sub, _ := nc.SubscribeSync("foo")
|
|
nc.PublishMsg(&nats.Msg{Subject: "foo", Reply: replyExpected, Data: []byte("Hello")})
|
|
msg, err := sub.NextMsg(1 * time.Second)
|
|
if err != nil {
|
|
t.Fatal("Received an err on NextMsg()")
|
|
}
|
|
if msg.Reply != replyExpected {
|
|
t.Fatalf("Did not receive correct reply arg in callback: "+
|
|
"('%s' vs '%s')", msg.Reply, replyExpected)
|
|
}
|
|
}
|
|
|
|
func TestUnsubscribe(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
received := int32(0)
|
|
max := int32(10)
|
|
ch := make(chan bool)
|
|
nc.Subscribe("foo", func(m *nats.Msg) {
|
|
atomic.AddInt32(&received, 1)
|
|
if received == max {
|
|
err := m.Sub.Unsubscribe()
|
|
if err != nil {
|
|
t.Fatal("Unsubscribe failed with err:", err)
|
|
}
|
|
ch <- true
|
|
}
|
|
})
|
|
send := 20
|
|
for i := 0; i < send; i++ {
|
|
nc.Publish("foo", []byte("hello"))
|
|
}
|
|
nc.Flush()
|
|
<-ch
|
|
|
|
r := atomic.LoadInt32(&received)
|
|
if r != max {
|
|
t.Fatalf("Received wrong # of messages after unsubscribe: %d vs %d",
|
|
r, max)
|
|
}
|
|
}
|
|
|
|
func TestDoubleUnsubscribe(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
sub, err := nc.SubscribeSync("foo")
|
|
if err != nil {
|
|
t.Fatal("Failed to subscribe: ", err)
|
|
}
|
|
if err = sub.Unsubscribe(); err != nil {
|
|
t.Fatal("Unsubscribe failed with err:", err)
|
|
}
|
|
if err = sub.Unsubscribe(); err == nil {
|
|
t.Fatal("Unsubscribe should have reported an error")
|
|
}
|
|
}
|
|
|
|
func TestRequestTimeout(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
if _, err := nc.Request("foo", []byte("help"), 10*time.Millisecond); err == nil {
|
|
t.Fatalf("Expected to receive a timeout error")
|
|
}
|
|
}
|
|
|
|
func TestOldRequest(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
|
|
nc, err := nats.Connect(nats.DefaultURL, nats.UseOldRequestStyle())
|
|
if err != nil {
|
|
t.Fatalf("Failed to connect: %v", err)
|
|
}
|
|
defer nc.Close()
|
|
|
|
response := []byte("I will help you")
|
|
nc.Subscribe("foo", func(m *nats.Msg) {
|
|
nc.Publish(m.Reply, response)
|
|
})
|
|
msg, err := nc.Request("foo", []byte("help"), 500*time.Millisecond)
|
|
if err != nil {
|
|
t.Fatalf("Received an error on Request test: %s", err)
|
|
}
|
|
if !bytes.Equal(msg.Data, response) {
|
|
t.Fatalf("Received invalid response")
|
|
}
|
|
}
|
|
|
|
func TestRequest(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
response := []byte("I will help you")
|
|
nc.Subscribe("foo", func(m *nats.Msg) {
|
|
nc.Publish(m.Reply, response)
|
|
})
|
|
msg, err := nc.Request("foo", []byte("help"), 500*time.Millisecond)
|
|
if err != nil {
|
|
t.Fatalf("Received an error on Request test: %s", err)
|
|
}
|
|
if !bytes.Equal(msg.Data, response) {
|
|
t.Fatalf("Received invalid response")
|
|
}
|
|
}
|
|
|
|
func TestRequestNoBody(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
response := []byte("I will help you")
|
|
nc.Subscribe("foo", func(m *nats.Msg) {
|
|
nc.Publish(m.Reply, response)
|
|
})
|
|
msg, err := nc.Request("foo", nil, 500*time.Millisecond)
|
|
if err != nil {
|
|
t.Fatalf("Received an error on Request test: %s", err)
|
|
}
|
|
if !bytes.Equal(msg.Data, response) {
|
|
t.Fatalf("Received invalid response")
|
|
}
|
|
}
|
|
|
|
func TestSimultaneousRequests(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
response := []byte("I will help you")
|
|
nc.Subscribe("foo", func(m *nats.Msg) {
|
|
nc.Publish(m.Reply, response)
|
|
})
|
|
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < 50; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
if _, err := nc.Request("foo", nil, 2*time.Second); err != nil {
|
|
t.Fatalf("Expected to receive a timeout error")
|
|
} else {
|
|
wg.Done()
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
}
|
|
|
|
func TestRequestClose(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
wg := &sync.WaitGroup{}
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
time.Sleep(100 * time.Millisecond)
|
|
nc.Close()
|
|
}()
|
|
if _, err := nc.Request("foo", []byte("help"), 2*time.Second); err != nats.ErrInvalidConnection && err != nats.ErrConnectionClosed {
|
|
t.Fatalf("Expected connection error: got %v", err)
|
|
}
|
|
wg.Wait()
|
|
}
|
|
|
|
func TestRequestCloseTimeout(t *testing.T) {
|
|
// Make sure we return a timeout when we close
|
|
// the connection even if response is queued.
|
|
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
response := []byte("I will help you")
|
|
nc.Subscribe("foo", func(m *nats.Msg) {
|
|
nc.Publish(m.Reply, response)
|
|
nc.Close()
|
|
})
|
|
if _, err := nc.Request("foo", nil, 1*time.Second); err == nil {
|
|
t.Fatalf("Expected to receive a timeout error")
|
|
}
|
|
}
|
|
|
|
func TestFlushInCB(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
ch := make(chan bool)
|
|
|
|
nc.Subscribe("foo", func(_ *nats.Msg) {
|
|
nc.Flush()
|
|
ch <- true
|
|
})
|
|
nc.Publish("foo", []byte("Hello"))
|
|
if e := Wait(ch); e != nil {
|
|
t.Fatal("Flush did not return properly in callback")
|
|
}
|
|
}
|
|
|
|
func TestReleaseFlush(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
|
|
for i := 0; i < 1000; i++ {
|
|
nc.Publish("foo", []byte("Hello"))
|
|
}
|
|
go nc.Close()
|
|
nc.Flush()
|
|
}
|
|
|
|
func TestInbox(t *testing.T) {
|
|
inbox := nats.NewInbox()
|
|
if matched, _ := regexp.Match(`_INBOX.\S`, []byte(inbox)); !matched {
|
|
t.Fatal("Bad INBOX format")
|
|
}
|
|
}
|
|
|
|
func TestStats(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
data := []byte("The quick brown fox jumped over the lazy dog")
|
|
iter := 10
|
|
|
|
for i := 0; i < iter; i++ {
|
|
nc.Publish("foo", data)
|
|
}
|
|
|
|
if nc.OutMsgs != uint64(iter) {
|
|
t.Fatalf("Not properly tracking OutMsgs: received %d, wanted %d\n", nc.OutMsgs, iter)
|
|
}
|
|
obb := uint64(iter * len(data))
|
|
if nc.OutBytes != obb {
|
|
t.Fatalf("Not properly tracking OutBytes: received %d, wanted %d\n", nc.OutBytes, obb)
|
|
}
|
|
|
|
// Clear outbound
|
|
nc.OutMsgs, nc.OutBytes = 0, 0
|
|
|
|
// Test both sync and async versions of subscribe.
|
|
nc.Subscribe("foo", func(_ *nats.Msg) {})
|
|
nc.SubscribeSync("foo")
|
|
|
|
for i := 0; i < iter; i++ {
|
|
nc.Publish("foo", data)
|
|
}
|
|
nc.Flush()
|
|
|
|
if nc.InMsgs != uint64(2*iter) {
|
|
t.Fatalf("Not properly tracking InMsgs: received %d, wanted %d\n", nc.InMsgs, 2*iter)
|
|
}
|
|
|
|
ibb := 2 * obb
|
|
if nc.InBytes != ibb {
|
|
t.Fatalf("Not properly tracking InBytes: received %d, wanted %d\n", nc.InBytes, ibb)
|
|
}
|
|
}
|
|
|
|
func TestRaceSafeStats(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
go nc.Publish("foo", []byte("Hello World"))
|
|
time.Sleep(200 * time.Millisecond)
|
|
|
|
stats := nc.Stats()
|
|
|
|
if stats.OutMsgs != uint64(1) {
|
|
t.Fatalf("Not properly tracking OutMsgs: received %d, wanted %d\n", nc.OutMsgs, 1)
|
|
}
|
|
}
|
|
|
|
func TestBadSubject(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
nc := NewDefaultConnection(t)
|
|
defer nc.Close()
|
|
|
|
err := nc.Publish("", []byte("Hello World"))
|
|
if err == nil {
|
|
t.Fatalf("Expected an error on bad subject to publish")
|
|
}
|
|
if err != nats.ErrBadSubject {
|
|
t.Fatalf("Expected a ErrBadSubject error: Got %v\n", err)
|
|
}
|
|
}
|
|
|
|
func TestOptions(t *testing.T) {
|
|
s := RunDefaultServer()
|
|
defer s.Shutdown()
|
|
|
|
nc, err := nats.Connect(nats.DefaultURL,
|
|
nats.Name("myName"),
|
|
nats.MaxReconnects(2),
|
|
nats.ReconnectWait(50*time.Millisecond),
|
|
nats.PingInterval(20*time.Millisecond))
|
|
if err != nil {
|
|
t.Fatalf("Failed to connect: %v", err)
|
|
}
|
|
defer nc.Close()
|
|
|
|
rch := make(chan bool)
|
|
cch := make(chan bool)
|
|
|
|
nc.SetReconnectHandler(func(_ *nats.Conn) { rch <- true })
|
|
nc.SetClosedHandler(func(_ *nats.Conn) { cch <- true })
|
|
|
|
s.Shutdown()
|
|
|
|
s = RunDefaultServer()
|
|
defer s.Shutdown()
|
|
|
|
if err := Wait(rch); err != nil {
|
|
t.Fatal("Failed getting reconnected cb")
|
|
}
|
|
|
|
nc.Close()
|
|
|
|
if err := Wait(cch); err != nil {
|
|
t.Fatal("Failed getting closed cb")
|
|
}
|
|
|
|
nc, err = nats.Connect(nats.DefaultURL, nats.NoReconnect())
|
|
if err != nil {
|
|
t.Fatalf("Failed to connect: %v", err)
|
|
}
|
|
defer nc.Close()
|
|
|
|
nc.SetReconnectHandler(func(_ *nats.Conn) { rch <- true })
|
|
nc.SetClosedHandler(func(_ *nats.Conn) { cch <- true })
|
|
|
|
s.Shutdown()
|
|
|
|
// We should not get a reconnect cb this time
|
|
if err := WaitTime(rch, time.Second); err == nil {
|
|
t.Fatal("Unexpected reconnect cb")
|
|
}
|
|
|
|
nc.Close()
|
|
|
|
if err := Wait(cch); err != nil {
|
|
t.Fatal("Failed getting closed cb")
|
|
}
|
|
}
|
|
|
|
func TestNilConnection(t *testing.T) {
|
|
var nc *nats.Conn
|
|
data := []byte("ok")
|
|
|
|
// Publish
|
|
if err := nc.Publish("foo", data); err == nil || err != nats.ErrInvalidConnection {
|
|
t.Fatalf("Expected ErrInvalidConnection error, got %v\n", err)
|
|
}
|
|
if err := nc.PublishMsg(nil); err == nil || err != nats.ErrInvalidMsg {
|
|
t.Fatalf("Expected ErrInvalidMsg error, got %v\n", err)
|
|
}
|
|
if err := nc.PublishMsg(&nats.Msg{}); err == nil || err != nats.ErrInvalidConnection {
|
|
t.Fatalf("Expected ErrInvalidConnection error, got %v\n", err)
|
|
}
|
|
if err := nc.PublishRequest("foo", "reply", data); err == nil || err != nats.ErrInvalidConnection {
|
|
t.Fatalf("Expected ErrInvalidConnection error, got %v\n", err)
|
|
}
|
|
|
|
// Subscribe
|
|
if _, err := nc.Subscribe("foo", nil); err == nil || err != nats.ErrInvalidConnection {
|
|
t.Fatalf("Expected ErrInvalidConnection error, got %v\n", err)
|
|
}
|
|
if _, err := nc.SubscribeSync("foo"); err == nil || err != nats.ErrInvalidConnection {
|
|
t.Fatalf("Expected ErrInvalidConnection error, got %v\n", err)
|
|
}
|
|
if _, err := nc.QueueSubscribe("foo", "bar", nil); err == nil || err != nats.ErrInvalidConnection {
|
|
t.Fatalf("Expected ErrInvalidConnection error, got %v\n", err)
|
|
}
|
|
ch := make(chan *nats.Msg)
|
|
if _, err := nc.ChanSubscribe("foo", ch); err == nil || err != nats.ErrInvalidConnection {
|
|
t.Fatalf("Expected ErrInvalidConnection error, got %v\n", err)
|
|
}
|
|
if _, err := nc.ChanQueueSubscribe("foo", "bar", ch); err == nil || err != nats.ErrInvalidConnection {
|
|
t.Fatalf("Expected ErrInvalidConnection error, got %v\n", err)
|
|
}
|
|
if _, err := nc.QueueSubscribeSyncWithChan("foo", "bar", ch); err == nil || err != nats.ErrInvalidConnection {
|
|
t.Fatalf("Expected ErrInvalidConnection error, got %v\n", err)
|
|
}
|
|
|
|
// Flush
|
|
if err := nc.Flush(); err == nil || err != nats.ErrInvalidConnection {
|
|
t.Fatalf("Expected ErrInvalidConnection error, got %v\n", err)
|
|
}
|
|
if err := nc.FlushTimeout(time.Millisecond); err == nil || err != nats.ErrInvalidConnection {
|
|
t.Fatalf("Expected ErrInvalidConnection error, got %v\n", err)
|
|
}
|
|
|
|
// Nil Subscribers
|
|
var sub *nats.Subscription
|
|
if sub.Type() != nats.NilSubscription {
|
|
t.Fatalf("Got wrong type for nil subscription, %v\n", sub.Type())
|
|
}
|
|
if sub.IsValid() {
|
|
t.Fatalf("Expected IsValid() to return false")
|
|
}
|
|
if err := sub.Unsubscribe(); err == nil || err != nats.ErrBadSubscription {
|
|
t.Fatalf("Expected Unsubscribe to return proper error, got %v\n", err)
|
|
}
|
|
if err := sub.AutoUnsubscribe(1); err == nil || err != nats.ErrBadSubscription {
|
|
t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
|
|
}
|
|
if _, err := sub.NextMsg(time.Millisecond); err == nil || err != nats.ErrBadSubscription {
|
|
t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
|
|
}
|
|
if _, err := sub.QueuedMsgs(); err == nil || err != nats.ErrBadSubscription {
|
|
t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
|
|
}
|
|
if _, _, err := sub.Pending(); err == nil || err != nats.ErrBadSubscription {
|
|
t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
|
|
}
|
|
if _, _, err := sub.MaxPending(); err == nil || err != nats.ErrBadSubscription {
|
|
t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
|
|
}
|
|
if err := sub.ClearMaxPending(); err == nil || err != nats.ErrBadSubscription {
|
|
t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
|
|
}
|
|
if _, _, err := sub.PendingLimits(); err == nil || err != nats.ErrBadSubscription {
|
|
t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
|
|
}
|
|
if err := sub.SetPendingLimits(1, 1); err == nil || err != nats.ErrBadSubscription {
|
|
t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
|
|
}
|
|
if _, err := sub.Delivered(); err == nil || err != nats.ErrBadSubscription {
|
|
t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
|
|
}
|
|
if _, err := sub.Dropped(); err == nil || err != nats.ErrBadSubscription {
|
|
t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
|
|
}
|
|
}
|