Function TestPullSubscribeFetchBatchWithHeartbeat

test/js_test.go:1448–1587

(t *testing.T)

}

func TestPullSubscribeFetchBatchWithHeartbeat(t *testing.T) {
	t.Skip("Since v2.10.26 server sends no responders if the consumer is deleted, we need to figure out how else to test missing heartbeats")
	s := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, s)

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

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

	sub, err := js.PullSubscribe("foo", "")
	if err != nil {
		t.Fatalf("Unexpected error: %s", err)
	}
	defer sub.Unsubscribe()
	for i := 0; i < 5; i++ {
		if _, err := js.Publish("foo", []byte("msg")); err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}
	}

	// fetch 5 messages, should finish immediately
	msgs, err := sub.FetchBatch(5, nats.PullHeartbeat(100*time.Millisecond))
	if err != nil {
		t.Fatalf("Unexpected error: %s", err)
	}
	var i int
	for msg := range msgs.Messages() {
		i++
		msg.Ack()
	}
	if i != 5 {
		t.Fatalf("Expected %d messages; got: %d", 5, i)
	}
	if msgs.Error() != nil {
		t.Fatalf("Unexpected error: %s", msgs.Error())
	}
	now := time.Now()
	// no messages available, should time out normally
	msgs, err = sub.FetchBatch(5, nats.PullHeartbeat(50*time.Millisecond), nats.MaxWait(300*time.Millisecond))
	if err != nil {
		t.Fatalf("Unexpected error: %s", err)
	}
	i = 0
	for msg := range msgs.Messages() {
		i++
		msg.Ack()
	}
	elapsed := time.Since(now)
	if i != 0 {
		t.Fatalf("Expected %d messages; got: %d", 0, i)
	}