Repository: nats-io/nats.go
Pull request #1332 (2023-06-23 07:14:30)

[IMPROVED] Replace empty interface with any
This PR restores changes reverted for patch release in #1330 


Patch:
diff --git a/context.go b/context.go
index 8c3c454..c4ef4be 100644
--- a/context.go
+++ b/context.go
@@ -217,7 +217,7 @@ func (nc *Conn) FlushWithContext(ctx context.Context) error {
 // RequestWithContext will create an Inbox and perform a Request
 // using the provided cancellation context with the Inbox reply
 // for the data v. A response will be decoded into the vPtr last parameter.
-func (c *EncodedConn) RequestWithContext(ctx context.Context, subject string, v interface{}, vPtr interface{}) error {
+func (c *EncodedConn) RequestWithContext(ctx context.Context, subject string, v any, vPtr any) error {
 	if ctx == nil {
 		return ErrInvalidContext
 	}
diff --git a/enc.go b/enc.go
index 450fa76..a1c54f2 100644
--- a/enc.go
+++ b/enc.go
@@ -26,8 +26,8 @@ import (
 
 // Encoder interface is for all register encoders
 type Encoder interface {
-	Encode(subject string, v interface{}) ([]byte, error)
-	Decode(subject string, data []byte, vPtr interface{}) error
+	Encode(subject string, v any) ([]byte, error)
+	Decode(subject string, data []byte, vPtr any) error
 }
 
 var encMap map[string]Encoder
@@ -88,7 +88,7 @@ func EncoderForType(encType string) Encoder {
 
 // Publish publishes the data argument to the given subject. The data argument
 // will be encoded using the associated encoder.
-func (c *EncodedConn) Publish(subject string, v interface{}) error {
+func (c *EncodedConn) Publish(subject string, v any) error {
 	b, err := c.Enc.Encode(subject, v)
 	if err != nil {
 		return err
@@ -99,7 +99,7 @@ func (c *EncodedConn) Publish(subject string, v interface{}) error {
 // PublishRequest will perform a Publish() expecting a response on the
 // reply subject. Use Request() for automatically waiting for a response
 // inline.
-func (c *EncodedConn) PublishRequest(subject, reply string, v interface{}) error {
+func (c *EncodedConn) PublishRequest(subject, reply string, v any) error {
 	b, err := c.Enc.Encode(subject, v)
 	if err != nil {
 		return err
@@ -110,7 +110,7 @@ func (c *EncodedConn) PublishRequest(subject, reply string, v interface{}) error
 // Request will create an Inbox and perform a Request() call
 // with the Inbox reply for the data v. A response will be
 // decoded into the vPtr Response.
-func (c *EncodedConn) Request(subject string, v interface{}, vPtr interface{}, timeout time.Duration) error {
+func (c *EncodedConn) Request(subject string, v any, vPtr any, timeout time.Duration) error {
 	b, err := c.Enc.Encode(subject, v)
 	if err != nil {
 		return err
@@ -129,7 +129,7 @@ func (c *EncodedConn) Request(subject string, v interface{}, vPtr interface{}, t
 }
 
 // Handler is a specific callback used for Subscribe. It is generalized to
-// an interface{}, but we will discover its format and arguments at runtime
+// an any, but we will discover its format and arguments at runtime
 // and perform the correct callback, including demarshaling encoded data
 // back into the appropriate struct based on the signature of the Handler.
 //
@@ -150,7 +150,7 @@ func (c *EncodedConn) Request(subject string, v interface{}, vPtr interface{}, t
 // and demarshal it into the given struct, e.g. person.
 // There are also variants where the callback wants either the subject, or the
 // subject and the reply subject.
-type Handler interface{}
+type Handler any
 
 // Dissect the cb Handler's signature
 func argInfo(cb Handler) (reflect.Type, int) {
diff --git a/encoders/builtin/default_enc.go b/encoders/builtin/default_enc.go
index 46d918e..65c2d68 100644
--- a/encoders/builtin/default_enc.go
+++ b/encoders/builtin/default_enc.go
@@ -35,7 +35,7 @@ var falseB = []byte("false")
 var nilB = []byte("")
 
 // Encode
-func (je *DefaultEncoder) Encode(subject string, v interface{}) ([]byte, error) {
+func (je *DefaultEncoder) Encode(subject string, v any) ([]byte, error) {
 	switch arg := v.(type) {
 	case string:
 		bytes := *(*[]byte)(unsafe.Pointer(&arg))
@@ -58,7 +58,7 @@ func (je *DefaultEncoder) Encode(subject string, v interface{}) ([]byte, error)
 }
 
 // Decode
-func (je *DefaultEncoder) Decode(subject string, data []byte, vPtr interface{}) error {
+func (je *DefaultEncoder) Decode(subject string, data []byte, vPtr any) error {
 	// Figure out what it's pointing to...
 	sData := *(*string)(unsafe.Pointer(&data))
 	switch arg := vPtr.(type) {
diff --git a/encoders/builtin/gob_enc.go b/encoders/builtin/gob_enc.go
index 632bcbd..4e7cecb 100644
--- a/encoders/builtin/gob_enc.go
+++ b/encoders/builtin/gob_enc.go
@@ -28,7 +28,7 @@ type GobEncoder struct {
 // FIXME(dlc) - This could probably be more efficient.
 
 // Encode
-func (ge *GobEncoder) Encode(subject string, v interface{}) ([]byte, error) {
+func (ge *GobEncoder) Encode(subject string, v any) ([]byte, error) {
 	b := new(bytes.Buffer)
 	enc := gob.NewEncoder(b)
 	if err := enc.Encode(v); err != nil {
@@ -38,7 +38,7 @@ func (ge *GobEncoder) Encode(subject string, v interface{}) ([]byte, error) {
 }
 
 // Decode
-func (ge *GobEncoder) Decode(subject string, data []byte, vPtr interface{}) (err error) {
+func (ge *GobEncoder) Decode(subject string, data []byte, vPtr any) (err error) {
 	dec := gob.NewDecoder(bytes.NewBuffer(data))
 	err = dec.Decode(vPtr)
 	return
diff --git a/encoders/builtin/json_enc.go b/encoders/builtin/json_enc.go
index c9670f3..9b6ffc0 100644
--- a/encoders/builtin/json_enc.go
+++ b/encoders/builtin/json_enc.go
@@ -26,7 +26,7 @@ type JsonEncoder struct {
 }
 
 // Encode
-func (je *JsonEncoder) Encode(subject string, v interface{}) ([]byte, error) {
+func (je *JsonEncoder) Encode(subject string, v any) ([]byte, error) {
 	b, err := json.Marshal(v)
 	if err != nil {
 		return nil, err
@@ -35,7 +35,7 @@ func (je *JsonEncoder) Encode(subject string, v interface{}) ([]byte, error) {
 }
 
 // Decode
-func (je *JsonEncoder) Decode(subject string, data []byte, vPtr interface{}) (err error) {
+func (je *JsonEncoder) Decode(subject string, data []byte, vPtr any) (err error) {
 	switch arg := vPtr.(type) {
 	case *string:
 		// If they want a string and it is a JSON string, strip quotes
diff --git a/encoders/protobuf/protobuf_enc.go b/encoders/protobuf/protobuf_enc.go
index 8f0785c..9cb53c0 100644
--- a/encoders/protobuf/protobuf_enc.go
+++ b/encoders/protobuf/protobuf_enc.go
@@ -43,7 +43,7 @@ var (
 )
 
 // Encode
-func (pb *ProtobufEncoder) Encode(subject string, v interface{}) ([]byte, error) {
+func (pb *ProtobufEncoder) Encode(subject string, v any) ([]byte, error) {
 	if v == nil {
 		return nil, nil
 	}
@@ -60,8 +60,8 @@ func (pb *ProtobufEncoder) Encode(subject string, v interface{}) ([]byte, error)
 }
 
 // Decode
-func (pb *ProtobufEncoder) Decode(subject string, data []byte, vPtr interface{}) error {
-	if _, ok := vPtr.(*interface{}); ok {
+func (pb *ProtobufEncoder) Decode(subject string, data []byte, vPtr any) error {
+	if _, ok := vPtr.(*any); ok {
 		return nil
 	}
 	i, found := vPtr.(proto.Message)
diff --git a/jetstream/api.go b/jetstream/api.go
index 4fef877..0402587 100644
--- a/jetstream/api.go
+++ b/jetstream/api.go
@@ -103,7 +103,7 @@ const (
 	apiMsgDeleteT = "STREAM.MSG.DELETE.%s"
 )
 
-func (js *jetStream) apiRequestJSON(ctx context.Context, subject string, resp interface{}, data ...[]byte) (*jetStreamMsg, error) {
+func (js *jetStream) apiRequestJSON(ctx context.Context, subject string, resp any, data ...[]byte) (*jetStreamMsg, error) {
 	jsMsg, err := js.apiRequest(ctx, subject, data...)
 	if err != nil {
 		return nil, err
diff --git a/jetstream/consumer_config.go b/jetstream/consumer_config.go
index beeef09..41764ba 100644
--- a/jetstream/consumer_config.go
+++ b/jetstream/consumer_config.go
@@ -78,6 +78,7 @@ type (
 		OptStartTime      *time.Time    `json:"opt_start_time,omitempty"`
 		ReplayPolicy      ReplayPolicy  `json:"replay_policy"`
 		InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
+		HeadersOnly       bool          `json:"headers_only,omitempty"`
 
 		// Maximum number of attempts for the consumer to be recreated
 		// Defaults to unlimited
diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
index 0d4c530..36fad23 100644
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -473,6 +473,10 @@ func (js *jetStream) OrderedConsumer(ctx context.Context, stream string, cfg Ord
 	if cfg.OptStartSeq != 0 {
 		oc.cursor.streamSeq = cfg.OptStartSeq - 1
 	}
+	err := oc.reset()
+	if err != nil {
+		return nil, err
+	}
 
 	return oc, nil
 }
diff --git a/jetstream/ordered.go b/jetstream/ordered.go
index a386486..2f19674 100644
--- a/jetstream/ordered.go
+++ b/jetstream/ordered.go
@@ -67,7 +67,6 @@ var errOrderedSequenceMismatch = errors.New("sequence mismatch")
 // Consume can be used to continuously receive messages and handle them with the provided callback function
 func (c *orderedConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (ConsumeContext, error) {
 	if c.consumerType == consumerTypeNotSet || c.consumerType == consumerTypeConsume && c.currentConsumer == nil {
-		c.consumerType = consumerTypeConsume
 		err := c.reset()
 		if err != nil {
 			return nil, err
@@ -78,6 +77,7 @@ func (c *orderedConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt
 	if c.consumerType == consumerTypeFetch {
 		return nil, ErrOrderConsumerUsedAsFetch
 	}
+	c.consumerType = consumerTypeConsume
 	consumeOpts, err := parseConsumeOpts(opts...)
 	if err != nil {
 		return nil, fmt.Errorf("%w: %s", ErrInvalidOption, err)
@@ -156,7 +156,6 @@ func (c *orderedConsumer) errHandler(serial int) func(cc ConsumeContext, err err
 // Messages returns [MessagesContext], allowing continuously iterating over messages on a stream.
 func (c *orderedConsumer) Messages(opts ...PullMessagesOpt) (MessagesContext, error) {
 	if c.consumerType == consumerTypeNotSet || c.consumerType == consumerTypeConsume && c.currentConsumer == nil {
-		c.consumerType = consumerTypeConsume
 		err := c.reset()
 		if err != nil {
 			return nil, err
@@ -167,6 +166,7 @@ func (c *orderedConsumer) Messages(opts ...PullMessagesOpt) (MessagesContext, er
 	if c.consumerType == consumerTypeFetch {
 		return nil, ErrOrderConsumerUsedAsFetch
 	}
+	c.consumerType = consumerTypeConsume
 	consumeOpts, err := parseMessagesOpts(opts...)
 	if err != nil {
 		return nil, fmt.Errorf("%w: %s", ErrInvalidOption, err)
@@ -236,12 +236,15 @@ func (c *orderedConsumer) Fetch(batch int, opts ...FetchOpt) (MessageBatch, erro
 	if c.consumerType == consumerTypeConsume {
 		return nil, ErrOrderConsumerUsedAsConsume
 	}
+	c.currentConsumer.Lock()
 	if c.runningFetch != nil {
 		if !c.runningFetch.done {
+			c.currentConsumer.Unlock()
 			return nil, ErrOrderedConsumerConcurrentRequests
 		}
 		c.cursor.streamSeq = c.runningFetch.sseq
 	}
+	c.currentConsumer.Unlock()
 	c.consumerType = consumerTypeFetch
 	err := c.reset()
 	if err != nil {
@@ -388,6 +391,7 @@ func (c *orderedConsumer) getConsumerConfigForSeq(seq uint64) *ConsumerConfig {
 		AckPolicy:         AckNonePolicy,
 		InactiveThreshold: 5 * time.Minute,
 		Replicas:          1,
+		HeadersOnly:       c.cfg.HeadersOnly,
 	}
 	if len(c.cfg.FilterSubjects) == 1 {
 		cfg.FilterSubject = c.cfg.FilterSubjects[0]
diff --git a/jetstream/pull.go b/jetstream/pull.go
index aef0361..28ff24a 100644
--- a/jetstream/pull.go
+++ b/jetstream/pull.go
@@ -638,7 +638,9 @@ func (p *pullConsumer) fetch(req *pullRequest) (MessageBatch, error) {
 		defer close(res.msgs)
 		for {
 			if receivedMsgs == req.Batch || (req.MaxBytes != 0 && receivedBytes == req.MaxBytes) {
+				p.Lock()
 				res.done = true
+				p.Unlock()
 				return
 			}
 			select {
diff --git a/jetstream/stream.go b/jetstream/stream.go
index 6dba3ae..8e4ba63 100644
--- a/jetstream/stream.go
+++ b/jetstream/stream.go
@@ -208,6 +208,10 @@ func (s *stream) OrderedConsumer(ctx context.Context, cfg OrderedConsumerConfig)
 	if cfg.OptStartSeq != 0 {
 		oc.cursor.streamSeq = cfg.OptStartSeq - 1
 	}
+	err := oc.reset()
+	if err != nil {
+		return nil, err
+	}
 
 	return oc, nil
 }
diff --git a/js.go b/js.go
index 76a55d4..9ecf1f5 100644
--- a/js.go
+++ b/js.go
@@ -1352,7 +1352,7 @@ func processConsInfo(info *ConsumerInfo, userCfg *ConsumerConfig, isPullMode boo
 }
 
 func checkConfig(s, u *ConsumerConfig) error {
-	makeErr := func(fieldName string, usrVal, srvVal interface{}) error {
+	makeErr := func(fieldName string, usrVal, srvVal any) error {
 		return fmt.Errorf("configuration requests %s to be %v, but consumer's value is %v", fieldName, usrVal, srvVal)
 	}
 
diff --git a/micro/request.go b/micro/request.go
index ce24985..eb4c0b8 100644
--- a/micro/request.go
+++ b/micro/request.go
@@ -43,7 +43,7 @@ type (
 
 		// RespondJSON marshals the given response value and responds to the request.
 		// Additional headers can be passed using [WithHeaders] option.
-		RespondJSON(interface{}, ...RespondOpt) error
+		RespondJSON(any, ...RespondOpt) error
 
 		// Error prepares and publishes error response from a handler.
 		// A response error should be set containing an error code and description.
@@ -111,7 +111,7 @@ func (r *request) Respond(response []byte, opts ...RespondOpt) error {
 
 // RespondJSON marshals the given response value and responds to the request.
 // Additional headers can be passed using [WithHeaders] option.
-func (r *request) RespondJSON(response interface{}, opts ...RespondOpt) error {
+func (r *request) RespondJSON(response any, opts ...RespondOpt) error {
 	resp, err := json.Marshal(response)
 	if err != nil {
 		return ErrMarshalResponse
diff --git a/micro/service.go b/micro/service.go
index 6c01570..cee4ef3 100644
--- a/micro/service.go
+++ b/micro/service.go
@@ -85,7 +85,7 @@ type (
 
 	// StatsHandler is a function used to configure a custom STATS endpoint.
 	// It should return a value which can be serialized to JSON.
-	StatsHandler func(*Endpoint) interface{}
+	StatsHandler func(*Endpoint) any
 
 	// ServiceIdentity contains fields helping to identity a service instance.
 	ServiceIdentity struct {
diff --git a/netchan.go b/netchan.go
index a1af9e0..060721e 100644
--- a/netchan.go
+++ b/netchan.go
@@ -23,7 +23,7 @@ import (
 // Data will be encoded and decoded via the EncodedConn and its associated encoders.
 
 // BindSendChan binds a channel for send operations to NATS.
-func (c *EncodedConn) BindSendChan(subject string, channel interface{}) error {
+func (c *EncodedConn) BindSendChan(subject string, channel any) error {
 	chVal := reflect.ValueOf(channel)
 	if chVal.Kind() != reflect.Chan {
 		return ErrChanArg
@@ -61,17 +61,17 @@ func chPublish(c *EncodedConn, chVal reflect.Value, subject string) {
 }
 
 // BindRecvChan binds a channel for receive operations from NATS.
-func (c *EncodedConn) BindRecvChan(subject string, channel interface{}) (*Subscription, error) {
+func (c *EncodedConn) BindRecvChan(subject string, channel any) (*Subscription, error) {
 	return c.bindRecvChan(subject, _EMPTY_, channel)
 }
 
 // BindRecvQueueChan binds a channel for queue-based receive operations from NATS.
-func (c *EncodedConn) BindRecvQueueChan(subject, queue string, channel interface{}) (*Subscription, error) {
+func (c *EncodedConn) BindRecvQueueChan(subject, queue string, channel any) (*Subscription, error) {
 	return c.bindRecvChan(subject, queue, channel)
 }
 
 // Internal function to bind receive operations for a channel.
-func (c *EncodedConn) bindRecvChan(subject, queue string, channel interface{}) (*Subscription, error) {
+func (c *EncodedConn) bindRecvChan(subject, queue string, channel any) (*Subscription, error) {
 	chVal := reflect.ValueOf(channel)
 	if chVal.Kind() != reflect.Chan {
 		return nil, ErrChanArg


Test patch:
diff --git a/encoders/protobuf/testdata/pbtest.pb.go b/encoders/protobuf/testdata/pbtest.pb.go
index 2d7ee63..4f05224 100644
--- a/encoders/protobuf/testdata/pbtest.pb.go
+++ b/encoders/protobuf/testdata/pbtest.pb.go
@@ -130,7 +130,7 @@ func file_pbtest_proto_rawDescGZIP() []byte {
 }
 
 var file_pbtest_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
-var file_pbtest_proto_goTypes = []interface{}{
+var file_pbtest_proto_goTypes = []any{
 	(*Person)(nil), // 0: testdata.Person
 	nil,            // 1: testdata.Person.ChildrenEntry
 }
@@ -150,7 +150,7 @@ func file_pbtest_proto_init() {
 		return
 	}
 	if !protoimpl.UnsafeEnabled {
-		file_pbtest_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+		file_pbtest_proto_msgTypes[0].Exporter = func(v any, i int) any {
 			switch v := v.(*Person); i {
 			case 0:
 				return &v.state
diff --git a/internal/parser/parse_test.go b/internal/parser/parse_test.go
index 602e56e..a40a527 100644
--- a/internal/parser/parse_test.go
+++ b/internal/parser/parse_test.go
@@ -35,7 +35,7 @@ func TestParseNum(t *testing.T) {
 		},
 		{
 			name:     "parse 0",
-			given:    "",
+			given:    "0",
 			expected: 0,
 		},
 		{
diff --git a/jetstream/test/helper_test.go b/jetstream/test/helper_test.go
index c8c4d18..ce7909d 100644
--- a/jetstream/test/helper_test.go
+++ b/jetstream/test/helper_test.go
@@ -322,3 +322,19 @@ func restartBasicJSServer(t *testing.T, s *server.Server) *server.Server {
 	s.WaitForShutdown()
 	return RunServerWithOptions(opts)
 }
+
+func checkFor(t *testing.T, totalWait, sleepDur time.Duration, f func() error) {
+	t.Helper()
+	timeout := time.Now().Add(totalWait)
+	var err error
+	for time.Now().Before(timeout) {
+		err = f()
+		if err == nil {
+			return
+		}
+		time.Sleep(sleepDur)
+	}
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+}
diff --git a/jetstream/test/jetstream_test.go b/jetstream/test/jetstream_test.go
index d22c5ba..bf9a63c 100644
--- a/jetstream/test/jetstream_test.go
+++ b/jetstream/test/jetstream_test.go
@@ -18,6 +18,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"reflect"
 	"testing"
 	"time"
 
@@ -262,6 +263,153 @@ func TestCreateStream(t *testing.T) {
 	}
 }
 
+func TestCreateStreamMirrorCrossDomains(t *testing.T) {
+	test := []struct {
+		name         string
+		streamConfig *jetstream.StreamConfig
+	}{
+		{
+			name: "create stream mirror cross domains",
+			streamConfig: &jetstream.StreamConfig{
+				Name: "MIRROR",
+				Mirror: &jetstream.StreamSource{
+					Name:   "TEST",
+					Domain: "HUB",
+				},
+			},
+		},
+		{
+			name: "create stream with source cross domains",
+			streamConfig: &jetstream.StreamConfig{
+				Name: "MIRROR",
+				Sources: []*jetstream.StreamSource{
+					{
+						Name:   "TEST",
+						Domain: "HUB",
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range test {
+		t.Run(test.name, func(t *testing.T) {
+			conf := createConfFile(t, []byte(`
+		server_name: HUB
+		listen: 127.0.0.1:-1
+		jetstream: { domain: HUB }
+		leafnodes { listen: 127.0.0.1:7422 }
+	}`))
+			defer os.Remove(conf)
+			srv, _ := RunServerWithConfig(conf)
+			defer shutdownJSServerAndRemoveStorage(t, srv)
+
+			lconf := createConfFile(t, []byte(`
+	server_name: LEAF
+	listen: 127.0.0.1:-1
+	 jetstream: { domain:LEAF }
+	 leafnodes {
+		  remotes = [ { url: "leaf://127.0.0.1" } ]
+	 }
+}`))
+			defer os.Remove(lconf)
+			ln, _ := RunServerWithConfig(lconf)
+			defer shutdownJSServerAndRemoveStorage(t, ln)
+
+			nc, err := nats.Connect(srv.ClientURL())
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			defer nc.Close()
+
+			ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+			defer cancel()
+			js, err := jetstream.New(nc)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+
+			_, err = js.CreateStream(ctx, jetstream.StreamConfig{
+				Name:     "TEST",
+				Subjects: []string{"foo"},
+			})
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if _, err := js.Publish(ctx, "foo", []byte("msg1")); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if _, err := js.Publish(ctx, "foo", []byte("msg2")); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+
+			lnc, err := nats.Connect(ln.ClientURL())
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			defer lnc.Close()
+			ljs, err := jetstream.New(lnc)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+
+			ccfg := *test.streamConfig
+			_, err = ljs.CreateStream(ctx, ccfg)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if !reflect.DeepEqual(test.streamConfig, &ccfg) {
+				t.Fatalf("Did not expect config to be altered: %+v vs %+v", test.streamConfig, ccfg)
+			}
+
+			// Make sure we sync.
+			checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
+				lStream, err := ljs.Stream(ctx, "MIRROR")
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				if lStream.CachedInfo().State.Msgs == 2 {
+					return nil
+				}
+				return fmt.Errorf("Did not get synced messages: %d", lStream.CachedInfo().State.Msgs)
+			})
+			if _, err := ljs.Publish(ctx, "foo", []byte("msg3")); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			lStream, err := ljs.Stream(ctx, "MIRROR")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+
+			if lStream.CachedInfo().State.Msgs != 3 {
+				t.Fatalf("Expected 3 msgs in stream; got: %d", lStream.CachedInfo().State.Msgs)
+			}
+
+			rjs, err := jetstream.NewWithDomain(lnc, "HUB")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+
+			_, err = rjs.Stream(ctx, "TEST")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if _, err := rjs.Publish(ctx, "foo", []byte("msg4")); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			rStream, err := rjs.Stream(ctx, "TEST")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+
+			if rStream.CachedInfo().State.Msgs != 4 {
+				t.Fatalf("Expected 3 msgs in stream; got: %d", rStream.CachedInfo().State.Msgs)
+			}
+		})
+	}
+}
+
 func TestUpdateStream(t *testing.T) {
 	tests := []struct {
 		name      string
diff --git a/jetstream/test/ordered_test.go b/jetstream/test/ordered_test.go
index 0333b4a..bb35843 100644
--- a/jetstream/test/ordered_test.go
+++ b/jetstream/test/ordered_test.go
@@ -402,6 +402,281 @@ func TestOrderedConsumerFetch(t *testing.T) {
 			t.Fatalf("Expected error: %s; got: %s", jetstream.ErrOrderConsumerUsedAsConsume, err)
 		}
 	})
+
+	t.Run("concurrent fetch requests", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.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(), 20*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		publishTestMsgs(t, nc)
+		res, err := c.Fetch(1, jetstream.FetchMaxWait(100*time.Millisecond))
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		_, err = c.Fetch(1)
+		if !errors.Is(err, jetstream.ErrOrderedConsumerConcurrentRequests) {
+			t.Fatalf("Expected error: %s; got: %s", jetstream.ErrOrderedConsumerConcurrentRequests, err)
+		}
+		for msg := range res.Messages() {
+			msg.Ack()
+		}
+	})
+}
+
+func TestOrderedConsumerFetchBytes(t *testing.T) {
+	testSubject := "FOO.123"
+	testMsgs := []string{"m1", "m2", "m3", "m4", "m5"}
+	publishTestMsgs := func(t *testing.T, nc *nats.Conn) {
+		for _, msg := range testMsgs {
+			if err := nc.Publish(testSubject, []byte(msg)); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+	}
+	t.Run("base usage, delete consumer", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.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(), 20*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		msgs := make([]jetstream.Msg, 0)
+
+		publishTestMsgs(t, nc)
+		res, err := c.FetchBytes(500, jetstream.FetchMaxWait(100*time.Millisecond))
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+
+		for msg := range res.Messages() {
+			msgs = append(msgs, msg)
+		}
+		if res.Error() != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		name := c.CachedInfo().Name
+		if err := s.DeleteConsumer(ctx, name); err != nil {
+			t.Fatal(err)
+		}
+		publishTestMsgs(t, nc)
+		res, err = c.Fetch(500, jetstream.FetchMaxWait(100*time.Millisecond))
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+
+		for msg := range res.Messages() {
+			msgs = append(msgs, msg)
+		}
+		if res.Error() != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		if len(msgs) != 2*len(testMsgs) {
+			t.Fatalf("Expected %d messages; got: %d", 2*len(testMsgs), len(msgs))
+		}
+	})
+
+	t.Run("consumer used as consume", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.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(), 20*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		cc, err := c.Consume(func(msg jetstream.Msg) {})
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		cc.Stop()
+
+		_, err = c.FetchBytes(500)
+		if !errors.Is(err, jetstream.ErrOrderConsumerUsedAsConsume) {
+			t.Fatalf("Expected error: %s; got: %s", jetstream.ErrOrderConsumerUsedAsConsume, err)
+		}
+	})
+
+	t.Run("concurrent fetch requests", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.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(), 20*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		publishTestMsgs(t, nc)
+		res, err := c.FetchBytes(500, jetstream.FetchMaxWait(100*time.Millisecond))
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		_, err = c.FetchBytes(500)
+		if !errors.Is(err, jetstream.ErrOrderedConsumerConcurrentRequests) {
+			t.Fatalf("Expected error: %s; got: %s", jetstream.ErrOrderedConsumerConcurrentRequests, err)
+		}
+		for msg := range res.Messages() {
+			msg.Ack()
+		}
+	})
+}
+
+func TestOrderedConsumerNext(t *testing.T) {
+	testSubject := "FOO.123"
+	testMsgs := []string{"m1", "m2", "m3", "m4", "m5"}
+	publishTestMsgs := func(t *testing.T, nc *nats.Conn) {
+		for _, msg := range testMsgs {
+			if err := nc.Publish(testSubject, []byte(msg)); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+	}
+	t.Run("base usage, delete consumer", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.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(), 20*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		publishTestMsgs(t, nc)
+		msg, err := c.Next()
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		msg.Ack()
+
+		name := c.CachedInfo().Name
+		if err := s.DeleteConsumer(ctx, name); err != nil {
+			t.Fatal(err)
+		}
+		msg, err = c.Next()
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		msg.Ack()
+	})
+
+	t.Run("consumer used as consume", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.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(), 20*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		cc, err := c.Consume(func(msg jetstream.Msg) {})
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		cc.Stop()
+
+		_, err = c.Next()
+		if !errors.Is(err, jetstream.ErrOrderConsumerUsedAsConsume) {
+			t.Fatalf("Expected error: %s; got: %s", jetstream.ErrOrderConsumerUsedAsConsume, err)
+		}
+	})
 }
 
 func TestOrderedConsumerFetchNoWait(t *testing.T) {
@@ -531,18 +806,10 @@ func TestOrderedConsumerInfo(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+	c, err := js.OrderedConsumer(ctx, "foo", jetstream.OrderedConsumerConfig{})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	_, err = c.Info(ctx)
-	if !errors.Is(err, jetstream.ErrOrderedConsumerNotCreated) {
-		t.Fatalf("Expected error: %v; got: %v", jetstream.ErrOrderedConsumerNotCreated, err)
-	}
-	info := c.CachedInfo()
-	if info != nil {
-		t.Fatalf("Cached info should be nil if consumer is not yet created")
-	}
 
 	cc, err := c.Consume(func(msg jetstream.Msg) {})
 	if err != nil {
@@ -550,7 +817,7 @@ func TestOrderedConsumerInfo(t *testing.T) {
 	}
 	defer cc.Stop()
 
-	info, err = c.Info(ctx)
+	info, err := c.Info(ctx)
 	if err != nil {
 		t.Fatalf("Unexpected error: %s", err)
 	}
diff --git a/micro/test/service_test.go b/micro/test/service_test.go
index 6129671..5386ae2 100644
--- a/micro/test/service_test.go
+++ b/micro/test/service_test.go
@@ -795,7 +795,7 @@ func TestMonitoringHandlers(t *testing.T) {
 		name             string
 		subject          string
 		withError        bool
-		expectedResponse interface{}
+		expectedResponse any
 	}{
 		{
 			name:    "PING all",
@@ -934,7 +934,7 @@ func TestMonitoringHandlers(t *testing.T) {
 				t.Fatalf("Unexpected error: %v", err)
 			}
 
-			respMap := make(map[string]interface{})
+			respMap := make(map[string]any)
 			if err := json.Unmarshal(resp.Data, &respMap); err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
@@ -942,7 +942,7 @@ func TestMonitoringHandlers(t *testing.T) {
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
-			expectedRespMap := make(map[string]interface{})
+			expectedRespMap := make(map[string]any)
 			if err := json.Unmarshal(expectedResponseJSON, &expectedRespMap); err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
@@ -1018,7 +1018,7 @@ func TestServiceStats(t *testing.T) {
 	tests := []struct {
 		name          string
 		config        micro.Config
-		expectedStats map[string]interface{}
+		expectedStats map[string]any
 	}{
 		{
 			name: "stats handler",
@@ -1032,13 +1032,13 @@ func TestServiceStats(t *testing.T) {
 			config: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				StatsHandler: func(e *micro.Endpoint) interface{} {
-					return map[string]interface{}{
+				StatsHandler: func(e *micro.Endpoint) any {
+					return map[string]any{
 						"key": "val",
 					}
 				},
 			},
-			expectedStats: map[string]interface{}{
+			expectedStats: map[string]any{
 				"key": "val",
 			},
 		},
@@ -1140,7 +1140,7 @@ func TestServiceStats(t *testing.T) {
 			}
 
 			if test.expectedStats != nil {
-				var data map[string]interface{}
+				var data map[string]any
 				if err := json.Unmarshal(stats.Endpoints[0].Data, &data); err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
@@ -1160,7 +1160,7 @@ func TestRequestRespond(t *testing.T) {
 
 	tests := []struct {
 		name             string
-		respondData      interface{}
+		respondData      any
 		respondHeaders   micro.Headers
 		errDescription   string
 		errCode          string
diff --git a/nats_test.go b/nats_test.go
index d697055..5cee524 100644
--- a/nats_test.go
+++ b/nats_test.go
@@ -64,7 +64,7 @@ func WaitTime(ch chan bool, timeout time.Duration) error {
 	return errors.New("timeout")
 }
 
-func stackFatalf(t *testing.T, f string, args ...interface{}) {
+func stackFatalf(t *testing.T, f string, args ...any) {
 	lines := make([]string, 0, 32)
 	msg := fmt.Sprintf(f, args...)
 	lines = append(lines, msg)
diff --git a/test/helper_test.go b/test/helper_test.go
index 992e6d7..632af70 100644
--- a/test/helper_test.go
+++ b/test/helper_test.go
@@ -30,8 +30,8 @@ import (
 
 // So that we can pass tests and benchmarks...
 type tLogger interface {
-	Fatalf(format string, args ...interface{})
-	Errorf(format string, args ...interface{})
+	Fatalf(format string, args ...any)
+	Errorf(format string, args ...any)
 }
 
 // TestLogger
@@ -52,7 +52,7 @@ func WaitTime(ch chan bool, timeout time.Duration) error {
 	return errors.New("timeout")
 }
 
-func stackFatalf(t tLogger, f string, args ...interface{}) {
+func stackFatalf(t tLogger, f string, args ...any) {
 	lines := make([]string, 0, 32)
 	msg := fmt.Sprintf(f, args...)
 	lines = append(lines, msg)
diff --git a/test/js_test.go b/test/js_test.go
index 3ce30be..da89222 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -6490,7 +6490,7 @@ func testJetStream_ClusterMultipleFetchPullSubscribe(t *testing.T, subject strin
 		gotNoMessages bool
 		count         = 0
 	)
-	queues.Range(func(k, v interface{}) bool {
+	queues.Range(func(k, v any) bool {
 		msgs := v.([]*nats.Msg)
 		count += len(msgs)
 
diff --git a/test/json_test.go b/test/json_test.go
index c907184..ca9e1da 100644
--- a/test/json_test.go
+++ b/test/json_test.go
@@ -126,7 +126,7 @@ func TestEncBuiltinJsonMarshalNull(t *testing.T) {
 
 	var testValue *TestType
 
-	ec.Subscribe("json_null", func(i interface{}) {
+	ec.Subscribe("json_null", func(i any) {
 		if i != nil {
 			t.Fatalf("Received test of '%v', wanted 'nil'\n", i)
 		}
diff --git a/test/protobuf_test.go b/test/protobuf_test.go
index 963e741..5016483 100644
--- a/test/protobuf_test.go
+++ b/test/protobuf_test.go
@@ -76,7 +76,7 @@ func TestEncProtoNilRequest(t *testing.T) {
 	testPerson := &pb.Person{Name: "Anatolii", Age: 25, Address: "Ukraine, Nikolaev"}
 
 	//Subscribe with empty interface shouldn't failed on empty message
-	ec.Subscribe("nil_test", func(_, reply string, _ interface{}) {
+	ec.Subscribe("nil_test", func(_, reply string, _ any) {
 		ec.Publish(reply, testPerson)
 	})