Function TestDrainQueueSub

test/drain_test.go:77–126

(t *testing.T)

}

func TestDrainQueueSub(t *testing.T) {
	s := RunDefaultServer()
	defer s.Shutdown()
	nc := NewDefaultConnection(t)
	defer nc.Close()

	done := make(chan bool)
	received := int32(0)
	expected := int32(4096)
	numSubs := int32(32)

	checkDone := func() int32 {
		rcvd := atomic.AddInt32(&received, 1)
		if rcvd >= expected {
			done <- true
		}
		return rcvd
	}

	callback := func(m *nats.Msg) {
		rcvd := checkDone()
		// Randomly replace this sub from time to time.
		if rcvd%3 == 0 {
			m.Sub.Drain()
			// Create a new one that we will not drain.
			nc.QueueSubscribe("foo", "bar", func(m *nats.Msg) { checkDone() })
		}
	}

	for i := int32(0); i < numSubs; i++ {
		_, err := nc.QueueSubscribe("foo", "bar", callback)
		if err != nil {
			t.Fatalf("Error creating subscription; %v", err)
		}
	}

	for i := int32(0); i < expected; i++ {
		nc.Publish("foo", []byte("Don't forget about me"))
	}

	select {
	case <-done:
		break
	case <-time.After(5 * time.Second):
		r := atomic.LoadInt32(&received)
		if r != expected {
			t.Fatalf("Did not receive all messages: %d of %d", r, expected)
		}
	}
}

func waitFor(t *testing.T, totalWait, sleepDur time.Duration, f func() error) {
	t.Helper()