Repository: nats-io/nats.go (go)

Issue:
new jetstream interface does not allow an OrderedConsumer to be HeadersOnly. Why?
Sometimes headers is enough, why force OrderedConsumer to bring body when normal consumer can be HeadersOnly? I think it makes sense to add HeadersOnly to the OrderedConsumerConfig.

/T

Pull request:
[ADDED] HeadersOnly field on OrderedConsumerConfig
Resolves #1323 

Patch:
diff --git a/README.md b/README.md
index 02e3d4e..ad839cb 100644
--- a/README.md
+++ b/README.md
@@ -29,7 +29,7 @@ When using or transitioning to Go modules support:
 ```bash
 # Go client latest or explicit version
 go get github.com/nats-io/nats.go/@latest
-go get github.com/nats-io/nats.go/@v1.27.0
+go get github.com/nats-io/nats.go/@v1.27.1
 
 # For latest NATS Server, add /v2 at the end
 go get github.com/nats-io/nats-server/v2
diff --git a/context.go b/context.go
index c4ef4be..8c3c454 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 any, vPtr any) error {
+func (c *EncodedConn) RequestWithContext(ctx context.Context, subject string, v interface{}, vPtr interface{}) error {
 	if ctx == nil {
 		return ErrInvalidContext
 	}
diff --git a/enc.go b/enc.go
index a1c54f2..450fa76 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 any) ([]byte, error)
-	Decode(subject string, data []byte, vPtr any) error
+	Encode(subject string, v interface{}) ([]byte, error)
+	Decode(subject string, data []byte, vPtr interface{}) 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 any) error {
+func (c *EncodedConn) Publish(subject string, v interface{}) error {
 	b, err := c.Enc.Encode(subject, v)
 	if err != nil {
 		return err
@@ -99,7 +99,7 @@ func (c *EncodedConn) Publish(subject string, v any) 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 any) error {
+func (c *EncodedConn) PublishRequest(subject, reply string, v interface{}) 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 any) 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 any, vPtr any, timeout time.Duration) error {
+func (c *EncodedConn) Request(subject string, v interface{}, vPtr interface{}, 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 any, vPtr any, timeout time.Dura
 }
 
 // Handler is a specific callback used for Subscribe. It is generalized to
-// an any, but we will discover its format and arguments at runtime
+// an interface{}, 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 any, vPtr any, timeout time.Dura
 // 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 any
+type Handler interface{}
 
 // 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 65c2d68..46d918e 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 any) ([]byte, error) {
+func (je *DefaultEncoder) Encode(subject string, v interface{}) ([]byte, error) {
 	switch arg := v.(type) {
 	case string:
 		bytes := *(*[]byte)(unsafe.Pointer(&arg))
@@ -58,7 +58,7 @@ func (je *DefaultEncoder) Encode(subject string, v any) ([]byte, error) {
 }
 
 // Decode
-func (je *DefaultEncoder) Decode(subject string, data []byte, vPtr any) error {
+func (je *DefaultEncoder) Decode(subject string, data []byte, vPtr interface{}) 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 4e7cecb..632bcbd 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 any) ([]byte, error) {
+func (ge *GobEncoder) Encode(subject string, v interface{}) ([]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 any) ([]byte, error) {
 }
 
 // Decode
-func (ge *GobEncoder) Decode(subject string, data []byte, vPtr any) (err error) {
+func (ge *GobEncoder) Decode(subject string, data []byte, vPtr interface{}) (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 9b6ffc0..c9670f3 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 any) ([]byte, error) {
+func (je *JsonEncoder) Encode(subject string, v interface{}) ([]byte, error) {
 	b, err := json.Marshal(v)
 	if err != nil {
 		return nil, err
@@ -35,7 +35,7 @@ func (je *JsonEncoder) Encode(subject string, v any) ([]byte, error) {
 }
 
 // Decode
-func (je *JsonEncoder) Decode(subject string, data []byte, vPtr any) (err error) {
+func (je *JsonEncoder) Decode(subject string, data []byte, vPtr interface{}) (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 9cb53c0..8f0785c 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 any) ([]byte, error) {
+func (pb *ProtobufEncoder) Encode(subject string, v interface{}) ([]byte, error) {
 	if v == nil {
 		return nil, nil
 	}
@@ -60,8 +60,8 @@ func (pb *ProtobufEncoder) Encode(subject string, v any) ([]byte, error) {
 }
 
 // Decode
-func (pb *ProtobufEncoder) Decode(subject string, data []byte, vPtr any) error {
-	if _, ok := vPtr.(*any); ok {
+func (pb *ProtobufEncoder) Decode(subject string, data []byte, vPtr interface{}) error {
+	if _, ok := vPtr.(*interface{}); ok {
 		return nil
 	}
 	i, found := vPtr.(proto.Message)
diff --git a/jetstream/README.md b/jetstream/README.md
index 81d7b27..0e64aa2 100644
--- a/jetstream/README.md
+++ b/jetstream/README.md
@@ -381,7 +381,7 @@ if msgs.Error() != nil {
     // handle error
 }
 
-// receive up to 1024kB of data
+// receive up to 1024 B of data
 msgs, _ := c.FetchBytes(1024)
 for msg := range msgs.Messages() {
     fmt.Printf("Received a JetStream message: %s\n", string(msg.Data()))
diff --git a/jetstream/api.go b/jetstream/api.go
index 0402587..4fef877 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 any, data ...[]byte) (*jetStreamMsg, error) {
+func (js *jetStream) apiRequestJSON(ctx context.Context, subject string, resp interface{}, 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 37c6e12..41764ba 100644
--- a/jetstream/consumer_config.go
+++ b/jetstream/consumer_config.go
@@ -38,29 +38,26 @@ type (
 
 	// ConsumerConfig is the configuration of a JetStream consumer.
 	ConsumerConfig struct {
-		Name            string          `json:"name,omitempty"`
-		Durable         string          `json:"durable_name,omitempty"`
-		Description     string          `json:"description,omitempty"`
-		DeliverPolicy   DeliverPolicy   `json:"deliver_policy"`
-		OptStartSeq     uint64          `json:"opt_start_seq,omitempty"`
-		OptStartTime    *time.Time      `json:"opt_start_time,omitempty"`
-		AckPolicy       AckPolicy       `json:"ack_policy"`
-		AckWait         time.Duration   `json:"ack_wait,omitempty"`
-		MaxDeliver      int             `json:"max_deliver,omitempty"`
-		BackOff         []time.Duration `json:"backoff,omitempty"`
-		FilterSubjects  []string        `json:"filter_subjects,omitempty"`
-		FilterSubject   string          `json:"filter_subject,omitempty"`
-		ReplayPolicy    ReplayPolicy    `json:"replay_policy"`
-		RateLimit       uint64          `json:"rate_limit_bps,omitempty"` // Bits per sec
-		SampleFrequency string          `json:"sample_freq,omitempty"`
-		MaxWaiting      int             `json:"max_waiting,omitempty"`
-		MaxAckPending   int             `json:"max_ack_pending,omitempty"`
-		HeadersOnly     bool            `json:"headers_only,omitempty"`
-
-		// Pull based options.
-		MaxRequestBatch    int           `json:"max_batch,omitempty"`
-		MaxRequestExpires  time.Duration `json:"max_expires,omitempty"`
-		MaxRequestMaxBytes int           `json:"max_bytes,omitempty"`
+		Name               string          `json:"name,omitempty"`
+		Durable            string          `json:"durable_name,omitempty"`
+		Description        string          `json:"description,omitempty"`
+		DeliverPolicy      DeliverPolicy   `json:"deliver_policy"`
+		OptStartSeq        uint64          `json:"opt_start_seq,omitempty"`
+		OptStartTime       *time.Time      `json:"opt_start_time,omitempty"`
+		AckPolicy          AckPolicy       `json:"ack_policy"`
+		AckWait            time.Duration   `json:"ack_wait,omitempty"`
+		MaxDeliver         int             `json:"max_deliver,omitempty"`
+		BackOff            []time.Duration `json:"backoff,omitempty"`
+		FilterSubject      string          `json:"filter_subject,omitempty"`
+		ReplayPolicy       ReplayPolicy    `json:"replay_policy"`
+		RateLimit          uint64          `json:"rate_limit_bps,omitempty"` // Bits per sec
+		SampleFrequency    string          `json:"sample_freq,omitempty"`
+		MaxWaiting         int             `json:"max_waiting,omitempty"`
+		MaxAckPending      int             `json:"max_ack_pending,omitempty"`
+		HeadersOnly        bool            `json:"headers_only,omitempty"`
+		MaxRequestBatch    int             `json:"max_batch,omitempty"`
+		MaxRequestExpires  time.Duration   `json:"max_expires,omitempty"`
+		MaxRequestMaxBytes int             `json:"max_bytes,omitempty"`
 
 		// Inactivity threshold.
 		InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
@@ -69,6 +66,9 @@ type (
 		Replicas int `json:"num_replicas"`
 		// Force memory storage.
 		MemoryStorage bool `json:"mem_storage,omitempty"`
+
+		// NOTE: FilterSubjects requires nats-server v2.10.0+
+		FilterSubjects []string `json:"filter_subjects,omitempty"`
 	}
 
 	OrderedConsumerConfig struct {
@@ -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/ordered.go b/jetstream/ordered.go
index a386486..906780d 100644
--- a/jetstream/ordered.go
+++ b/jetstream/ordered.go
@@ -388,6 +388,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/js.go b/js.go
index 9ecf1f5..76a55d4 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 any) error {
+	makeErr := func(fieldName string, usrVal, srvVal interface{}) error {
 		return fmt.Errorf("configuration requests %s to be %v, but consumer's value is %v", fieldName, usrVal, srvVal)
 	}
 
diff --git a/jsm.go b/jsm.go
index a2b6841..c668469 100644
--- a/jsm.go
+++ b/jsm.go
@@ -361,20 +361,29 @@ func (js *js) upsertConsumer(stream, consumerName string, cfg *ConsumerConfig, o
 
 	var ccSubj string
 	if consumerName == _EMPTY_ {
-		// if consumer name is empty, use the legacy ephemeral endpoint
+		// if consumer name is empty (neither Durable nor Name is set), use the legacy ephemeral endpoint
 		ccSubj = fmt.Sprintf(apiLegacyConsumerCreateT, stream)
 	} else if err := checkConsumerName(consumerName); err != nil {
 		return nil, err
-	} else if !js.nc.serverMinVersion(2, 9, 0) || (cfg.Durable != "" && js.opts.featureFlags.useDurableConsumerCreate) {
-		// if server version is lower than 2.9.0 or user set the useDurableConsumerCreate flag, use the legacy DURABLE.CREATE endpoint
-		ccSubj = fmt.Sprintf(apiDurableCreateT, stream, consumerName)
-	} else {
-		// if above server version 2.9.0, use the endpoints with consumer name
-		if cfg.FilterSubject == _EMPTY_ || cfg.FilterSubject == ">" {
+	} else if js.nc.serverMinVersion(2, 9, 0) {
+		if cfg.Durable != "" && js.opts.featureFlags.useDurableConsumerCreate {
+			// if user set the useDurableConsumerCreate flag, use the legacy DURABLE.CREATE endpoint
+			ccSubj = fmt.Sprintf(apiDurableCreateT, stream, consumerName)
+		} else if cfg.FilterSubject == _EMPTY_ || cfg.FilterSubject == ">" {
+			// if filter subject is empty or ">", use the endpoint without filter subject
 			ccSubj = fmt.Sprintf(apiConsumerCreateT, stream, consumerName)
 		} else {
+			// if filter subject is not empty, use the endpoint with filter subject
 			ccSubj = fmt.Sprintf(apiConsumerCreateWithFilterSubjectT, stream, consumerName, cfg.FilterSubject)
 		}
+	} else {
+		if cfg.Durable != "" {
+			// if Durable is set, use the DURABLE.CREATE endpoint
+			ccSubj = fmt.Sprintf(apiDurableCreateT, stream, consumerName)
+		} else {
+			// if Durable is not set, use the legacy ephemeral endpoint
+			ccSubj = fmt.Sprintf(apiLegacyConsumerCreateT, stream)
+		}
 	}
 
 	resp, err := js.apiRequestWithContext(o.ctx, js.apiSubj(ccSubj), req)
diff --git a/micro/request.go b/micro/request.go
index eb4c0b8..ce24985 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(any, ...RespondOpt) error
+		RespondJSON(interface{}, ...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 any, opts ...RespondOpt) error {
+func (r *request) RespondJSON(response interface{}, opts ...RespondOpt) error {
 	resp, err := json.Marshal(response)
 	if err != nil {
 		return ErrMarshalResponse
diff --git a/micro/service.go b/micro/service.go
index cee4ef3..6c01570 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) any
+	StatsHandler func(*Endpoint) interface{}
 
 	// ServiceIdentity contains fields helping to identity a service instance.
 	ServiceIdentity struct {
diff --git a/nats.go b/nats.go
index 00ec369..94f551c 100644
--- a/nats.go
+++ b/nats.go
@@ -47,7 +47,7 @@ import (
 
 // Default Constants
 const (
-	Version                   = "1.27.0"
+	Version                   = "1.27.1"
 	DefaultURL                = "nats://127.0.0.1:4222"
 	DefaultPort               = 4222
 	DefaultMaxReconnect       = 60
diff --git a/netchan.go b/netchan.go
index 060721e..a1af9e0 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 any) error {
+func (c *EncodedConn) BindSendChan(subject string, channel interface{}) 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 any) (*Subscription, error) {
+func (c *EncodedConn) BindRecvChan(subject string, channel interface{}) (*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 any) (*Subscription, error) {
+func (c *EncodedConn) BindRecvQueueChan(subject, queue string, channel interface{}) (*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 any) (*Subscription, error) {
+func (c *EncodedConn) bindRecvChan(subject, queue string, channel interface{}) (*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 4f05224..2d7ee63 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 = []any{
+var file_pbtest_proto_goTypes = []interface{}{
 	(*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 any, i int) any {
+		file_pbtest_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*Person); i {
 			case 0:
 				return &v.state
diff --git a/micro/test/service_test.go b/micro/test/service_test.go
index 5386ae2..6129671 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 any
+		expectedResponse interface{}
 	}{
 		{
 			name:    "PING all",
@@ -934,7 +934,7 @@ func TestMonitoringHandlers(t *testing.T) {
 				t.Fatalf("Unexpected error: %v", err)
 			}
 
-			respMap := make(map[string]any)
+			respMap := make(map[string]interface{})
 			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]any)
+			expectedRespMap := make(map[string]interface{})
 			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]any
+		expectedStats map[string]interface{}
 	}{
 		{
 			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) any {
-					return map[string]any{
+				StatsHandler: func(e *micro.Endpoint) interface{} {
+					return map[string]interface{}{
 						"key": "val",
 					}
 				},
 			},
-			expectedStats: map[string]any{
+			expectedStats: map[string]interface{}{
 				"key": "val",
 			},
 		},
@@ -1140,7 +1140,7 @@ func TestServiceStats(t *testing.T) {
 			}
 
 			if test.expectedStats != nil {
-				var data map[string]any
+				var data map[string]interface{}
 				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      any
+		respondData      interface{}
 		respondHeaders   micro.Headers
 		errDescription   string
 		errCode          string
diff --git a/nats_test.go b/nats_test.go
index 5cee524..d697055 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 ...any) {
+func stackFatalf(t *testing.T, f string, args ...interface{}) {
 	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 632af70..992e6d7 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 ...any)
-	Errorf(format string, args ...any)
+	Fatalf(format string, args ...interface{})
+	Errorf(format string, args ...interface{})
 }
 
 // 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 ...any) {
+func stackFatalf(t tLogger, f string, args ...interface{}) {
 	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 da89222..3ce30be 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 any) bool {
+	queues.Range(func(k, v interface{}) bool {
 		msgs := v.([]*nats.Msg)
 		count += len(msgs)
 
diff --git a/test/json_test.go b/test/json_test.go
index ca9e1da..c907184 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 any) {
+	ec.Subscribe("json_null", func(i interface{}) {
 		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 5016483..963e741 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, _ any) {
+	ec.Subscribe("nil_test", func(_, reply string, _ interface{}) {
 		ec.Publish(reply, testPerson)
 	})