Function TestPublishAsyncRetryInErrHandler

jetstream/test/publish_test.go:1860–1924

(t *testing.T)

}

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

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

	streamCreated := make(chan struct{})
	errCB := func(js jetstream.JetStream, m *nats.Msg, e error) {
		<-streamCreated
		_, err := js.PublishMsgAsync(m)
		if err != nil {
			t.Fatalf("Unexpected error when republishing: %v", err)
		}
	}

	js, err := jetstream.New(nc, jetstream.WithPublishAsyncErrHandler(errCB))
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	errs := make(chan error, 1)
	done := make(chan struct{}, 1)
	go func() {
		for i := 0; i < 10; i++ {
			if _, err := js.PublishAsync("FOO.A", []byte("hello"), jetstream.WithRetryAttempts(0)); err != nil {
				errs <- err
				return
			}
		}
		done <- struct{}{}
	}()
	select {
	case <-done:
	case err := <-errs:
		t.Fatalf("Unexpected error during publish: %v", err)
	case <-time.After(5 * time.Second):
		t.Fatalf("Did not receive completion signal")
	}
	stream, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	close(streamCreated)
	select {
	case <-js.PublishAsyncComplete():
	case <-time.After(5 * time.Second):
		t.Fatalf("Did not receive completion signal")
	}

	info, err := stream.Info(context.Background())
	if err != nil {