Function withJSClusterAndStream

jetstream/test/helper_test.go:249–290

(t *testing.T, clusterName string, size int, stream jetstream.StreamConfig, tfn func(t *testing.T, subject string, srvs ...*jsServer))

}

func withJSClusterAndStream(t *testing.T, clusterName string, size int, stream jetstream.StreamConfig, tfn func(t *testing.T, subject string, srvs ...*jsServer)) {
	t.Helper()

	withJSCluster(t, clusterName, size, func(t *testing.T, nodes ...*jsServer) {
		srvA := nodes[0]
		nc, err := nats.Connect(srvA.ClientURL())
		if err != nil {
			t.Error(err)
		}
		defer nc.Close()

		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		jsm, err := jetstream.New(nc)
		if err != nil {
			t.Fatal(err)
		}
	CreateStream:
		for {
			select {
			case <-ctx.Done():
				if err != nil {
					t.Fatalf("Unexpected error creating stream: %v", err)
				}
				t.Fatalf("Unable to create stream on cluster")
			case <-time.After(500 * time.Millisecond):
				_, err = jsm.AccountInfo(ctx)
				if err != nil {
					// Backoff for a bit until cluster and resources are ready.
					time.Sleep(500 * time.Millisecond)
				}
				_, err = jsm.CreateStream(ctx, stream)
				if err != nil {
					continue CreateStream
				}
				break CreateStream
			}
		}

		tfn(t, stream.Name, nodes...)
	})
}

func withJSCluster(t *testing.T, clusterName string, size int, tfn func(t *testing.T, srvs ...*jsServer)) {
	t.Helper()