Repository: nats-io/nats.go
Pull request #1312 (2023-06-13 07:45:53)

[CHANGED] Simplified API for listing streams and stream names, added option to filter by stream name
- `ListStreams` and `StreamNames` now allow for iterating over returned channel (channel is always closed)
- `ListStreams` and `StreamNames` now both accept `WithStreamListSubject` for filtering out returned streams


Patch:
diff --git a/jetstream/README.md b/jetstream/README.md
index 9cd4366..ad09cf2 100644
--- a/jetstream/README.md
+++ b/jetstream/README.md
@@ -165,28 +165,19 @@ js.DeleteStream(ctx, "ORDERS")
 ```go
 // list streams
 streams := js.ListStreams(ctx)
-var err error
-for err == nil {
-    select {
-    case s := <-streams.Info():
-        fmt.Println(s.Config.Name)
-    case err = <-streams.Err():
-    }
+for s := range streams.Info() {
+    fmt.Println(s.Config.Name)
 }
-if err != nil && !errors.Is(err, jetstream.ErrEndOfData) {
+if streams.Err() != nil {
     fmt.Println("Unexpected error ocurred")
 }
 
 // list stream names
 names := js.StreamNames(ctx)
-for err == nil {
-    select {
-    case name := <-names.Name():
-        fmt.Println(name)
-    case err = <-names.Err():
-    }
+for name := range names.Name() {
+    fmt.Println(name)
 }
-if err != nil && !errors.Is(err, jetstream.ErrEndOfData) {
+if names.Err() != nil {
     fmt.Println("Unexpected error ocurred")
 }
 ```
@@ -202,13 +193,13 @@ Using `Stream` interface, it is also possible to:
 _ = s.Purge(ctx)
 
 // remove all messages from a stream that are stored on a specific subject
-_ = s.Purge(ctx, jetstream.WithSubject("ORDERS.new"))
+_ = s.Purge(ctx, jetstream.WithPurgeSubject("ORDERS.new"))
 
 // remove all messages up to specified sequence number
-_ = s.Purge(ctx, jetstream.WithSequence(100))
+_ = s.Purge(ctx, jetstream.WithPurgeSequence(100))
 
 // remove messages, but keep 10 newest
-_ = s.Purge(ctx, jetstream.WithKeep(10))
+_ = s.Purge(ctx, jetstream.WithPurgeKeep(10))
 ```
 
 - Get and messages from stream
@@ -310,29 +301,20 @@ fmt.Println(cachedInfo.Config.Durable)
 ```go
 // list consumers
 consumers := s.ListConsumers(ctx)
