Function TestJetStreamAckPending_Push

test/js_test.go:2131–2280

(t *testing.T)

}

func TestJetStreamAckPending_Push(t *testing.T) {
	s := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, s)

	nc, js := jsClient(t, s)
	defer nc.Close()

	var err error

	_, err = js.AddStream(&nats.StreamConfig{
		Name:     "TEST",
		Subjects: []string{"foo"},
	})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	const totalMsgs = 3
	for i := 0; i < totalMsgs; i++ {
		if _, err := js.Publish("foo", []byte(fmt.Sprintf("msg %d", i))); err != nil {
			t.Fatal(err)
		}
	}

	sub, err := js.SubscribeSync("foo",
		nats.Durable("dname-wait"),
		nats.AckWait(100*time.Millisecond),
		nats.MaxDeliver(5),
		nats.MaxAckPending(3),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer sub.Unsubscribe()

	// 3 messages delivered 5 times.
	expected := 15
	timeout := time.Now().Add(2 * time.Second)
	pending := 0
	for time.Now().Before(timeout) {
		if pending, _, _ = sub.Pending(); pending >= expected {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	if pending < expected {
		t.Errorf("Expected %v, got %v", expected, pending)
	}

	info, err := sub.ConsumerInfo()
	if err != nil {
		t.Fatal(err)
	}

	got := info.NumRedelivered
	expected = 3
	if got < expected {
		t.Errorf("Expected %v, got: %v", expected, got)