Function TestPublishAsyncRetry

jetstream/test/publish_test.go:1772–1858

(t *testing.T)

}

func TestPublishAsyncRetry(t *testing.T) {
	tests := []struct {
		name     string
		pubOpts  []jetstream.PublishOpt
		ackError error
	}{
		{
			name: "retry until stream is ready",
			pubOpts: []jetstream.PublishOpt{
				jetstream.WithRetryAttempts(10),
				jetstream.WithRetryWait(100 * time.Millisecond),
			},
		},
		{
			name: "fail after max retries",
			pubOpts: []jetstream.PublishOpt{
				jetstream.WithRetryAttempts(2),
				jetstream.WithRetryWait(50 * time.Millisecond),
			},
			ackError: jetstream.ErrNoStreamResponse,
		},
		{
			name: "retries disabled",
			pubOpts: []jetstream.PublishOpt{
				jetstream.WithRetryAttempts(0),
			},
			ackError: jetstream.ErrNoStreamResponse,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			s := RunBasicJetStreamServer()
			defer shutdownJSServerAndRemoveStorage(t, s)

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

			// set max pending to 1 so that we can test if retries don't cause stall
			js, err := jetstream.New(nc, jetstream.WithPublishAsyncMaxPending(1))
			if err != nil {
				t.Fatalf("Unexpected error: %v", err)
			}
			defer nc.Close()
			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()

			test.pubOpts = append(test.pubOpts, jetstream.WithStallWait(1*time.Nanosecond))
			ack, err := js.PublishAsync("foo", []byte("hello"), test.pubOpts...)
			if err != nil {
				t.Fatalf("Unexpected error: %v", err)
			}
			publishComplete := js.PublishAsyncComplete()
			errs := make(chan error, 1)
			go func() {
				// create stream with delay so that publish will receive no responders