Function TestJetStreamAckPending_Pull

test/js_test.go:2079–2129

(t *testing.T)

}

func TestJetStreamAckPending_Pull(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 = 4
	for i := 0; i < totalMsgs; i++ {
		if _, err := js.Publish("foo", []byte(fmt.Sprintf("msg %d", i))); err != nil {
			t.Fatal(err)
		}
	}

	ackPendingLimit := 3
	sub, err := js.PullSubscribe("foo", "dname-pull-ack-wait", nats.MaxAckPending(ackPendingLimit))
	if err != nil {
		t.Fatal(err)
	}
	defer sub.Unsubscribe()

	var msgs []*nats.Msg
	for i := 0; i < ackPendingLimit; i++ {
		ms, err := sub.Fetch(1)
		if err != nil {
			t.Fatalf("Error on fetch: %v", err)
		}
		msgs = append(msgs, ms...)
	}

	// Since we don't ack, the next fetch should time out because the server
	// won't send new ones until we ack some.
	if _, err := sub.Fetch(1, nats.MaxWait(250*time.Millisecond)); err != nats.ErrTimeout {
		t.Fatalf("Expected timeout, got: %v", err)
	}
	// Ack one message, then we should be able to get the next
	msgs[0].Ack()
	if _, err := sub.Fetch(1); err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
}

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