Function TestAsyncSubscriptionPending

test/sub_test.go:1129–1214

(t *testing.T)

}

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

	nc := NewDefaultConnection(t)
	defer nc.Close()

	// Send some messages to ourselves.
	total := 100
	msg := []byte("0123456789")

	inCb := make(chan bool)
	block := make(chan bool)

	sub, _ := nc.Subscribe("foo", func(m *nats.Msg) {
		inCb <- true
		<-block
		// Avoid repeated calls to this callback
		m.Sub.Unsubscribe()
	})
	defer sub.Unsubscribe()

	for range total {
		nc.Publish("foo", msg)
	}
	nc.Flush()

	// Wait that a message is received, so checks are safe
	if err := Wait(inCb); err != nil {
		t.Fatal("No message received")
	}

	// Test old way
	q, _, _ := sub.Pending()
	if q != total && q != total-1 {
		t.Fatalf("Expected %d or %d, got %d", total, total-1, q)
	}

	// New way, make sure the same and check bytes.
	m, b, _ := sub.Pending()
	mlen := len(msg)
	totalSize := total * mlen

	if m != total && m != total-1 {
		t.Fatalf("Expected msgs of %d or %d, got %d", total, total-1, m)
	}
	if b != totalSize && b != totalSize-mlen {
		t.Fatalf("Expected bytes of %d or %d, got %d",
			totalSize, totalSize-mlen, b)
	}

	// Make sure max has been set. Since we block after the first message is
	// received, MaxPending should be >= total - 1 and <= total
	mm, bm, _ := sub.MaxPending()
	if mm < total-1 || mm > total {
		t.Fatalf("Expected max msgs (%d) to be between %d and %d",
			mm, total-1, total)
	}