Function TestPullSubscribeFetchWithHeartbeat

test/js_test.go:1229–1337

(t *testing.T)

}

func TestPullSubscribeFetchWithHeartbeat(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.Fetch(5, nats.PullHeartbeat(100*time.Millisecond))
	if err != nil {
		t.Fatalf("Unexpected error: %s", err)
	}
	if len(msgs) != 5 {
		t.Fatalf("Expected %d messages; got: %d", 5, len(msgs))
	}
	now := time.Now()
	// no messages available, should time out normally
	_, err = sub.Fetch(5, nats.PullHeartbeat(50*time.Millisecond), nats.MaxWait(300*time.Millisecond))
	elapsed := time.Since(now)
	if elapsed < 300*time.Millisecond {
		t.Fatalf("Expected timeout after 300ms; got: %v", elapsed)
	}
	if !errors.Is(err, nats.ErrTimeout) {
		t.Fatalf("Expected timeout error; got: %v", err)
	}

	// delete consumer to verify heartbeats are not sent anymore
	info, err := sub.ConsumerInfo()
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if err := js.DeleteConsumer("TEST", info.Name); err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	_, err = sub.Fetch(5, nats.PullHeartbeat(100*time.Millisecond), nats.MaxWait(1*time.Second))
	if !errors.Is(err, nats.ErrNoHeartbeat) {
		t.Fatalf("Expected no heartbeat error; got: %v", err)
	}