Function TestAutoUnsubAndReconnect

test/sub_test.go:179–235

(t *testing.T)

}

func TestAutoUnsubAndReconnect(t *testing.T) {
	s := RunDefaultServer()
	defer s.Shutdown()

	rch := make(chan bool)

	nc, err := nats.Connect(nats.DefaultURL,
		nats.ReconnectWait(50*time.Millisecond),
		nats.ReconnectJitter(0, 0),
		nats.ReconnectHandler(func(_ *nats.Conn) { rch <- true }))
	if err != nil {
		t.Fatalf("Unable to connect: %v", err)
	}
	defer nc.Close()

	received := int32(0)
	max := int32(10)
	sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {
		atomic.AddInt32(&received, 1)
	})
	if err != nil {
		t.Fatalf("Failed to subscribe: %v", err)
	}
	sub.AutoUnsubscribe(int(max))

	// Send less than the max
	total := int(max / 2)
	for range total {
		nc.Publish("foo", []byte("Hello"))
	}
	nc.Flush()

	// Restart the server
	s.Shutdown()
	s = RunDefaultServer()
	defer s.Shutdown()

	// and wait to reconnect
	if err := Wait(rch); err != nil {
		t.Fatal("Failed to get the reconnect cb")
	}

	// Now send more than the total max.
	total = int(3 * max)
	for range total {
		nc.Publish("foo", []byte("Hello"))
	}
	nc.Flush()

	// Wait a bit before checking.
	time.Sleep(50 * time.Millisecond)

	// We should have received only up-to-max messages.
	if atomic.LoadInt32(&received) != max {
		t.Fatalf("Received %d msgs, wanted only %d\n", received, max)
	}
}