Function TestPublishAsyncClearStall

jetstream/test/publish_test.go:1989–2038

(t *testing.T)

}

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

	nc, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	js, err := jetstream.New(nc,
		jetstream.WithPublishAsyncTimeout(500*time.Millisecond),
		jetstream.WithPublishAsyncMaxPending(100))
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()

	// use stream with no acks to test stalling
	_, err = js.CreateStream(context.Background(), jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}, NoAck: true})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	for range 100 {
		_, err := js.PublishAsync("FOO.A", []byte("hello"), jetstream.WithStallWait(1*time.Nanosecond))
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
	}
	// after publishing 100 messages, next one should fail with ErrTooManyStalledMsgs
	_, err = js.PublishAsync("FOO.A", []byte("hello"), jetstream.WithStallWait(50*time.Millisecond))
	if !errors.Is(err, jetstream.ErrTooManyStalledMsgs) {
		t.Fatalf("Expected error: %v; got: %v", jetstream.ErrTooManyStalledMsgs, err)
	}

	// after publish timeout all pending messages should be cleared
	// and we should be able to publish again
	select {
	case <-js.PublishAsyncComplete():
	case <-time.After(2 * time.Second):
		t.Fatalf("Did not receive completion signal")
	}

	if _, err = js.PublishAsync("FOO.A", []byte("hello")); err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if js.PublishAsyncPending() != 1 {
		t.Fatalf("Expected 1 pending message; got: %d", js.PublishAsyncPending())
	}
}

func TestPublishWithScheduleAt(t *testing.T) {
	srv := RunBasicJetStreamServer()