Repository: nats-io/nats.go
Pull request #1180 (2023-01-04 16:53:43)

[ADDED] Support for multi-endpoint services


Patch:
diff --git a/.travis.yml b/.travis.yml
index 6da4a9e..5381fb3 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,11 +1,11 @@
 language: go
 go:
+- 1.19.x
 - 1.18.x
-- 1.17.x
 go_import_path: github.com/nats-io/nats.go
 install:
 - go get -t ./...
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.18 ]]; then
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.19 ]]; then
     go install github.com/mattn/goveralls@latest;
     go install github.com/wadey/gocovmerge@latest;
     go install honnef.co/go/tools/cmd/staticcheck@latest;
@@ -14,12 +14,12 @@ install:
 before_script:
 - $(exit $(go fmt ./... | wc -l))
 - go vet -modfile=go_test.mod ./...
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.18 ]]; then
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.19 ]]; then
     find . -type f -name "*.go" | xargs misspell -error -locale US;
     GOFLAGS="-mod=mod -modfile=go_test.mod" staticcheck ./...;
   fi
 script:
 - go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.18 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off; fi
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.19 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off; fi
 after_success:
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.18 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.19 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi
diff --git a/go.mod b/go.mod
index 0f9e0b9..a89cd9a 100644
--- a/go.mod
+++ b/go.mod
@@ -1,8 +1,10 @@
 module github.com/nats-io/nats.go
 
-go 1.16
+go 1.18
 
 require (
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
-)
\ No newline at end of file
+)
+
+require golang.org/x/crypto v0.5.0 // indirect
diff --git a/go.sum b/go.sum
index 00fe58c..042fbae 100644
--- a/go.sum
+++ b/go.sum
@@ -2,10 +2,5 @@ github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
-golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b h1:wSOdpTq0/eI46Ez/LkDwIsAKA71YP2SRKBODiRWM0as=
-golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
-golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
-golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
\ No newline at end of file
+golang.org/x/crypto v0.5.0 h1:U/0M97KRkSFvyD/3FSmdP5W5swImpNgle/EHFhOsQPE=
+golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU=
diff --git a/jsm.go b/jsm.go
index d4dc4b5..89646dd 100644
--- a/jsm.go
+++ b/jsm.go
@@ -319,6 +319,8 @@ func (js *js) AddConsumer(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*C
 			sameConfig := checkConfig(&consInfo.Config, cfg)
 			if sameConfig != nil {
 				return nil, fmt.Errorf("%w: creating consumer %q on stream %q", ErrConsumerNameAlreadyInUse, consumerName, stream)
+			} else {
+				return consInfo, nil
 			}
 		}
 	}
diff --git a/micro/request.go b/micro/request.go
index dbc4ce0..2d1b056 100644
--- a/micro/request.go
+++ b/micro/request.go
@@ -1,4 +1,4 @@
-// Copyright 2022 The NATS Authors
+// Copyright 2022-2023 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
diff --git a/micro/service.go b/micro/service.go
index 7dd401f..ac257b6 100644
--- a/micro/service.go
+++ b/micro/service.go
@@ -1,4 +1,4 @@
-// Copyright 2022 The NATS Authors
+// Copyright 2022-2023 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -18,6 +18,7 @@ import (
 	"errors"
 	"fmt"
 	"regexp"
+	"strings"
 	"sync"
 	"time"
 
