Function TestAutoUnsubWithParallelNextMsgCalls

test/sub_test.go:237–309

(t *testing.T)

}

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

	rch := make(chan bool, 1)

	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()

	numRoutines := 3
	max := 100
	total := max * 2
	received := int64(0)

	var wg sync.WaitGroup

	sub, err := nc.SubscribeSync("foo")
	if err != nil {
		t.Fatalf("Failed to subscribe: %v", err)
	}
	sub.AutoUnsubscribe(int(max))
	nc.Flush()

	wg.Add(numRoutines)

	for i := range numRoutines {
		go func(s *nats.Subscription, idx int) {
			for {
				// The first to reach the max delivered will cause the
				// subscription to be removed, which will kick out all
				// other calls to NextMsg. So don't be afraid of the long
				// timeout.
				_, err := s.NextMsg(3 * time.Second)
				if err != nil {
					break
				}
				atomic.AddInt64(&received, 1)
			}
			wg.Done()
		}(sub, i)
	}

	msg := []byte("Hello")
	for range max / 2 {
		nc.Publish("foo", msg)
	}
	nc.Flush()

	s.Shutdown()
	s = RunDefaultServer()
	defer s.Shutdown()