Function TestPublishAsyncAckTimeout

jetstream/test/publish_test.go:1926–1987

(t *testing.T)

}

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

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

	errs := make(chan error, 1)
	js, err := jetstream.New(nc,
		jetstream.WithPublishAsyncTimeout(50*time.Millisecond),
		jetstream.WithPublishAsyncErrHandler(func(js jetstream.JetStream, m *nats.Msg, e error) {
			errs <- e
		}),
	)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()

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

	ack, err := js.PublishAsync("FOO.A", []byte("hello"))
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	select {
	case <-ack.Ok():
		t.Fatalf("Expected timeout")
	case err := <-ack.Err():
		if !errors.Is(err, jetstream.ErrAsyncPublishTimeout) {
			t.Fatalf("Expected error: %v; got: %v", jetstream.ErrAsyncPublishTimeout, err)
		}
	case <-time.After(time.Second):
		t.Fatalf("Did not receive ack timeout")
	}

	// check if error callback is called
	select {
	case err := <-errs:
		if !errors.Is(err, jetstream.ErrAsyncPublishTimeout) {
			t.Fatalf("Expected error: %v; got: %v", jetstream.ErrAsyncPublishTimeout, err)
		}
	case <-time.After(time.Second):
		t.Fatalf("Did not receive error from error handler")
	}

	if js.PublishAsyncPending() != 0 {
		t.Fatalf("Expected no pending messages")
	}

	select {
	case <-js.PublishAsyncComplete():