@@ -30,14 +31,23 @@ import (
 // This functionality is EXPERIMENTAL and may be changed in later releases.
 
 type (
+
+	// Service exposes methods to operate on a service instance.
 	Service interface {
+		// AddEndpoint registers endpoint with given name on a specific subject.
+		AddEndpoint(string, Handler, ...EndpointOpt) error
+
+		// AddGroup returns a Group interface, allowing for more complex endpoint topologies.
+		// A group can be used to register endpoints with given prefix.
+		AddGroup(string) Group
+
 		// Info returns the service info.
 		Info() Info
 
 		// Stats returns statistics for the service endpoint and all monitoring endpoints.
 		Stats() Stats
 
-		// Reset resets all statistics on a service instance.
+		// Reset resets all statistics (for all endpoints) on a service instance.
 		Reset()
 
 		// Stop drains the endpoint subscriptions and marks the service as stopped.
@@ -47,6 +57,25 @@ type (
 		Stopped() bool
 	}
 
+	// Group allows for grouping endpoints on a service.
+	//
+	// Endpoints created using AddEndpoint will be grouped under common prefix (group name)
+	// New groups can also be derived from a group using AddGroup.
+	Group interface {
+		// AddGroup creates a new group, prefixed by this group's prefix.
+		AddGroup(string) Group
+
+		// AddEndpoint registers new endpoints on a service.
+		// The endpoint's subject will be prefixed with the group prefix.
+		AddEndpoint(string, Handler, ...EndpointOpt) error
+	}
+
+	EndpointOpt func(*endpointOpts) error
+
+	endpointOpts struct {
+		subject string
+	}
+
 	// ErrHandler is a function used to configure a custom error handler for a service,
 	ErrHandler func(Service, *NATSError)
 
@@ -55,7 +84,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) interface{}
 
 	// ServiceIdentity contains fields helping to identity a service instance.
 	ServiceIdentity struct {
@@ -65,16 +94,23 @@ type (
 	}
 
 	// Stats is the type returned by STATS monitoring endpoint.
-	// It contains stats for a specific endpoint (either request handler or monitoring enpoints).
+	// It contains stats of all registered endpoints.
 	Stats struct {
 		ServiceIdentity
-		Type                  string          `json:"type"`
+		Type      string           `json:"type"`
+		Started   time.Time        `json:"started"`
+		Endpoints []*EndpointStats `json:"endpoints"`
+	}
+
+	// EndpointStats contains stats for a specific endpoint.
+	EndpointStats struct {
+		Name                  string          `json:"name"`
+		Subject               string          `json:"subject"`
 		NumRequests           int             `json:"num_requests"`
 		NumErrors             int             `json:"num_errors"`
 		LastError             string          `json:"last_error"`
 		ProcessingTime        time.Duration   `json:"processing_time"`
 		AverageProcessingTime time.Duration   `json:"average_processing_time"`
-		Started               time.Time       `json:"started"`
 		Data                  json.RawMessage `json:"data,omitempty"`
 	}
 
@@ -87,9 +123,9 @@ type (
 	// Info is the basic information about a service type.
 	Info struct {
 		ServiceIdentity
-		Type        string `json:"type"`
-		Description string `json:"description"`
-		Subject     string `json:"subject"`
+		Type        string   `json:"type"`
+		Description string   `json:"description"`
+		Subjects    []string `json:"subjects"`
 	}
 
 	// SchemaResp is the response value for SCHEMA requests.
@@ -106,10 +142,18 @@ type (
 		Response string `json:"response"`
 	}
 
-	// Endpoint is used to configure a subject and handler for a service.
+	// Endpoint manages a service endpoint.
 	Endpoint struct {
-		Subject string `json:"subject"`
-		Handler Handler
+		EndpointConfig
+		service *service
+
+		stats        EndpointStats
+		subscription *nats.Subscription
+	}
+
+	group struct {
+		service *service
+		prefix  string
 	}
 
 	// Verb represents a name of the monitoring service.
@@ -117,16 +161,39 @@ type (
 
 	// Config is a configuration of a service.
 	Config struct {
-		Name         string   `json:"name"`
-		Version      string   `json:"version"`
-		Description  string   `json:"description"`
-		Schema       Schema   `json:"schema"`
-		Endpoint     Endpoint `json:"endpoint"`
+		// Name represents the name of the service.
+		Name string `json:"name"`
+
+		// Endpoint is an optional endpoint configuration.
+		// More complex, multi-endpoint services can be configured using
+		// Service.AddGroup and Service.AddEndpoint methods.
+		Endpoint *EndpointConfig `json:"endpoint"`
+
+		// Version is a SemVer compatible version string.
+		Version string `json:"version"`
+
+		// Description of the service.
+		Description string `json:"description"`
+
+		// Service schema
+		Schema Schema `json:"schema"`
+
+		// StatsHandler is a user-defined custom function
+		// used to calculate additional service stats.
 		StatsHandler StatsHandler
-		DoneHandler  DoneHandler
+
+		// DoneHandler is invoked when all service subscription are stopped.
+		DoneHandler DoneHandler
+
+		// ErrorHandler is invoked on any nats-related service error.
 		ErrorHandler ErrHandler
 	}
 
+	EndpointConfig struct {
+		Subject string
+		Handler Handler
+	}
+
 	// NATSError represents an error returned by a NATS Subscription.
 	// It contains a subject on which the subscription failed, so that
 	// it can be linked with a specific service endpoint.
@@ -144,10 +211,10 @@ type (
 
 		m            sync.Mutex
 		id           string
-		reqSub       *nats.Subscription
+		endpoints    []*Endpoint
 		verbSubs     map[string]*nats.Subscription
-		stats        *Stats
-		conn         *nats.Conn
+		started      time.Time
+		nc           *nats.Conn
 		natsHandlers handlers
 		stopped      bool
 
@@ -195,8 +262,9 @@ const (
 
 var (
 	// this regular expression is suggested regexp for semver validation: https://semver.org/
-	semVerRegexp      = regexp.MustCompile(`^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$`)
-	serviceNameRegexp = regexp.MustCompile(`^[A-Za-z0-9\-_]+$`)
+	semVerRegexp  = regexp.MustCompile(`^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$`)
+	nameRegexp    = regexp.MustCompile(`^[A-Za-z0-9\-_]+$`)
+	subjectRegexp = regexp.MustCompile(`^[^ >]+[>]?$`)
 )
 
 // Common errors returned by the Service framework.
@@ -227,8 +295,8 @@ func (s Verb) String() string {
 }
 
 // AddService adds a microservice.
-// It will enable internal common services (PING, STATS, INFO and SCHEMA) as well as
-// the actual service handler on the subject provided in config.Endpoint
+// It will enable internal common services (PING, STATS, INFO and SCHEMA).
+// Request handlers have to be registered separately using Service.AddEndpoint.
 // A service name, version and Endpoint configuration are required to add a service.
 // AddService returns a [Service] interface, allowing service management.
 // Each service is assigned a unique ID.
@@ -240,42 +308,32 @@ func AddService(nc *nats.Conn, config Config) (Service, error) {
 	id := nuid.Next()
 	svc := &service{
 		Config: config,
-		conn:   nc,
+		nc:     nc,
 		id:     id,
 		asyncDispatcher: asyncCallbacksHandler{
 			cbQueue: make(chan func(), 100),
 		},
+		verbSubs:  make(map[string]*nats.Subscription),
+		endpoints: make([]*Endpoint, 0),
 	}
 	svcIdentity := ServiceIdentity{
 		Name:    config.Name,
 		ID:      id,
 		Version: config.Version,
 	}
-	svc.verbSubs = make(map[string]*nats.Subscription)
-	svc.stats = &Stats{
-		ServiceIdentity: svcIdentity,
-	}
 
 	svc.setupAsyncCallbacks()
 
 	go svc.asyncDispatcher.asyncCBDispatcher()
 
-	// Setup internal subscriptions.
-	var err error
-
-	svc.reqSub, err = nc.QueueSubscribe(config.Endpoint.Subject, QG, func(m *nats.Msg) {
-		svc.reqHandler(&request{msg: m})
-	})
-	if err != nil {
-		svc.asyncDispatcher.close()
-		return nil, err
-	}
-
-	ping := Ping{
-		ServiceIdentity: svcIdentity,
-		Type:            PingResponseType,
+	if config.Endpoint != nil {
+		if err := addEndpoint(svc, "default", config.Endpoint.Subject, config.Endpoint.Handler); err != nil {
+			svc.asyncDispatcher.close()
+			return nil, err
+		}
 	}
 
+	// Setup internal subscriptions.
 	infoHandler := func(req Request) {
 		response, _ := json.Marshal(svc.Info())
 		if err := req.Respond(response); err != nil {
@@ -285,6 +343,10 @@ func AddService(nc *nats.Conn, config Config) (Service, error) {
 		}
 	}
 
+	ping := Ping{
+		ServiceIdentity: svcIdentity,
+		Type:            PingResponseType,
+	}
 	pingHandler := func(req Request) {
 		response, _ := json.Marshal(ping)
 		if err := req.Respond(response); err != nil {
@@ -334,11 +396,65 @@ func AddService(nc *nats.Conn, config Config) (Service, error) {
 		svc.asyncDispatcher.close()
 		return nil, err
 	}
-	svc.stats.Started = time.Now().UTC()
 
+	svc.started = time.Now().UTC()
 	return svc, nil
 }
 
+func (s *service) AddEndpoint(name string, handler Handler, opts ...EndpointOpt) error {
+	var options endpointOpts
+	for _, opt := range opts {
+		if err := opt(&options); err != nil {
+			return err
+		}
+	}
+	subject := name
+	if options.subject != "" {
+		subject = options.subject
+	}
+	return addEndpoint(s, name, subject, handler)
+}
+
+func addEndpoint(s *service, name, subject string, handler Handler) error {
+	if !nameRegexp.MatchString(name) {
+		return fmt.Errorf("%w: invalid endpoint name", ErrConfigValidation)
+	}
+	if !subjectRegexp.MatchString(subject) {
+		return fmt.Errorf("%w: invalid endpoint subject", ErrConfigValidation)
+	}
+	endpoint := &Endpoint{
+		service: s,
+		EndpointConfig: EndpointConfig{
+			Subject: subject,
+			Handler: handler,
+		},
+	}
+	sub, err := s.nc.QueueSubscribe(
+		subject,
+		QG,
+		func(m *nats.Msg) {
+			s.reqHandler(endpoint, &request{msg: m})
+		},
+	)
+	if err != nil {
+		return err
+	}
+	endpoint.subscription = sub
+	s.endpoints = append(s.endpoints, endpoint)
+	endpoint.stats = EndpointStats{
+		Name:    name,
+		Subject: subject,
+	}
+	return nil
+}
+
+func (s *service) AddGroup(name string) Group {
+	return &group{
+		service: s,
+		prefix:  name,
+	}
+}
+
 // dispatch is responsible for calling any async callbacks
 func (ac *asyncCallbacksHandler) asyncCBDispatcher() {
 	for {
@@ -359,90 +475,108 @@ func (ac *asyncCallbacksHandler) close() {
 	close(ac.cbQueue)
 }
 
-func (s *Config) valid() error {
-	if !serviceNameRegexp.MatchString(s.Name) {
+func (c *Config) valid() error {
+	if !nameRegexp.MatchString(c.Name) {
 		return fmt.Errorf("%w: service name: name should not be empty and should consist of alphanumerical charactest, dashes and underscores", ErrConfigValidation)
 	}
-	if !semVerRegexp.MatchString(s.Version) {
+	if !semVerRegexp.MatchString(c.Version) {
 		return fmt.Errorf("%w: version: version should not be empty should match the SemVer format", ErrConfigValidation)
 	}
-	return s.Endpoint.valid()
-}
-
-func (e *Endpoint) valid() error {
-	if e.Subject == "" {
-		return fmt.Errorf("%w: endpoint: subject is required", ErrConfigValidation)
-	}
-	if e.Handler == nil {
-		return fmt.Errorf("%w: endpoint: handler is required", ErrConfigValidation)
-	}
 	return nil
 }
 
-func (svc *service) setupAsyncCallbacks() {
-	svc.m.Lock()
-	defer svc.m.Unlock()
-	svc.natsHandlers.closed = svc.conn.ClosedHandler()
-	if svc.natsHandlers.closed != nil {
-		svc.conn.SetClosedHandler(func(c *nats.Conn) {
-			svc.Stop()
-			svc.natsHandlers.closed(c)
+func (s *service) setupAsyncCallbacks() {
+	s.m.Lock()
+	defer s.m.Unlock()
+	s.natsHandlers.closed = s.nc.ClosedHandler()
+	if s.natsHandlers.closed != nil {
+		s.nc.SetClosedHandler(func(c *nats.Conn) {
+			s.Stop()
+			s.natsHandlers.closed(c)
 		})
 	} else {
-		svc.conn.SetClosedHandler(func(c *nats.Conn) {
-			svc.Stop()
+		s.nc.SetClosedHandler(func(c *nats.Conn) {
+			s.Stop()
 		})
 	}
 
-	svc.natsHandlers.asyncErr = svc.conn.ErrorHandler()
-	if svc.natsHandlers.asyncErr != nil {
-		svc.conn.SetErrorHandler(func(c *nats.Conn, s *nats.Subscription, err error) {
-			if !svc.matchSubscriptionSubject(s.Subject) {
-				svc.natsHandlers.asyncErr(c, s, err)
+	s.natsHandlers.asyncErr = s.nc.ErrorHandler()
+	if s.natsHandlers.asyncErr != nil {
+		s.nc.SetErrorHandler(func(c *nats.Conn, sub *nats.Subscription, err error) {
+			endpoint, match := s.matchSubscriptionSubject(sub.Subject)
+			if !match {
+				s.natsHandlers.asyncErr(c, sub, err)
+				return
 			}
-			if svc.Config.ErrorHandler != nil {
-				svc.Config.ErrorHandler(svc, &NATSError{
-					Subject:     s.Subject,
+			if s.Config.ErrorHandler != nil {
+				s.Config.ErrorHandler(s, &NATSError{
+					Subject:     sub.Subject,
 					Description: err.Error(),
 				})
 			}
-			svc.m.Lock()
-			svc.stats.NumErrors++
-			svc.stats.LastError = err.Error()
-			svc.m.Unlock()
-			svc.Stop()
-			svc.natsHandlers.asyncErr(c, s, err)
+			s.m.Lock()
+			if endpoint != nil {
+				endpoint.stats.NumErrors++
+				endpoint.stats.LastError = err.Error()
+			}
+			s.m.Unlock()
+			s.Stop()
+			s.natsHandlers.asyncErr(c, sub, err)
 		})
 	} else {
-		svc.conn.SetErrorHandler(func(c *nats.Conn, s *nats.Subscription, err error) {
-			if !svc.matchSubscriptionSubject(s.Subject) {
+		s.nc.SetErrorHandler(func(c *nats.Conn, sub *nats.Subscription, err error) {
+			endpoint, match := s.matchSubscriptionSubject(sub.Subject)
+			if !match {
 				return
 			}
-			if svc.Config.ErrorHandler != nil {
-				svc.Config.ErrorHandler(svc, &NATSError{
-					Subject:     s.Subject,
+			if s.Config.ErrorHandler != nil {
+				s.Config.ErrorHandler(s, &NATSError{
+					Subject:     sub.Subject,
 					Description: err.Error(),
 				})
 			}
-			svc.m.Lock()
-			svc.stats.NumErrors++
-			svc.stats.LastError = err.Error()
-			svc.m.Unlock()
-			svc.Stop()
+			s.m.Lock()
+			if endpoint != nil {
+				endpoint.stats.NumErrors++
+				endpoint.stats.LastError = err.Error()
+			}
+			s.m.Unlock()
+			s.Stop()
 		})
 	}
 }
 
-func (svc *service) matchSubscriptionSubject(subj string) bool {
-	if svc.reqSub.Subject == subj {
-		return true
-	}
-	for _, verbSub := range svc.verbSubs {
+func (s *service) matchSubscriptionSubject(subj string) (*Endpoint, bool) {
+	s.m.Lock()
+	defer s.m.Unlock()
+	for _, verbSub := range s.verbSubs {
 		if verbSub.Subject == subj {
+			return nil, true
+		}
+	}
+	for _, e := range s.endpoints {
+		if matchEndpointSubject(e.Subject, subj) {
+			return e, true
+		}
+	}
+	return nil, false
+}
+
+func matchEndpointSubject(endpointSubject, literalSubject string) bool {
+	subjectTokens := strings.Split(literalSubject, ".")
+	endpointTokens := strings.Split(endpointSubject, ".")
+	if len(endpointTokens) > len(subjectTokens) {
+		return false
+	}
+	for i, et := range endpointTokens {
+		if i == len(endpointTokens)-1 && et == ">" {
 			return true
 		}
+		if et != subjectTokens[i] && et != "*" {
+			return false
+		}
 	}
-	return false
+	return true
 }
 
 // verbHandlers generates control handlers for a specific verb.
@@ -480,18 +614,18 @@ func (s *service) addInternalHandler(nc *nats.Conn, verb Verb, kind, id, name st
 }
 
 // reqHandler invokes the service request handler and modifies service stats
-func (s *service) reqHandler(req *request) {
+func (s *service) reqHandler(endpoint *Endpoint, req *request) {
 	start := time.Now()
-	s.Endpoint.Handler.Handle(req)
+	endpoint.Handler.Handle(req)
 	s.m.Lock()
-	s.stats.NumRequests++
-	s.stats.ProcessingTime += time.Since(start)
-	avgProcessingTime := s.stats.ProcessingTime.Nanoseconds() / int64(s.stats.NumRequests)
-	s.stats.AverageProcessingTime = time.Duration(avgProcessingTime)
+	endpoint.stats.NumRequests++
+	endpoint.stats.ProcessingTime += time.Since(start)
+	avgProcessingTime := endpoint.stats.ProcessingTime.Nanoseconds() / int64(endpoint.stats.NumRequests)
+	endpoint.stats.AverageProcessingTime = time.Duration(avgProcessingTime)
 
 	if req.respondError != nil {
-		s.stats.NumErrors++
-		s.stats.LastError = req.respondError.Error()
+		endpoint.stats.NumErrors++
+		endpoint.stats.LastError = req.respondError.Error()
 	}
 	s.m.Unlock()
 }
@@ -499,15 +633,14 @@ func (s *service) reqHandler(req *request) {
 // Stop drains the endpoint subscriptions and marks the service as stopped.
 func (s *service) Stop() error {
 	s.m.Lock()
+	defer s.m.Unlock()
 	if s.stopped {
 		return nil
 	}
-	defer s.m.Unlock()
-	if s.reqSub != nil {
-		if err := s.reqSub.Drain(); err != nil {
-			return fmt.Errorf("draining subscription for request handler: %w", err)
+	for _, e := range s.endpoints {
+		if err := e.stop(); err != nil {
+			return err
 		}
-		s.reqSub = nil
 	}
 	var keys []string
 	for key, sub := range s.verbSubs {
@@ -519,7 +652,7 @@ func (s *service) Stop() error {
 	for _, key := range keys {
 		delete(s.verbSubs, key)
 	}
-	restoreAsyncHandlers(s.conn, s.natsHandlers)
+	restoreAsyncHandlers(s.nc, s.natsHandlers)
 	s.stopped = true
 	if s.DoneHandler != nil {
 		s.asyncDispatcher.push(func() { s.DoneHandler(s) })
@@ -535,6 +668,10 @@ func restoreAsyncHandlers(nc *nats.Conn, handlers handlers) {
 
 // Info returns information about the service
 func (s *service) Info() Info {
+	endpoints := make([]string, 0, len(s.endpoints))
+	for _, e := range s.endpoints {
+		endpoints = append(endpoints, e.Subject)
+	}
 	return Info{
 		ServiceIdentity: ServiceIdentity{
 			Name:    s.Config.Name,
@@ -543,7 +680,7 @@ func (s *service) Info() Info {
 		},
 		Type:        InfoResponseType,
 		Description: s.Config.Description,
-		Subject:     s.Config.Endpoint.Subject,
+		Subjects:    endpoints,
 	}
 }
 
@@ -551,32 +688,39 @@ func (s *service) Info() Info {
 func (s *service) Stats() Stats {
 	s.m.Lock()
 	defer s.m.Unlock()
-	if s.StatsHandler != nil {
-		s.stats.Data, _ = json.Marshal(s.StatsHandler(s.Endpoint))
-	}
 	info := s.Info()
-	return Stats{
+	stats := Stats{
 		ServiceIdentity: ServiceIdentity{
 			Name:    info.Name,
 			ID:      info.ID,
 			Version: info.Version,
 		},
-		Type:                  StatsResponseType,
-		NumRequests:           s.stats.NumRequests,
-		NumErrors:             s.stats.NumErrors,
-		ProcessingTime:        s.stats.ProcessingTime,
-		AverageProcessingTime: s.stats.AverageProcessingTime,
-		Started:               s.stats.Started,
-		Data:                  s.stats.Data,
+		Endpoints: make([]*EndpointStats, 0),
+	}
+	for _, endpoint := range s.endpoints {
+		endpointStats := &EndpointStats{
+			NumRequests:           endpoint.stats.NumRequests,
+			NumErrors:             endpoint.stats.NumErrors,
+			LastError:             endpoint.stats.LastError,
+			ProcessingTime:        endpoint.stats.ProcessingTime,
+			AverageProcessingTime: endpoint.stats.AverageProcessingTime,
+		}
+		if s.StatsHandler != nil {
+			data, _ := json.Marshal(s.StatsHandler(endpoint))
+			endpointStats.Data = data
+		}
+		stats.Endpoints = append(stats.Endpoints, endpointStats)
 	}
+	return stats
 }
 
 // Reset resets all statistics on a service instance.
 func (s *service) Reset() {
 	s.m.Lock()
-	s.stats = &Stats{
-		ServiceIdentity: s.Info().ServiceIdentity,
+	for _, endpoint := range s.endpoints {
+		endpoint.reset()
 	}
+	s.started = time.Now().UTC()
 	s.m.Unlock()
 }
 
@@ -587,6 +731,63 @@ func (s *service) Stopped() bool {
 	return s.stopped
 }
 
+func (e *NATSError) Error() string {
+	return fmt.Sprintf("%q: %s", e.Subject, e.Description)
+}
+
+func (g *group) AddEndpoint(name string, handler Handler, opts ...EndpointOpt) error {
+	var options endpointOpts
+	for _, opt := range opts {
+		if err := opt(&options); err != nil {
+			return err
+		}
+	}
+	subject := name
+	if options.subject != "" {
+		subject = options.subject
+	}
+	endpointSubject := fmt.Sprintf("%s.%s", g.prefix, subject)
+	if g.prefix == "" {
+		endpointSubject = subject
+	}
+	return addEndpoint(g.service, name, endpointSubject, handler)
+}
+
+func (g *group) AddGroup(name string) Group {
+	parts := make([]string, 0, 2)
+	if g.prefix != "" {
+		parts = append(parts, g.prefix)
+	}
+	if name != "" {
+		parts = append(parts, name)
+	}
+	prefix := strings.Join(parts, ".")
+
+	return &group{
+		service: g.service,
+		prefix:  prefix,
+	}
+}
+
+func (e *Endpoint) stop() error {
+	if err := e.subscription.Drain(); err != nil {
+		return fmt.Errorf("draining subscription for request handler: %w", err)
+	}
+	for i, endpoint := range e.service.endpoints {
+		if endpoint.Subject == e.Subject {
+			e.service.endpoints = append(e.service.endpoints[:i], e.service.endpoints[i+1:]...)
+		}
+	}
+	return nil
+}
+
+func (e *Endpoint) reset() {
+	e.stats = EndpointStats{
+		Name:    e.stats.Name,
+		Subject: e.stats.Subject,
+	}
+}
+
 // ControlSubject returns monitoring subjects used by the Service.
 // Providing a verb is mandatory (it should be one of Ping, Schema, Info or Stats).
 // Depending on whether kind and id are provided, ControlSubject will return one of the following:
@@ -610,6 +811,9 @@ func ControlSubject(verb Verb, name, id string) (string, error) {
 	return fmt.Sprintf("%s.%s.%s.%s", APIPrefix, verbStr, name, id), nil
 }
 
-func (e *NATSError) Error() string {
-	return fmt.Sprintf("%q: %s", e.Subject, e.Description)
+func WithEndpointSubject(subject string) EndpointOpt {
+	return func(e *endpointOpts) error {
+		e.subject = subject
+		return nil
+	}
 }
diff --git a/nats.go b/nats.go
index 6ccb351..53e5159 100644
--- a/nats.go
+++ b/nats.go
@@ -839,7 +839,7 @@ func RootCAs(file ...string) Option {
 		for _, f := range file {
 			rootPEM, err := os.ReadFile(f)
 			if err != nil || rootPEM == nil {
-				return fmt.Errorf("nats: error loading or parsing rootCA file: %v", err)
+				return fmt.Errorf("nats: error loading or parsing rootCA file: %w", err)
 			}
 			ok := pool.AppendCertsFromPEM(rootPEM)
 			if !ok {
@@ -861,11 +861,11 @@ func ClientCert(certFile, keyFile string) Option {
 	return func(o *Options) error {
 		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
 		if err != nil {
-			return fmt.Errorf("nats: error loading client certificate: %v", err)
+			return fmt.Errorf("nats: error loading client certificate: %w", err)
 		}
 		cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
 		if err != nil {
-			return fmt.Errorf("nats: error parsing client certificate: %v", err)
+			return fmt.Errorf("nats: error parsing client certificate: %w", err)
 		}
 		if o.TLSConfig == nil {
 			o.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12}
@@ -1113,7 +1113,7 @@ func UserJWTAndSeed(jwt string, seed string) Option {
 	sigCB := func(nonce []byte) ([]byte, error) {
 		kp, err := nkeys.FromSeed([]byte(seed))
 		if err != nil {
-			return nil, fmt.Errorf("unable to extract key pair from seed: %v", err)
+			return nil, fmt.Errorf("unable to extract key pair from seed: %w", err)
 		}
 		// Wipe our key on exit.
 		defer kp.Wipe()
@@ -1136,6 +1136,12 @@ func UserJWT(userCB UserJWTHandler, sigCB SignatureHandler) Option {
 		if sigCB == nil {
 			return ErrUserButNoSigCB
 		}
+		// Smoke test the user callback to ensure it is setup properly
+		// when processing options.
+		if _, err := userCB(); err != nil {
+			return err
+		}
+
 		o.UserJWT = userCB
 		o.SignatureCB = sigCB
 		return nil
@@ -2382,7 +2388,7 @@ func (nc *Conn) connectProto() (string, error) {
 		}
 		sigraw, err := o.SignatureCB([]byte(nc.info.Nonce))
 		if err != nil {
-			return _EMPTY_, fmt.Errorf("error signing nonce: %v", err)
+			return _EMPTY_, fmt.Errorf("error signing nonce: %w", err)
 		}
 		sig = base64.RawURLEncoding.EncodeToString(sigraw)
 	}
@@ -3553,7 +3559,8 @@ const (
 
 // decodeHeadersMsg will decode and headers.
 func decodeHeadersMsg(data []byte) (Header, error) {
-	tp := textproto.NewReader(bufio.NewReader(bytes.NewReader(data)))
+	br := bufio.NewReaderSize(bytes.NewReader(data), 128)
+	tp := textproto.NewReader(br)
 	l, err := tp.ReadLine()
 	if err != nil || len(l) < hdrPreEnd || l[:hdrPreEnd] != hdrLine[:hdrPreEnd] {
 		return nil, ErrBadHeaderMsg
@@ -5432,12 +5439,12 @@ func wipeSlice(buf []byte) {
 func userFromFile(userFile string) (string, error) {
 	path, err := expandPath(userFile)
 	if err != nil {
-		return _EMPTY_, fmt.Errorf("nats: %v", err)
+		return _EMPTY_, fmt.Errorf("nats: %w", err)
 	}
 
 	contents, err := os.ReadFile(path)
 	if err != nil {
-		return _EMPTY_, fmt.Errorf("nats: %v", err)
+		return _EMPTY_, fmt.Errorf("nats: %w", err)
 	}
 	defer wipeSlice(contents)
 	return nkeys.ParseDecoratedJWT(contents)
@@ -5486,7 +5493,7 @@ func expandPath(p string) (string, error) {
 func nkeyPairFromSeedFile(seedFile string) (nkeys.KeyPair, error) {
 	contents, err := os.ReadFile(seedFile)
 	if err != nil {
-		return nil, fmt.Errorf("nats: %v", err)
+		return nil, fmt.Errorf("nats: %w", err)
 	}
 	defer wipeSlice(contents)
 	return nkeys.ParseDecoratedNKey(contents)
@@ -5497,7 +5504,7 @@ func nkeyPairFromSeedFile(seedFile string) (nkeys.KeyPair, error) {
 func sigHandler(nonce []byte, seedFile string) ([]byte, error) {
 	kp, err := nkeyPairFromSeedFile(seedFile)
 	if err != nil {
-		return nil, fmt.Errorf("unable to extract key pair from file %q: %v", seedFile, err)
+		return nil, fmt.Errorf("unable to extract key pair from file %q: %w", seedFile, err)
 	}
 	// Wipe our key on exit.
 	defer kp.Wipe()


Test patch:
diff --git a/go_test.mod b/go_test.mod
index 961ccea..401685f 100644
--- a/go_test.mod
+++ b/go_test.mod
@@ -1,6 +1,6 @@
 module github.com/nats-io/nats.go
 
-go 1.17
+go 1.18
 
 require (
 	github.com/golang/protobuf v1.4.2
@@ -14,7 +14,7 @@ require (
 	github.com/klauspost/compress v1.15.11 // indirect
 	github.com/minio/highwayhash v1.0.2 // indirect
 	github.com/nats-io/jwt/v2 v2.3.0 // indirect
-	golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be // indirect
-	golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec // indirect
+	golang.org/x/crypto v0.5.0 // indirect
+	golang.org/x/sys v0.4.0 // indirect
 	golang.org/x/time v0.0.0-20220922220347-f3bd1da661af // indirect
 )
diff --git a/go_test.sum b/go_test.sum
index 23c41c4..1662e18 100644
--- a/go_test.sum
+++ b/go_test.sum
@@ -1,5 +1,3 @@
-github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
 github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
 github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
@@ -13,40 +11,26 @@ github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/klauspost/compress v1.15.11 h1:Lcadnb3RKGin4FYM/orgq0qde+nc15E5Cbqg4B9Sx9c=
 github.com/klauspost/compress v1.15.11/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
-github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
-github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
-github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/nats-io/jwt/v2 v2.3.0 h1:z2mA1a7tIf5ShggOFlR1oBPgd6hGqcDYsISxZByUzdI=
 github.com/nats-io/jwt/v2 v2.3.0/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
 github.com/nats-io/nats-server/v2 v2.9.6 h1:RTtK+rv/4CcliOuqGsy58g7MuWkBaWmF5TUNwuUo9Uw=
 github.com/nats-io/nats-server/v2 v2.9.6/go.mod h1:AB6hAnGZDlYfqb7CTAm66ZKMZy9DpfierY1/PbpvI2g=
-github.com/nats-io/nats.go v1.19.0/go.mod h1:tLqubohF7t4z3du1QDPYJIQQyhb4wl6DhjxEajSI7UA=
 github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
-github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/prashantv/gostub v1.1.0/go.mod h1:A5zLQHz7ieHGG7is6LLXLz7I8+3LZzsrV0P1IAHhP5U=
-github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-go.uber.org/automaxprocs v1.5.1/go.mod h1:BF4eumQw0P9GtnuxxovUd06vwm1o18oMzFtK66vU6XU=
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be h1:fmw3UbQh+nxngCAHrDCCztao/kbYFnWjoqop8dHx05A=
-golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.5.0 h1:U/0M97KRkSFvyD/3FSmdP5W5swImpNgle/EHFhOsQPE=
+golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
-golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec h1:BkDtF2Ih9xZ7le9ndzTA7KJow28VbQW3odyk/8drmuI=
-golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.4.0 h1:Zr2JFtRQNX3BCZ8YtxRE9hNJYC8J6I1MVbMg6owUp18=
+golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/time v0.0.0-20220922220347-f3bd1da661af h1:Yx9k8YCG3dvF87UAn2tu2HQLf2dt/eR1bXxpLMWeH+Y=
 golang.org/x/time v0.0.0-20220922220347-f3bd1da661af/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
@@ -59,7 +43,3 @@ google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miE
 google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
 google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM=
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/micro/example_handler_test.go b/micro/example_handler_test.go
index 6eff366..27b1db5 100644
--- a/micro/example_handler_test.go
+++ b/micro/example_handler_test.go
@@ -1,3 +1,16 @@
+// Copyright 2022-2023 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package micro_test
 
 import (
@@ -32,9 +45,9 @@ func ExampleHandler() {
 	config := micro.Config{
 		Name:    "RectangleAreaService",
 		Version: "0.1.0",
-		Endpoint: micro.Endpoint{
+		Endpoint: &micro.EndpointConfig{
+			Subject: "area.rectangle",
 			Handler: rec,
-			Subject: "rectangle.area",
 		},
 	}
 	svc, err := micro.AddService(nc, config)
diff --git a/micro/example_package_test.go b/micro/example_package_test.go
index 9b8cdf8..1472c8a 100644
--- a/micro/example_package_test.go
+++ b/micro/example_package_test.go
@@ -30,8 +30,13 @@ func Example() {
 	}
 	defer nc.Close()
 
-	// service handler - in this case, HandlerFunc is used,
+	// endpoint handler - in this case, HandlerFunc is used,
 	// which is a built-in implementation of Handler interface
+	echoHandler := func(req micro.Request) {
+		req.Respond(req.Data())
+	}
+
+	// second endpoint
 	incrementHandler := func(req micro.Request) {
 		val, err := strconv.Atoi(string(req.Data()))
 		if err != nil {
@@ -43,29 +48,50 @@ func Example() {
 		req.Respond([]byte(strconv.Itoa(responseData)))
 	}
 
+	// third endpoint
+	multiplyHandler := func(req micro.Request) {
+		val, err := strconv.Atoi(string(req.Data()))
+		if err != nil {
+			req.Error("400", "request data should be a number", nil)
+			return
+		}
+
+		responseData := val * 2
+		req.Respond([]byte(strconv.Itoa(responseData)))
+	}
+
 	config := micro.Config{
 		Name:        "IncrementService",
 		Version:     "0.1.0",
 		Description: "Increment numbers",
-		Endpoint: micro.Endpoint{
-			// service handler
-			Handler: micro.HandlerFunc(incrementHandler),
-			// a unique subject serving as a service endpoint
-			Subject: "numbers.increment",
+
+		// base handler - for simple services with single endpoints this is sufficient
+		Endpoint: &micro.EndpointConfig{
+			Subject: "echo",
+			Handler: micro.HandlerFunc(echoHandler),
 		},
 	}
-	// Multiple instances of the servcice with the same name can be created.
-	// Requests to a service with the same name will be load-balanced.
-	for i := 0; i < 5; i++ {
-		svc, err := micro.AddService(nc, config)
-		if err != nil {
-			log.Fatal(err)
-		}
-		defer svc.Stop()
+	svc, err := micro.AddService(nc, config)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer svc.Stop()
+
+	// add a group to aggregate endpoints under common prefix
+	numbers := svc.AddGroup("numbers")
+
+	// register endpoints in a group
+	err = numbers.AddEndpoint("Increment", micro.HandlerFunc(incrementHandler))
+	if err != nil {
+		log.Fatal(err)
+	}
+	err = numbers.AddEndpoint("Multiply", micro.HandlerFunc(multiplyHandler))
+	if err != nil {
+		log.Fatal(err)
 	}
 
 	// send a request to a service
-	resp, err := nc.Request("numbers.increment", []byte("3"), 1*time.Second)
+	resp, err := nc.Request("numbers.Increment", []byte("3"), 1*time.Second)
 	if err != nil {
 		log.Fatal(err)
 	}
diff --git a/micro/example_test.go b/micro/example_test.go
index 43a3248..1e16170 100644
--- a/micro/example_test.go
+++ b/micro/example_test.go
@@ -1,4 +1,4 @@
-// Copyright 2022 The NATS Authors
+// Copyright 2022-2023 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -37,11 +37,6 @@ func ExampleAddService() {
 		Name:        "EchoService",
 		Version:     "1.0.0",
 		Description: "Send back what you receive",
-		Endpoint: micro.Endpoint{
-			Subject: "echo",
-			Handler: micro.HandlerFunc(echoHandler),
-		},
-
 		// DoneHandler can be set to customize behavior on stopping a service.
 		DoneHandler: func(srv micro.Service) {
 			info := srv.Info()
@@ -53,6 +48,12 @@ func ExampleAddService() {
 			info := srv.Info()
 			fmt.Printf("Service %q returned an error on subject %q: %s", info.Name, err.Subject, err.Description)
 		},
+
+		// optional base handler
+		Endpoint: &micro.EndpointConfig{
+			Subject: "echo",
+			Handler: micro.HandlerFunc(echoHandler),
+		},
 	}
 
 	srv, err := micro.AddService(nc, config)
@@ -62,6 +63,92 @@ func ExampleAddService() {
 	defer srv.Stop()
 }
 
+func ExampleService_AddEndpoint() {
+	nc, err := nats.Connect("127.0.0.1:4222")
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer nc.Close()
+
+	echoHandler := func(req micro.Request) {
+		req.Respond(req.Data())
+	}
+
+	config := micro.Config{
+		Name:    "EchoService",
+		Version: "1.0.0",
+	}
+
+	srv, err := micro.AddService(nc, config)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// endpoint will be registered under "Echo" subject
+	err = srv.AddEndpoint("Echo", micro.HandlerFunc(echoHandler))
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
+func ExampleWithEndpointSubject() {
+	nc, err := nats.Connect("127.0.0.1:4222")
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer nc.Close()
+
+	echoHandler := func(req micro.Request) {
+		req.Respond(req.Data())
+	}
+
+	config := micro.Config{
+		Name:    "EchoService",
+		Version: "1.0.0",
+	}
+
+	srv, err := micro.AddService(nc, config)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// endpoint will be registered under "service.echo" subject
+	err = srv.AddEndpoint("Echo", micro.HandlerFunc(echoHandler), micro.WithEndpointSubject("service.echo"))
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
+func ExampleService_AddGroup() {
+	nc, err := nats.Connect("127.0.0.1:4222")
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer nc.Close()
+
+	echoHandler := func(req micro.Request) {
+		req.Respond(req.Data())
+	}
+
+	config := micro.Config{
+		Name:    "EchoService",
+		Version: "1.0.0",
+	}
+
+	srv, err := micro.AddService(nc, config)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	v1 := srv.AddGroup("v1")
+
+	// endpoint will be registered under "v1.Echo" subject
+	err = v1.AddEndpoint("Echo", micro.HandlerFunc(echoHandler))
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
 func ExampleService_Info() {
 	nc, err := nats.Connect("127.0.0.1:4222")
 	if err != nil {
@@ -71,10 +158,6 @@ func ExampleService_Info() {
 
 	config := micro.Config{
 		Name: "EchoService",
-		Endpoint: micro.Endpoint{
-			Subject: "echo",
-			Handler: micro.HandlerFunc(func(micro.Request) {}),
-		},
 	}
 
 	srv, _ := micro.AddService(nc, config)
@@ -86,7 +169,7 @@ func ExampleService_Info() {
 	fmt.Println(info.Name)
 	fmt.Println(info.Description)
 	fmt.Println(info.Version)
-	fmt.Println(info.Subject)
+	fmt.Println(info.Subjects)
 }
 
 func ExampleService_Stats() {
@@ -99,19 +182,18 @@ func ExampleService_Stats() {
 	config := micro.Config{
 		Name:    "EchoService",
 		Version: "0.1.0",
-		Endpoint: micro.Endpoint{
+		Endpoint: &micro.EndpointConfig{
 			Subject: "echo",
-			Handler: micro.HandlerFunc(func(micro.Request) {}),
+			Handler: micro.HandlerFunc(func(r micro.Request) {}),
 		},
 	}
 
 	srv, _ := micro.AddService(nc, config)
-
 	// stats of a service instance
 	stats := srv.Stats()
 
-	fmt.Println(stats.AverageProcessingTime)
-	fmt.Println(stats.ProcessingTime)
+	fmt.Println(stats.Endpoints[0].AverageProcessingTime)
+	fmt.Println(stats.Endpoints[0].ProcessingTime)
 
 }
 
@@ -125,10 +207,6 @@ func ExampleService_Stop() {
 	config := micro.Config{
 		Name:    "EchoService",
 		Version: "0.1.0",
-		Endpoint: micro.Endpoint{
-			Subject: "echo",
-			Handler: micro.HandlerFunc(func(micro.Request) {}),
-		},
 	}
 
 	srv, _ := micro.AddService(nc, config)
@@ -156,10 +234,6 @@ func ExampleService_Stopped() {
 	config := micro.Config{
 		Name:    "EchoService",
 		Version: "0.1.0",
-		Endpoint: micro.Endpoint{
-			Subject: "echo",
-			Handler: micro.HandlerFunc(func(micro.Request) {}),
-		},
 	}
 
 	srv, _ := micro.AddService(nc, config)
@@ -185,10 +259,6 @@ func ExampleService_Reset() {
 	config := micro.Config{
 		Name:    "EchoService",
 		Version: "0.1.0",
-		Endpoint: micro.Endpoint{
-			Subject: "echo",
-			Handler: micro.HandlerFunc(func(micro.Request) {}),
-		},
 	}
 
 	srv, _ := micro.AddService(nc, config)
diff --git a/micro/test/service_test.go b/micro/test/service_test.go
index 38816cc..4ec52b3 100644
--- a/micro/test/service_test.go
+++ b/micro/test/service_test.go
@@ -1,4 +1,4 @@
-// Copyright 2022 The NATS Authors
+// Copyright 2022-2023 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -66,11 +66,11 @@ func TestServiceBasics(t *testing.T) {
 		Name:        "CoolAddService",
 		Version:     "0.1.0",
 		Description: "Add things together",
-		Endpoint: micro.Endpoint{
+		Schema:      micro.Schema{Request: "", Response: ""},
+		Endpoint: &micro.EndpointConfig{
 			Subject: "svc.add",
 			Handler: micro.HandlerFunc(doAdd),
 		},
-		Schema: micro.Schema{Request: "", Response: ""},
 	}
 
 	for i := 0; i < 5; i++ {
@@ -114,9 +114,6 @@ func TestServiceBasics(t *testing.T) {
 	if err := json.Unmarshal(info.Data, &inf); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	if inf.Subject != "svc.add" {
-		t.Fatalf("expected service subject to be srv.add: %s", inf.Subject)
-	}
 
 	// Ping all services. Multiple responses.
 	inbox := nats.NewInbox()
@@ -169,7 +166,7 @@ func TestServiceBasics(t *testing.T) {
 		if err := json.Unmarshal(resp.Data, &srvStats); err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
-		requestsNum += srvStats.NumRequests
+		requestsNum += srvStats.Endpoints[0].NumRequests
 		stats = append(stats, srvStats)
 	}
 	if len(stats) != 5 {
@@ -182,12 +179,8 @@ func TestServiceBasics(t *testing.T) {
 	}
 	// Reset stats for a service
 	svcs[0].Reset()
-	emptyStats := micro.Stats{
-		Type:            micro.StatsResponseType,
-		ServiceIdentity: svcs[0].Info().ServiceIdentity,
-	}
 
-	if !reflect.DeepEqual(svcs[0].Stats(), emptyStats) {
+	if svcs[0].Stats().Endpoints[0].NumRequests != 0 {
 		t.Fatalf("Expected empty stats after reset; got: %+v", svcs[0].Stats())
 	}
 
@@ -203,6 +196,7 @@ func TestAddService(t *testing.T) {
 	tests := []struct {
 		name              string
 		givenConfig       micro.Config
+		endpoints         []string
 		natsClosedHandler nats.ConnHandler
 		natsErrorHandler  nats.ErrHandler
 		asyncErrorSubject string
@@ -214,8 +208,22 @@ func TestAddService(t *testing.T) {
 			givenConfig: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
+			},
+			expectedPing: micro.Ping{
+				Type: micro.PingResponseType,
+				ServiceIdentity: micro.ServiceIdentity{
+					Name:    "test_service",
+					Version: "0.1.0",
+				},
+			},
+		},
+		{
+			name: "with single base endpoint",
+			givenConfig: micro.Config{
+				Name:    "test_service",
+				Version: "0.1.0",
+				Endpoint: &micro.EndpointConfig{
+					Subject: "test",
 					Handler: micro.HandlerFunc(testHandler),
 				},
 			},
@@ -228,18 +236,34 @@ func TestAddService(t *testing.T) {
 			},
 		},
 		{
-			name: "with done handler, no handlers on nats connection",
+			name: "with base endpoint and additional endpoints",
 			givenConfig: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
+				Endpoint: &micro.EndpointConfig{
+					Subject: "test",
 					Handler: micro.HandlerFunc(testHandler),
 				},
+			},
+			endpoints: []string{"func1", "func2", "func3"},
+			expectedPing: micro.Ping{
+				Type: micro.PingResponseType,
+				ServiceIdentity: micro.ServiceIdentity{
+					Name:    "test_service",
+					Version: "0.1.0",
+				},
+			},
+		},
+		{
+			name: "with done handler, no handlers on nats connection",
+			givenConfig: micro.Config{
+				Name:    "test_service",
+				Version: "0.1.0",
 				DoneHandler: func(micro.Service) {
 					doneService <- struct{}{}
 				},
 			},
+			endpoints: []string{"func"},
 			expectedPing: micro.Ping{
 				Type: micro.PingResponseType,
 				ServiceIdentity: micro.ServiceIdentity{
@@ -253,14 +277,11 @@ func TestAddService(t *testing.T) {
 			givenConfig: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
-					Handler: micro.HandlerFunc(testHandler),
-				},
 				ErrorHandler: func(micro.Service, *micro.NATSError) {
 					errService <- struct{}{}
 				},
 			},
+			endpoints: []string{"func"},
 			expectedPing: micro.Ping{
 				Type: micro.PingResponseType,
 				ServiceIdentity: micro.ServiceIdentity{
@@ -268,21 +289,18 @@ func TestAddService(t *testing.T) {
 					Version: "0.1.0",
 				},
 			},
-			asyncErrorSubject: "test.sub",
+			asyncErrorSubject: "func",
 		},
 		{
 			name: "with error handler, no handlers on nats connection, error on monitoring subject",
 			givenConfig: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
-					Handler: micro.HandlerFunc(testHandler),
-				},
 				ErrorHandler: func(micro.Service, *micro.NATSError) {
 					errService <- struct{}{}
 				},
 			},
+			endpoints: []string{"func"},
 			expectedPing: micro.Ping{
 				Type: micro.PingResponseType,
 				ServiceIdentity: micro.ServiceIdentity{
@@ -290,21 +308,18 @@ func TestAddService(t *testing.T) {
 					Version: "0.1.0",
 				},
 			},
-			asyncErrorSubject: "$SVC.PING.TEST_SERVICE",
+			asyncErrorSubject: "$SRV.PING.test_service",
 		},
 		{
 			name: "with done handler, append to nats handlers",
 			givenConfig: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
-					Handler: micro.HandlerFunc(testHandler),
-				},
 				DoneHandler: func(micro.Service) {
 					doneService <- struct{}{}
 				},
 			},
+			endpoints: []string{"func"},
 			natsClosedHandler: func(c *nats.Conn) {
 				closedNats <- struct{}{}
 			},
@@ -325,14 +340,11 @@ func TestAddService(t *testing.T) {
 			givenConfig: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
-					Handler: micro.HandlerFunc(testHandler),
-				},
 				DoneHandler: func(micro.Service) {
 					doneService <- struct{}{}
 				},
 			},
+			endpoints: []string{"func"},
 			natsClosedHandler: func(c *nats.Conn) {
 				closedNats <- struct{}{}
 			},
@@ -352,14 +364,11 @@ func TestAddService(t *testing.T) {
 			givenConfig: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
-					Handler: micro.HandlerFunc(testHandler),
-				},
 				DoneHandler: func(micro.Service) {
 					doneService <- struct{}{}
 				},
 			},
+			endpoints: []string{"func"},
 			natsClosedHandler: func(c *nats.Conn) {
 				closedNats <- struct{}{}
 			},
@@ -373,18 +382,15 @@ func TestAddService(t *testing.T) {
 					Version: "0.1.0",
 				},
 			},
-			asyncErrorSubject: "$SVC.PING.TEST_SERVICE",
+			asyncErrorSubject: "$SRV.PING.TEST_SERVICE",
 		},
 		{
 			name: "validation error, invalid service name",
 			givenConfig: micro.Config{
 				Name:    "test_service!",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
-					Handler: micro.HandlerFunc(testHandler),
-				},
 			},
+			endpoints: []string{"func"},
 			withError: micro.ErrConfigValidation,
 		},
 		{
@@ -392,35 +398,17 @@ func TestAddService(t *testing.T) {
 			givenConfig: micro.Config{
 				Name:    "test_service!",
 				Version: "abc",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
-					Handler: micro.HandlerFunc(testHandler),
-				},
 			},
+			endpoints: []string{"func"},
 			withError: micro.ErrConfigValidation,
 		},
 		{
-			name: "validation error, empty subject",
+			name: "validation error, invalid endpoint name",
 			givenConfig: micro.Config{
-				Name:    "test_service",
-				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "",
-					Handler: micro.HandlerFunc(testHandler),
-				},
-			},
-			withError: micro.ErrConfigValidation,
-		},
-		{
-			name: "validation error, no handler",
-			givenConfig: micro.Config{
-				Name:    "test_service",
-				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test_subject",
-					Handler: nil,
-				},
+				Name:    "test_service!",
+				Version: "abc",
 			},
+			endpoints: []string{"endpoint name"},
 			withError: micro.ErrConfigValidation,
 		},
 	}
@@ -449,8 +437,20 @@ func TestAddService(t *testing.T) {
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
+			for _, endpoint := range test.endpoints {
+				if err := srv.AddEndpoint(endpoint, micro.HandlerFunc(testHandler)); err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+			}
 
 			info := srv.Info()
+			subjectsNum := len(test.endpoints)
+			if test.givenConfig.Endpoint != nil {
+				subjectsNum += 1
+			}
+			if subjectsNum != len(info.Subjects) {
+				t.Fatalf("Invalid number of registered endpoints; want: %d; got: %d", subjectsNum, len(info.Subjects))
+			}
 			pingSubject, err := micro.ControlSubject(micro.PingVerb, info.Name, info.ID)
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
@@ -534,6 +534,192 @@ func TestAddService(t *testing.T) {
 	}
 }
 
+func TestErrHandlerSubjectMatch(t *testing.T) {
+	tests := []struct {
+		name             string
+		endpointSubject  string
+		errSubject       string
+		expectServiceErr bool
+	}{
+		{
+			name:             "exact match",
+			endpointSubject:  "foo.bar.baz",
+			errSubject:       "foo.bar.baz",
+			expectServiceErr: true,
+		},
+		{
+			name:             "match with *",
+			endpointSubject:  "foo.*.baz",
+			errSubject:       "foo.bar.baz",
+			expectServiceErr: true,
+		},
+		{
+			name:             "match with >",
+			endpointSubject:  "foo.bar.>",
+			errSubject:       "foo.bar.baz.1",
+			expectServiceErr: true,
+		},
+		{
+			name:             "monitoring handler",
+			endpointSubject:  "foo.bar.>",
+			errSubject:       "$SRV.PING",
+			expectServiceErr: true,
+		},
+		{
+			name:             "endpoint longer than subject",
+			endpointSubject:  "foo.bar.baz",
+			errSubject:       "foo.bar",
+			expectServiceErr: false,
+		},
+		{
+			name:             "no match",
+			endpointSubject:  "foo.bar.baz",
+			errSubject:       "foo.baz.bar",
+			expectServiceErr: false,
+		},
+		{
+			name:             "no match with *",
+			endpointSubject:  "foo.*.baz",
+			errSubject:       "foo.bar.foo",
+			expectServiceErr: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			coreNatsAsyncErrors := []nats.ErrHandler{nil, func(c *nats.Conn, s *nats.Subscription, err error) {}}
+			for _, cb := range coreNatsAsyncErrors {
+				errChan := make(chan struct{})
+				errHandler := func(s micro.Service, err *micro.NATSError) {
+					errChan <- struct{}{}
+				}
+				s := RunServerOnPort(-1)
+				defer s.Shutdown()
+
+				nc, err := nats.Connect(s.ClientURL())
+				if err != nil {
+					t.Fatalf("Expected to connect to server, got %v", err)
+				}
+				defer nc.Close()
+				nc.SetErrorHandler(cb)
+				svc, err := micro.AddService(nc, micro.Config{
+					Name:         "test_service",
+					Version:      "0.0.1",
+					ErrorHandler: micro.ErrHandler(errHandler),
+					Endpoint: &micro.EndpointConfig{
+						Subject: test.endpointSubject,
+						Handler: micro.HandlerFunc(func(r micro.Request) {}),
+					},
+				})
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				defer svc.Stop()
+
+				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.errSubject}, fmt.Errorf("oops"))
+				if test.expectServiceErr {
+					select {
+					case <-errChan:
+					case <-time.After(10 * time.Millisecond):
+						t.Fatalf("Expected service error callback")
+					}
+				} else {
+					select {
+					case <-errChan:
+						t.Fatalf("Expected no service error callback")
+					case <-time.After(10 * time.Millisecond):
+					}
+				}
+			}
+		})
+	}
+}
+
+func TestGroups(t *testing.T) {
+	tests := []struct {
+		name            string
+		endpointName    string
+		groups          []string
+		expectedSubject string
+	}{
+		{
+			name:            "no groups",
+			endpointName:    "foo",
+			expectedSubject: "foo",
+		},
+		{
+			name:            "single group",
+			endpointName:    "foo",
+			groups:          []string{"g1"},
+			expectedSubject: "g1.foo",
+		},
+		{
+			name:            "single empty group",
+			endpointName:    "foo",
+			groups:          []string{""},
+			expectedSubject: "foo",
+		},
+		{
+			name:            "empty groups",
+			endpointName:    "foo",
+			groups:          []string{"", "g1", ""},
+			expectedSubject: "g1.foo",
+		},
+		{
+			name:            "multiple groups",
+			endpointName:    "foo",
+			groups:          []string{"g1", "g2", "g3"},
+			expectedSubject: "g1.g2.g3.foo",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			s := RunServerOnPort(-1)
+			defer s.Shutdown()
+
+			nc, err := nats.Connect(s.ClientURL())
+			if err != nil {
+				t.Fatalf("Expected to connect to server, got %v", err)
+			}
+			defer nc.Close()
+
+			srv, err := micro.AddService(nc, micro.Config{
+				Name:    "test_service",
+				Version: "0.0.1",
+			})
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			defer srv.Stop()
+
+			if len(test.groups) > 0 {
+				group := srv.AddGroup(test.groups[0])
+				for _, g := range test.groups[1:] {
+					group = group.AddGroup(g)
+				}
+				err = group.AddEndpoint(test.endpointName, micro.HandlerFunc(func(r micro.Request) {}))
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+			} else {
+				err = srv.AddEndpoint(test.endpointName, micro.HandlerFunc(func(r micro.Request) {}))
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+			}
+
+			info := srv.Info()
+			if len(info.Subjects) != 1 {
+				t.Fatalf("Expected 1 registered endpoint; got: %d", len(info.Subjects))
+			}
+			if info.Subjects[0] != test.expectedSubject {
+				t.Fatalf("Invalid subject; want: %s, got: %s", test.expectedSubject, info.Subjects[0])
+			}
+		})
+	}
+}
+
 func TestMonitoringHandlers(t *testing.T) {
 	s := RunServerOnPort(-1)
 	defer s.Shutdown()
@@ -552,14 +738,14 @@ func TestMonitoringHandlers(t *testing.T) {
 	config := micro.Config{
 		Name:    "test_service",
 		Version: "0.1.0",
-		Endpoint: micro.Endpoint{
-			Subject: "test.sub",
-			Handler: micro.HandlerFunc(func(micro.Request) {}),
-		},
 		Schema: micro.Schema{
 			Request: "some_schema",
 		},
 		ErrorHandler: errHandler,
+		Endpoint: &micro.EndpointConfig{
+			Subject: "test.func",
+			Handler: micro.HandlerFunc(func(r micro.Request) {}),
+		},
 	}
 	srv, err := micro.AddService(nc, config)
 	if err != nil {
@@ -626,7 +812,7 @@ func TestMonitoringHandlers(t *testing.T) {
 					Version: "0.1.0",
 					ID:      info.ID,
 				},
-				Subject: "test.sub",
+				Subjects: []string{"test.func"},
 			},
 		},
 		{
@@ -639,7 +825,7 @@ func TestMonitoringHandlers(t *testing.T) {
 					Version: "0.1.0",
 					ID:      info.ID,
 				},
-				Subject: "test.sub",
+				Subjects: []string{"test.func"},
 			},
 		},
 		{
@@ -652,7 +838,7 @@ func TestMonitoringHandlers(t *testing.T) {
 					Version: "0.1.0",
 					ID:      info.ID,
 				},
-				Subject: "test.sub",
+				Subjects: []string{"test.func"},
 			},
 		},
 		{
@@ -776,10 +962,6 @@ func TestServiceStats(t *testing.T) {
 			config: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
-					Handler: micro.HandlerFunc(handler),
-				},
 			},
 		},
 		{
@@ -787,11 +969,7 @@ func TestServiceStats(t *testing.T) {
 			config: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
-					Handler: micro.HandlerFunc(handler),
-				},
-				StatsHandler: func(e micro.Endpoint) interface{} {
+				StatsHandler: func(e *micro.Endpoint) interface{} {
 					return map[string]interface{}{
 						"key": "val",
 					}
@@ -806,10 +984,6 @@ func TestServiceStats(t *testing.T) {
 			config: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
-					Handler: micro.HandlerFunc(handler),
-				},
 				Schema: micro.Schema{
 					Request: "some_schema",
 				},
@@ -820,14 +994,10 @@ func TestServiceStats(t *testing.T) {
 			config: micro.Config{
 				Name:    "test_service",
 				Version: "0.1.0",
-				Endpoint: micro.Endpoint{
-					Subject: "test.sub",
-					Handler: micro.HandlerFunc(handler),
-				},
 				Schema: micro.Schema{
 					Request: "some_schema",
 				},
-				StatsHandler: func(e micro.Endpoint) interface{} {
+				StatsHandler: func(e *micro.Endpoint) interface{} {
 					return map[string]interface{}{
 						"key": "val",
 					}
@@ -854,16 +1024,19 @@ func TestServiceStats(t *testing.T) {
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
+			if err := srv.AddEndpoint("func", micro.HandlerFunc(handler), micro.WithEndpointSubject("test.func")); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
 			defer srv.Stop()
 			for i := 0; i < 10; i++ {
-				if _, err := nc.Request(srv.Info().Subject, []byte("msg"), time.Second); err != nil {
+				if _, err := nc.Request("test.func", []byte("msg"), time.Second); err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
 			}
 
 			// Malformed request, missing reply subjtct
 			// This should be reflected in errors
-			if err := nc.Publish(srv.Info().Subject, []byte("err")); err != nil {
+			if err := nc.Publish("test.func", []byte("err")); err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
 			time.Sleep(10 * time.Millisecond)
@@ -885,15 +1058,15 @@ func TestServiceStats(t *testing.T) {
 			if stats.ID != info.ID {
 				t.Errorf("Unexpected service name; want: %s; got: %s", info.ID, stats.ID)
 			}
-			if stats.NumRequests != 11 {
-				t.Errorf("Unexpected num_requests; want: 11; got: %d", stats.NumRequests)
+			if stats.Endpoints[0].NumRequests != 11 {
+				t.Errorf("Unexpected num_requests; want: 11; got: %d", stats.Endpoints[0].NumRequests)
 			}
-			if stats.NumErrors != 1 {
-				t.Errorf("Unexpected num_errors; want: 1; got: %d", stats.NumErrors)
+			if stats.Endpoints[0].NumErrors != 1 {
+				t.Errorf("Unexpected num_errors; want: 1; got: %d", stats.Endpoints[0].NumErrors)
 			}
 			if test.expectedStats != nil {
 				var data map[string]interface{}
-				if err := json.Unmarshal(stats.Data, &data); err != nil {
+				if err := json.Unmarshal(stats.Endpoints[0].Data, &data); err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
 				if !reflect.DeepEqual(data, test.expectedStats) {
@@ -1007,6 +1180,9 @@ func TestRequestRespond(t *testing.T) {
 				if val := req.Headers().Get("key"); val != "value" {
 					t.Fatalf("Expected headers in the request")
 				}
+				if !bytes.Equal(req.Data(), []byte("req")) {
+					t.Fatalf("Invalid request data; want: %q; got: %q", "req", req.Data())
+				}
 				if errCode == "" && errDesc == "" {
 					if resp, ok := respData.([]byte); ok {
 						err := req.Respond(resp, micro.WithHeaders(test.respondHeaders))
@@ -1050,8 +1226,8 @@ func TestRequestRespond(t *testing.T) {
 				Name:        "CoolService",
 				Version:     "0.1.0",
 				Description: "test service",
-				Endpoint: micro.Endpoint{
-					Subject: "svc.test",
+				Endpoint: &micro.EndpointConfig{
+					Subject: "test.func",
 					Handler: micro.HandlerFunc(handler),
 				},
 			})
@@ -1061,8 +1237,8 @@ func TestRequestRespond(t *testing.T) {
 			defer svc.Stop()
 
 			resp, err := nc.RequestMsg(&nats.Msg{
-				Subject: svc.Info().Subject,
-				Data:    nil,
+				Subject: "test.func",
+				Data:    []byte("req"),
 				Header:  nats.Header{"key": []string{"value"}},
 			}, 50*time.Millisecond)
 			if test.withRespondError != nil {
diff --git a/nats_test.go b/nats_test.go
index f1bfabf..33b2df4 100644
--- a/nats_test.go
+++ b/nats_test.go
@@ -2967,3 +2967,46 @@ func TestServerListWithTrailingComma(t *testing.T) {
 		t.Fatalf("There should be only 1 URL in the list, got %v", l)
 	}
 }
+
+func BenchmarkHeaderDecode(b *testing.B) {
+	benchmarks := []struct {
+		name   string
+		header Header
+	}{
+		{"Small - 25", Header{
+			"Msg-ID": []string{"123"}},
+		},
+		{"Medium - 141", Header{
+			"CorrelationID": []string{"123"},
+			"Msg-ID":        []string{"456"},
+			"X-NATS-Keys":   []string{"A", "B", "C"},
+			"X-Test-Keys":   []string{"D", "E", "F"},
+		}},
+		{"Large - 368", Header{
+			"CorrelationID":     []string{"123"},
+			"Msg-ID":            []string{"456"},
+			"X-NATS-Keys":       []string{"A", "B", "C"},
+			"X-Test-Keys":       []string{"D", "E", "F"},
+			"X-A-Long-Header-1": []string{strings.Repeat("A", 100)},
+			"X-A-Long-Header-2": []string{strings.Repeat("A", 100)},
+		}},
+	}
+	for _, bm := range benchmarks {
+		b.Run(bm.name, func(b *testing.B) {
+			b.ReportAllocs()
+
+			m := NewMsg("foo")
+			m.Header = bm.header
+			hdr, err := m.headerBytes()
+			if err != nil {
+				b.Fatalf("Unexpected error: %v", err)
+			}
+
+			for i := 0; i < b.N; i++ {
+				if _, err := decodeHeadersMsg(hdr); err != nil {
+					b.Fatalf("Unexpected error: %v", err)
+				}
+			}
+		})
+	}
+}
diff --git a/test/auth_test.go b/test/auth_test.go
index 32d80e5..b0aa060 100644
--- a/test/auth_test.go
+++ b/test/auth_test.go
@@ -16,6 +16,7 @@ package test
 import (
 	"errors"
 	"fmt"
+	"io/fs"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -364,3 +365,15 @@ func TestPermViolation(t *testing.T) {
 		t.Fatal("Connection should be not be closed")
 	}
 }
+
+func TestConnectMissingCreds(t *testing.T) {
+	s := RunServerOnPort(-1)
+	defer s.Shutdown()
+
+	// Using TEST-NET sample address from RFC 5737.
+	testnetaddr := "nats://192.0.2.2:4222"
+	_, err := nats.Connect(fmt.Sprintf("%s,%s", s.ClientURL(), testnetaddr), nats.UserCredentials("missing"), nats.DontRandomize())
+	if !errors.Is(err, fs.ErrNotExist) {
+		t.Fatalf("Expected not exists error, got: %v", err)
+	}
+}