Function TestPublishAsyncResetPendingOnReconnect

jetstream/test/publish_test.go:1703–1770

(t *testing.T)

}

func TestPublishAsyncResetPendingOnReconnect(t *testing.T) {
	s := RunBasicJetStreamServer()

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

	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_, err = js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	errs := make(chan error, 1)
	done := make(chan struct{}, 1)
	acks := make(chan jetstream.PubAckFuture, 100)
	wg := sync.WaitGroup{}
	go func() {
		for i := 0; i < 100; i++ {
			if ack, err := js.PublishAsync("FOO.A", []byte("hello")); err != nil {
				errs <- err
				return
			} else {
				acks <- ack
			}
			wg.Add(1)
		}
		close(acks)
		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")
	}
	for ack := range acks {
		go func(paf jetstream.PubAckFuture) {
			select {
			case <-paf.Ok():
			case err := <-paf.Err():
				if !errors.Is(err, nats.ErrDisconnected) && !errors.Is(err, nats.ErrNoResponders) {
					errs <- fmt.Errorf("Expected error: %v or %v; got: %v", nats.ErrDisconnected, nats.ErrNoResponders, err)
				}
			case <-time.After(5 * time.Second):
				errs <- errors.New("Did not receive completion signal")
			}
			wg.Done()
		}(ack)
	}