-var err error
-for err != nil {
-    select {
-    case s := <-consumers.Info():
-        fmt.Println(s.Name)
-    case err = <-consumers.Err():
-    }
+for cons := range consumers.Info() {
+    fmt.Println(cons.Name)
 }
-if err != nil && !errors.Is(err, jetstream.ErrEndOfData) {
-    fmt.Println("Unexpected error occured")
+if consumers.Err() != nil {
+    fmt.Println("Unexpected error ocurred")
 }
 
 // list consumer names
 names := s.ConsumerNames(ctx)
-for err != nil {
-    select {
-    case name := <-names.Name():
-        fmt.Println(name)
-    case err = <-names.Err():
-    }
+for name := range names.Name() {
+    fmt.Println(name)
 }
-if err != nil && !errors.Is(err, jetstream.ErrEndOfData) {
-    fmt.Println("Unexpected error occured")
+if names.Err() != nil {
+    fmt.Println("Unexpected error ocurred")
 }
 ```
 
diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
index 6d30bb4..cf1d211 100644
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -75,9 +75,9 @@ type (
 		// DeleteStream removes a stream with given name
 		DeleteStream(ctx context.Context, stream string) error
 		// ListStreams returns StreamInfoLister enabling iterating over a channel of stream infos
-		ListStreams(context.Context) StreamInfoLister
+		ListStreams(context.Context, ...StreamListOpt) StreamInfoLister
 		// StreamNames returns a  StreamNameLister enabling iterating over a channel of stream names
-		StreamNames(context.Context) StreamNameLister
+		StreamNames(context.Context, ...StreamListOpt) StreamNameLister
 	}
 
 	StreamConsumerManager interface {
@@ -96,6 +96,8 @@ type (
 		DeleteConsumer(ctx context.Context, stream string, consumer string) error
 	}
 
+	StreamListOpt func(*streamsRequest) error
+
 	// AccountInfo contains info about the JetStream usage from the current account.
 	AccountInfo struct {
 		Memory    uint64        `json:"memory"`
@@ -158,12 +160,12 @@ type (
 
 	StreamInfoLister interface {
 		Info() <-chan *StreamInfo
-		Err() <-chan error
+		Err() error
 	}
 
 	StreamNameLister interface {
 		Name() <-chan string
-		Err() <-chan error
+		Err() error
 	}
 
 	apiPagedRequest struct {
@@ -176,7 +178,7 @@ type (
 
 		streams chan *StreamInfo
 		names   chan string
-		errs    chan error
+		err     error
 	}
 
 	streamListResponse struct {
@@ -192,6 +194,7 @@ type (
 	}
 
 	streamsRequest struct {
+		apiPagedRequest
 		Subject string `json:"subject,omitempty"`
 	}
 )
@@ -565,33 +568,43 @@ func (js *jetStream) AccountInfo(ctx context.Context) (*AccountInfo, error) {
 }
 
 // ListStreams returns StreamInfoLister enabling iterating over a channel of stream infos
-func (js *jetStream) ListStreams(ctx context.Context) StreamInfoLister {
+//
+// Available options:
+// [WithStreamListSubject] - allows filtering returned streams by provided subject
+func (js *jetStream) ListStreams(ctx context.Context, opts ...StreamListOpt) StreamInfoLister {
 	l := &streamLister{
 		js:      js,
 		streams: make(chan *StreamInfo),
-		errs:    make(chan error, 1),
+	}
+	var streamsReq streamsRequest
+	for _, opt := range opts {
+		if err := opt(&streamsReq); err != nil {
+			l.err = err
+			close(l.streams)
+			return l
+		}
 	}
 	go func() {
+		defer close(l.streams)
 		ctx, cancel := wrapContextWithoutDeadline(ctx)
 		if cancel != nil {
 			defer cancel()
 		}
 		for {
-			page, err := l.streamInfos(ctx)
+			page, err := l.streamInfos(ctx, streamsReq)
 			if err != nil && !errors.Is(err, ErrEndOfData) {
-				l.errs <- err
+				l.err = err
 				return
 			}
 			for _, info := range page {
 				select {
 				case l.streams <- info:
 				case <-ctx.Done():
-					l.errs <- ctx.Err()
+					l.err = ctx.Err()
 					return
 				}
 			}
 			if errors.Is(err, ErrEndOfData) {
-				l.errs <- err
 				return
 			}
 		}
@@ -606,38 +619,48 @@ func (s *streamLister) Info() <-chan *StreamInfo {
 }
 
 // Err returns an error channel which will be populated with error from [ListStreams] or [StreamNames] request
-func (s *streamLister) Err() <-chan error {
-	return s.errs
+func (s *streamLister) Err() error {
+	return s.err
 }
 
 // StreamNames returns a [StreamNameLister] enabling iterating over a channel of stream names
-func (js *jetStream) StreamNames(ctx context.Context) StreamNameLister {
+//
+// Available options:
+// [WithStreamListSubject] - allows filtering returned streams by provided subject
+func (js *jetStream) StreamNames(ctx context.Context, opts ...StreamListOpt) StreamNameLister {
 	l := &streamLister{
 		js:    js,
 		names: make(chan string),
-		errs:  make(chan error, 1),
+	}
+	var streamsReq streamsRequest
+	for _, opt := range opts {
+		if err := opt(&streamsReq); err != nil {
+			l.err = err
+			close(l.streams)
+			return l
+		}
 	}
 	go func() {
 		ctx, cancel := wrapContextWithoutDeadline(ctx)
 		if cancel != nil {
 			defer cancel()
 		}
+		defer close(l.names)
 		for {
-			page, err := l.streamNames(ctx)
+			page, err := l.streamNames(ctx, streamsReq)
 			if err != nil && !errors.Is(err, ErrEndOfData) {
-				l.errs <- err
+				l.err = err
 				return
 			}
 			for _, info := range page {
 				select {
 				case l.names <- info:
 				case <-ctx.Done():
-					l.errs <- ctx.Err()
+					l.err = ctx.Err()
 					return
 				}
 			}
 			if errors.Is(err, ErrEndOfData) {
-				l.errs <- err
 				return
 			}
 		}
@@ -682,21 +705,28 @@ func (s *streamLister) Name() <-chan string {
 }
 
 // infos fetches the next [StreamInfo] page
-func (s *streamLister) streamInfos(ctx context.Context) ([]*StreamInfo, error) {
+func (s *streamLister) streamInfos(ctx context.Context, streamsReq streamsRequest) ([]*StreamInfo, error) {
 	if s.pageInfo != nil && s.offset >= s.pageInfo.Total {
 		return nil, ErrEndOfData
 	}
 
-	req, err := json.Marshal(
-		apiPagedRequest{Offset: s.offset},
-	)
+	req := streamsRequest{
+		apiPagedRequest: apiPagedRequest{
+			Offset: s.offset,
+		},
+		Subject: streamsReq.Subject,
+	}
+	reqJSON, err := json.Marshal(req)
+	if err != nil {
+		return nil, err
+	}
 	if err != nil {
 		return nil, err
 	}
 
 	slSubj := apiSubj(s.js.apiPrefix, apiStreamListT)
 	var resp streamListResponse
-	_, err = s.js.apiRequestJSON(ctx, slSubj, &resp, req)
+	_, err = s.js.apiRequestJSON(ctx, slSubj, &resp, reqJSON)
 	if err != nil {
 		return nil, err
 	}
@@ -710,21 +740,25 @@ func (s *streamLister) streamInfos(ctx context.Context) ([]*StreamInfo, error) {
 }
 
 // streamNames fetches the next stream names page
-func (s *streamLister) streamNames(ctx context.Context) ([]string, error) {
+func (s *streamLister) streamNames(ctx context.Context, streamsReq streamsRequest) ([]string, error) {
 	if s.pageInfo != nil && s.offset >= s.pageInfo.Total {
 		return nil, ErrEndOfData
 	}
 
-	req, err := json.Marshal(
-		apiPagedRequest{Offset: s.offset},
-	)
+	req := streamsRequest{
+		apiPagedRequest: apiPagedRequest{
+			Offset: s.offset,
+		},
+		Subject: streamsReq.Subject,
+	}
+	reqJSON, err := json.Marshal(req)
 	if err != nil {
 		return nil, err
 	}
 
 	slSubj := apiSubj(s.js.apiPrefix, apiStreams)
 	var resp streamNamesResponse
-	_, err = s.js.apiRequestJSON(ctx, slSubj, &resp, req)
+	_, err = s.js.apiRequestJSON(ctx, slSubj, &resp, reqJSON)
 	if err != nil {
 		return nil, err
 	}
diff --git a/jetstream/options.go b/jetstream/options.go
index 3c97425..ce79fc5 100644
--- a/jetstream/options.go
+++ b/jetstream/options.go
@@ -241,6 +241,15 @@ func WithSubjectFilter(subject string) StreamInfoOpt {
 	}
 }
 
+// WithStreamListSubject can be used to filter results of ListStreams and StreamNames requests
+// to only streams that have given subject in their configuration
+func WithStreamListSubject(subject string) StreamListOpt {
+	return func(req *streamsRequest) error {
+		req.Subject = subject
+		return nil
+	}
+}
+
 // WithMsgID sets the message ID used for deduplication.
 func WithMsgID(id string) PublishOpt {
 	return func(opts *pubOpts) error {
diff --git a/jetstream/stream.go b/jetstream/stream.go
index e3ea7e3..dfb5244 100644
--- a/jetstream/stream.go
+++ b/jetstream/stream.go
@@ -162,12 +162,12 @@ type (
 
 	ConsumerInfoLister interface {
 		Info() <-chan *ConsumerInfo
-		Err() <-chan error
+		Err() error
 	}
 
 	ConsumerNameLister interface {
 		Name() <-chan string
-		Err() <-chan error
+		Err() error
 	}
 
 	consumerLister struct {
@@ -177,7 +177,7 @@ type (
 
 		consumers chan *ConsumerInfo
 		names     chan string
-		errs      chan error
+		err       error
 	}
 
 	consumerListResponse struct {
@@ -479,14 +479,14 @@ func (s *stream) deleteMsg(ctx context.Context, req *msgDeleteRequest) error {
 	return nil
 }
 
-// ListConsumers returns ConsumerInfoLister enabling iterating over a channel of consumer infos
+// ListConsumers returns ConsumerInfoLister enabling iterating over a channel of consumer infos.
 func (s *stream) ListConsumers(ctx context.Context) ConsumerInfoLister {
 	l := &consumerLister{
 		js:        s.jetStream,
 		consumers: make(chan *ConsumerInfo),
-		errs:      make(chan error, 1),
 	}
 	go func() {
+		defer close(l.consumers)
 		ctx, cancel := wrapContextWithoutDeadline(ctx)
 		if cancel != nil {
 			defer cancel()
@@ -494,13 +494,13 @@ func (s *stream) ListConsumers(ctx context.Context) ConsumerInfoLister {
 		for {
 			page, err := l.consumerInfos(ctx, s.name)
 			if err != nil && !errors.Is(err, ErrEndOfData) {
-				l.errs <- err
+				l.err = err
 				return
 			}
 			for _, info := range page {
 				select {
 				case <-ctx.Done():
-					l.errs <- ctx.Err()
+					l.err = ctx.Err()
 					return
 				default:
 				}
@@ -509,7 +509,6 @@ func (s *stream) ListConsumers(ctx context.Context) ConsumerInfoLister {
 				}
 			}
 			if errors.Is(err, ErrEndOfData) {
-				l.errs <- err
 				return
 			}
 		}
@@ -522,8 +521,8 @@ func (s *consumerLister) Info() <-chan *ConsumerInfo {
 	return s.consumers
 }
 
-func (s *consumerLister) Err() <-chan error {
-	return s.errs
+func (s *consumerLister) Err() error {
+	return s.err
 }
 
 // ConsumerNames returns a ConsumerNameLister enabling iterating over a channel of consumer names
@@ -531,9 +530,9 @@ func (s *stream) ConsumerNames(ctx context.Context) ConsumerNameLister {
 	l := &consumerLister{
 		js:    s.jetStream,
 		names: make(chan string),
-		errs:  make(chan error, 1),
 	}
 	go func() {
+		defer close(l.names)
 		ctx, cancel := wrapContextWithoutDeadline(ctx)
 		if cancel != nil {
 			defer cancel()
@@ -541,19 +540,18 @@ func (s *stream) ConsumerNames(ctx context.Context) ConsumerNameLister {
 		for {
 			page, err := l.consumerNames(ctx, s.name)
 			if err != nil && !errors.Is(err, ErrEndOfData) {
-				l.errs <- err
+				l.err = err
 				return
 			}
 			for _, info := range page {
 				select {
 				case l.names <- info:
 				case <-ctx.Done():
-					l.errs <- ctx.Err()
+					l.err = ctx.Err()
 					return
 				}
 			}
 			if errors.Is(err, ErrEndOfData) {
-				l.errs <- err
 				return
 			}
 		}


Test patch:
diff --git a/jetstream/test/jetstream_test.go b/jetstream/test/jetstream_test.go
index 9742c89..c41312c 100644
--- a/jetstream/test/jetstream_test.go
+++ b/jetstream/test/jetstream_test.go
@@ -244,7 +244,7 @@ func TestCreateStream(t *testing.T) {
 			name:      "context timeout",
 			stream:    "foo",
 			subject:   "BAR.123",
-			timeout:   10 * time.Microsecond,
+			timeout:   1 * time.Microsecond,
 			withError: context.DeadlineExceeded,
 		},
 	}
@@ -475,7 +475,7 @@ func TestUpdateStream(t *testing.T) {
 			name:      "context timeout",
 			stream:    "foo",
 			subject:   "FOO.123",
-			timeout:   10 * time.Microsecond,
+			timeout:   1 * time.Microsecond,
 			withError: context.DeadlineExceeded,
 		},
 	}
@@ -563,7 +563,7 @@ func TestStream(t *testing.T) {
 		{
 			name:      "context timeout",
 			stream:    "foo",
-			timeout:   10 * time.Microsecond,
+			timeout:   1 * time.Microsecond,
 			withError: context.DeadlineExceeded,
 		},
 	}
@@ -780,26 +780,43 @@ func TestListStreams(t *testing.T) {
 		name       string
 		streamsNum int
 		timeout    time.Duration
+		subject    string
+		expected   int
 		withError  error
 	}{
 		{
 			name:       "list streams",
 			streamsNum: 260,
 			timeout:    10 * time.Second,
+			expected:   260,
 		},
 		{
 			name:       "with empty context",
 			streamsNum: 260,
+			expected:   260,
 		},
 		{
 			name:       "no stream available",
 			timeout:    10 * time.Second,
 			streamsNum: 0,
+			expected:   0,
+		},
+		{
+			name:       "list streams with subject filter",
+			subject:    "FOO.123",
+			streamsNum: 260,
+			expected:   1,
+		},
+		{
+			name:       "list streams with subject filter, no match",
+			subject:    "FOO.500",
+			streamsNum: 100,
+			expected:   0,
 		},
 		{
 			name:       "context timeout",
 			streamsNum: 260,
-			timeout:    10 * time.Microsecond,
+			timeout:    1 * time.Microsecond,
 			withError:  context.DeadlineExceeded,
 		},
 	}
@@ -830,30 +847,25 @@ func TestListStreams(t *testing.T) {
 					t.Fatalf("Unexpected error: %v", err)
 				}
 			}
-			streamsList := js.ListStreams(ctx)
+			opts := []jetstream.StreamListOpt{}
+			if test.subject != "" {
+				opts = append(opts, jetstream.WithStreamListSubject(test.subject))
+			}
+			streamsList := js.ListStreams(ctx, opts...)
 			streams := make([]*jetstream.StreamInfo, 0)
-		Loop:
-			for {
-				select {
-				case s := <-streamsList.Info():
-					streams = append(streams, s)
-					if test.withError != nil {
-						t.Fatalf("Expected error: %v; got none", test.withError)
-					}
-				case err := <-streamsList.Err():
-					if test.withError != nil {
-						if !errors.Is(err, test.withError) {
-							t.Fatalf("Expected error: %v; got: %v", test.withError, err)
-						}
-						return
-					}
-					if !errors.Is(err, jetstream.ErrEndOfData) {
-						t.Fatalf("Unexpected error: %v", err)
-					}
-					break Loop
+			for si := range streamsList.Info() {
+				streams = append(streams, si)
+			}
+			if test.withError != nil {
+				if !errors.Is(streamsList.Err(), test.withError) {
+					t.Fatalf("Expected error: %v; got: %v", test.withError, streamsList.Err())
 				}
+				return
 			}
-			if len(streams) != test.streamsNum {
+			if streamsList.Err() != nil {
+				t.Fatalf("Unexpected error: %s", streamsList.Err())
+			}
+			if len(streams) != test.expected {
 				t.Fatalf("Wrong number of streams; want: %d; got: %d", test.streamsNum, len(streams))
 			}
 		})
@@ -864,6 +876,8 @@ func TestStreamNames(t *testing.T) {
 	tests := []struct {
 		name       string
 		streamsNum int
+		subject    string
+		expected   int
 		timeout    time.Duration
 		withError  error
 	}{
@@ -871,20 +885,36 @@ func TestStreamNames(t *testing.T) {
 			name:       "list streams",
 			streamsNum: 500,
 			timeout:    10 * time.Second,
+			expected:   500,
 		},
 		{
 			name:       "with empty context",
 			streamsNum: 500,
+			expected:   500,
 		},
 		{
 			name:       "no stream available",
 			streamsNum: 0,
+			expected:   0,
+			timeout:    10 * time.Second,
+		},
+		{
+			name:       "list streams with subject filter",
+			subject:    "FOO.123",
+			streamsNum: 260,
+			expected:   1,
+		},
+		{
+			name:       "list streams with subject filter, no match",
+			subject:    "FOO.500",
+			streamsNum: 100,
+			expected:   0,
 			timeout:    10 * time.Second,
 		},
 		{
 			name:       "context timeout",
 			streamsNum: 500,
-			timeout:    10 * time.Microsecond,
+			timeout:    1 * time.Microsecond,
 			withError:  context.DeadlineExceeded,
 		},
 	}
@@ -915,30 +945,25 @@ func TestStreamNames(t *testing.T) {
 					t.Fatalf("Unexpected error: %v", err)
 				}
 			}
-			streamsList := js.StreamNames(ctx)
+			opts := []jetstream.StreamListOpt{}
+			if test.subject != "" {
+				opts = append(opts, jetstream.WithStreamListSubject(test.subject))
+			}
+			streamsList := js.StreamNames(ctx, opts...)
 			streams := make([]string, 0)
-		Loop:
-			for {
-				select {
-				case s := <-streamsList.Name():
-					streams = append(streams, s)
-					if test.withError != nil {
-						t.Fatalf("Expected error: %v; got none", test.withError)
-					}
-				case err := <-streamsList.Err():
-					if test.withError != nil {
-						if !errors.Is(err, test.withError) {
-							t.Fatalf("Expected error: %v; got: %v", test.withError, err)
-						}
-						return
-					}
-					if !errors.Is(err, jetstream.ErrEndOfData) {
-						t.Fatalf("Unexpected error: %v", err)
-					}
-					break Loop
+			for s := range streamsList.Name() {
+				streams = append(streams, s)
+			}
+			if test.withError != nil {
+				if !errors.Is(streamsList.Err(), test.withError) {
+					t.Fatalf("Expected error: %v; got: %v", test.withError, streamsList.Err())
 				}
+				return
+			}
+			if streamsList.Err() != nil {
+				t.Fatalf("Unexpected error: %s", streamsList.Err())
 			}
-			if len(streams) != test.streamsNum {
+			if len(streams) != test.expected {
 				t.Fatalf("Wrong number of streams; want: %d; got: %d", test.streamsNum, len(streams))
 			}
 		})
@@ -1005,7 +1030,7 @@ func TestJetStream_CreateOrUpdateConsumer(t *testing.T) {
 			name:           "context timeout",
 			consumerConfig: jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy},
 			stream:         "foo",
-			timeout:        10 * time.Microsecond,
+			timeout:        1 * time.Microsecond,
 			withError:      context.DeadlineExceeded,
 		},
 	}
@@ -1113,7 +1138,7 @@ func TestJetStream_Consumer(t *testing.T) {
 			name:      "context timeout",
 			stream:    "foo",
 			durable:   "dur",
-			timeout:   10 * time.Microsecond,
+			timeout:   1 * time.Microsecond,
 			withError: context.DeadlineExceeded,
 		},
 	}
@@ -1213,7 +1238,7 @@ func TestJetStream_DeleteConsumer(t *testing.T) {
 			name:      "context timeout",
 			stream:    "foo",
 			durable:   "dur",
-			timeout:   10 * time.Microsecond,
+			timeout:   1 * time.Microsecond,
 			withError: context.DeadlineExceeded,
 		},
 	}
@@ -1312,7 +1337,7 @@ func TestStreamNameBySubject(t *testing.T) {
 		{
 			name:      "context timeout",
 			subject:   "FOO.123",
-			timeout:   10 * time.Microsecond,
+			timeout:   1 * time.Microsecond,
 			withError: context.DeadlineExceeded,
 		},
 	}
diff --git a/jetstream/test/stream_test.go b/jetstream/test/stream_test.go
index 9b867f2..b90a79b 100644
--- a/jetstream/test/stream_test.go
+++ b/jetstream/test/stream_test.go
@@ -295,7 +295,7 @@ func TestStreamInfo(t *testing.T) {
 		},
 		{
 			name:      "context timeout",
-			timeout:   10 * time.Microsecond,
+			timeout:   1 * time.Microsecond,
 			withError: context.DeadlineExceeded,
 		},
 	}
@@ -486,7 +486,7 @@ func TestGetMsg(t *testing.T) {
 		{
 			name:      "context timeout",
 			seq:       1,
-			timeout:   10 * time.Microsecond,
+			timeout:   1 * time.Microsecond,
 			withError: context.DeadlineExceeded,
 		},
 	}
@@ -650,7 +650,7 @@ func TestGetLastMsgForSubject(t *testing.T) {
 		{
 			name:      "context timeout",
 			subject:   "*.A",
-			timeout:   10 * time.Microsecond,
+			timeout:   1 * time.Microsecond,
 			withError: context.DeadlineExceeded,
 		},
 	}
@@ -764,7 +764,7 @@ func TestDeleteMsg(t *testing.T) {
 		{
 			name:      "context timeout",
 			seq:       1,
-			timeout:   10 * time.Microsecond,
+			timeout:   1 * time.Microsecond,
 			withError: context.DeadlineExceeded,
 		},
 	}
@@ -929,7 +929,7 @@ func TestListConsumers(t *testing.T) {
 		{
 			name:         "context timeout",
 			consumersNum: 500,
-			timeout:      10 * time.Microsecond,
+			timeout:      1 * time.Microsecond,
 			withError:    context.DeadlineExceeded,
 		},
 	}
@@ -966,26 +966,17 @@ func TestListConsumers(t *testing.T) {
 			}
 			consumersList := s.ListConsumers(ctx)
 			consumers := make([]*jetstream.ConsumerInfo, 0)
-		Loop:
-			for {
-				select {
-				case s := <-consumersList.Info():
-					consumers = append(consumers, s)
-					if test.withError != nil {
-						t.Fatalf("Expected error: %v; got none", test.withError)
-					}
-				case err := <-consumersList.Err():
-					if test.withError != nil {
-						if !errors.Is(err, test.withError) {
-							t.Fatalf("Expected error: %v; got: %v", test.withError, err)
-						}
-						return
-					}
-					if !errors.Is(err, jetstream.ErrEndOfData) {
-						t.Fatalf("Unexpected error: %v", err)
-					}
-					break Loop
+			for s := range consumersList.Info() {
+				consumers = append(consumers, s)
+			}
+			if test.withError != nil {
+				if !errors.Is(consumersList.Err(), test.withError) {
+					t.Fatalf("Expected error: %v; got: %v", test.withError, consumersList.Err())
 				}
+				return
+			}
+			if consumersList.Err() != nil {
+				t.Fatalf("Unexpected error: %v", consumersList.Err())
 			}
 			if len(consumers) != test.consumersNum {
 				t.Fatalf("Wrong number of streams; want: %d; got: %d", test.consumersNum, len(consumers))
@@ -1018,7 +1009,7 @@ func TestConsumerNames(t *testing.T) {
 		{
 			name:         "context timeout",
 			consumersNum: 500,
-			timeout:      10 * time.Microsecond,
+			timeout:      1 * time.Microsecond,
 			withError:    context.DeadlineExceeded,
 		},
 	}
@@ -1055,26 +1046,17 @@ func TestConsumerNames(t *testing.T) {
 			}
 			consumersList := s.ConsumerNames(ctx)
 			consumers := make([]string, 0)
-		Loop:
-			for {
-				select {
-				case s := <-consumersList.Name():
-					consumers = append(consumers, s)
-					if test.withError != nil {
-						t.Fatalf("Expected error: %v; got none", test.withError)
-					}
-				case err := <-consumersList.Err():
-					if test.withError != nil {
-						if !errors.Is(err, test.withError) {
-							t.Fatalf("Expected error: %v; got: %v", test.withError, err)
-						}
-						return
-					}
-					if !errors.Is(err, jetstream.ErrEndOfData) {
-						t.Fatalf("Unexpected error: %v", err)
-					}
-					break Loop
+			for name := range consumersList.Name() {
+				consumers = append(consumers, name)
+			}
+			if test.withError != nil {
+				if !errors.Is(consumersList.Err(), test.withError) {
+					t.Fatalf("Expected error: %v; got: %v", test.withError, consumersList.Err())
 				}
+				return
+			}
+			if consumersList.Err() != nil {
+				t.Fatalf("Unexpected error: %v", consumersList.Err())
 			}
 			if len(consumers) != test.consumersNum {
 				t.Fatalf("Wrong number of streams; want: %d; got: %d", test.consumersNum, len(consumers))
@@ -1137,7 +1119,7 @@ func TestPurgeStream(t *testing.T) {
 		},
 		{
 			name:      "context timeout",
-			timeout:   10 * time.Microsecond,
+			timeout:   1 * time.Microsecond,
 			withError: context.DeadlineExceeded,
 		},
 	}