Function TestAsyncSubscriptionPendingDrain

test/sub_test.go:1216–1254

(t *testing.T)

}

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

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

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

	sub, _ := nc.Subscribe("foo", func(_ *nats.Msg) {})
	defer sub.Unsubscribe()

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

	// Wait for all delivered.
	waitFor(t, 2*time.Second, 15*time.Millisecond, func() error {
		if d, _ := sub.Delivered(); d != int64(total) {
			return fmt.Errorf("Wrong delivered count: %v vs %v", d, total)
		}
		m, b, _ := sub.Pending()
		if m != 0 {
			return fmt.Errorf("Expected msgs of 0, got %d", m)
		}
		if b != 0 {
			return fmt.Errorf("Expected bytes of 0, got %d", b)
		}
		return nil
	})

	sub.Unsubscribe()
	if _, err := sub.Delivered(); err == nil {
		t.Fatal("Calling Delivered() on closed subscription should fail")
	}
}

func TestSyncSubscriptionPendingDrain(t *testing.T) {
	s := RunDefaultServer()