func (s *Service) TokenEvent(cid string, token interface{}) {
	if atomic.LoadInt32(&s.state) != stateStarted {
		s.errorf("Failed to send token event: service not started")
		return
	}
	if !isValidPart(cid) {
		panic(`res: invalid connection ID`)
	}
	s.event("conn."+cid+".token", tokenEvent{Token: token})
}
TokenEvent sends a connection token event that sets the connection's access
token, discarding any previously set token.

A change of token will invalidate any previous access response received using
the old token.

A nil token clears any previously set token.
package res
import (
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/jirenius/go-res/logger"
	"github.com/jirenius/timerqueue"
	nats "github.com/nats-io/nats.go"
)
const protocolVersion = "1.2.2"
const inChannelSize = 256
const workerCount = 32
const defaultQueryEventDuration = time.Second * 3
var (
	errNotStopped = errors.New("res: service is not stopped")
	errNotStarted = errors.New("res: service is not started")
)
type Option interface{ SetOption(*Handler) }
type OptionFunc func(*Handler)
func (f OptionFunc) SetOption(hs *Handler) { f(hs) }
type AccessHandler func(AccessRequest)
type GetHandler func(GetRequest)
type ModelHandler func(ModelRequest)
type CollectionHandler func(CollectionRequest)
type CallHandler func(CallRequest)
type NewHandler func(NewRequest)
type AuthHandler func(AuthRequest)
type ApplyChangeHandler func(r Resource, changes map[string]interface{}) (map[string]interface{}, error)
type ApplyAddHandler func(r Resource, value interface{}, idx int) error
type ApplyRemoveHandler func(r Resource, idx int) (interface{}, error)
type ApplyCreateHandler func(r Resource, data interface{}) error
type ApplyDeleteHandler func(r Resource) (interface{}, error)
type Handler struct {
	Type ResourceType
	Access AccessHandler
	Get GetHandler
	Call map[string]CallHandler
	New NewHandler
	Auth map[string]AuthHandler
	ApplyChange ApplyChangeHandler
	ApplyAdd ApplyAddHandler
	ApplyRemove ApplyRemoveHandler
	ApplyCreate ApplyCreateHandler
	ApplyDelete ApplyDeleteHandler
	Group string
	OnRegister func(service *Service, pattern Pattern, rh Handler)
	Listeners map[string]func(*Event)
}
const (
	stateStopped = iota
	stateStarting
	stateStarted
	stateStopping
)
var (
	Model = OptionFunc(func(hs *Handler) {
		if hs.Type != TypeUnset {
			panic("res: resource type set multiple times")
		}
		hs.Type = TypeModel
	})
	Collection = OptionFunc(func(hs *Handler) {
		if hs.Type != TypeUnset {
			panic("res: resource type set multiple times")
		}
		hs.Type = TypeCollection
	})
)
func (h *Handler) Option(hf ...Option) {
	for _, f := range hf {
		f.SetOption(h)
	}
}
type Service struct {
	*Mux
	state          int32
	nc             Conn                    
	inCh           chan *nats.Msg          
	rwork          map[string]*work        
	workCh         chan *work              
	wg             sync.WaitGroup          
	mu             sync.Mutex              
	logger         logger.Logger           
	queueGroup     string                  
	resetResources []string                
	resetAccess    []string                
	queryTQ        *timerqueue.Queue       
	queryDuration  time.Duration           
	onServe        func(*Service)          
	onDisconnect   func(*Service)          
	onReconnect    func(*Service)          
	onError        func(*Service, string)  
}
func NewService(name string) *Service {
	s := &Service{
		Mux:           NewMux(name),
		queueGroup:    name,
		logger:        logger.NewStdLogger(),
		queryDuration: defaultQueryEventDuration,
	}
	s.Mux.Register(s)
	return s
}
func (s *Service) SetLogger(l logger.Logger) *Service {
	if s.nc != nil {
		panic("res: service already started")
	}
	s.logger = l
	return s
}
func (s *Service) SetQueryEventDuration(d time.Duration) *Service {
	if s.nc != nil {
		panic("res: service already started")
	}
	s.queryDuration = d
	return s
}
func (s *Service) SetQueueGroup(queue string) *Service {
	s.queueGroup = queue
	return s
}
func (s *Service) SetOnServe(f func(*Service)) {
	s.onServe = f
}
func (s *Service) SetOnDisconnect(f func(*Service)) {
	s.onDisconnect = f
}
func (s *Service) SetOnReconnect(f func(*Service)) {
	s.onReconnect = f
}
func (s *Service) SetOnError(f func(*Service, string)) {
	s.onError = f
}
func (s *Service) Logger() logger.Logger {
	return s.logger
}
func (s *Service) ProtocolVersion() string {
	return protocolVersion
}
func (s *Service) Conn() Conn {
	return s.nc
}
func (s *Service) infof(format string, v ...interface{}) {
	if s.logger == nil {
		return
	}
	s.logger.Infof(format, v...)
}
func (s *Service) errorf(format string, v ...interface{}) {
	if s.logger == nil {
		return
	}
	s.logger.Errorf(format, v...)
	if s.onError != nil {
		s.onError(s, fmt.Sprintf(format, v...))
	}
}
func (s *Service) tracef(format string, v ...interface{}) {
	if s.logger == nil {
		return
	}
	s.logger.Tracef(format, v...)
}
func Access(h AccessHandler) Option {
	return OptionFunc(func(hs *Handler) {
		if hs.Access != nil {
			panic("res: multiple access handlers")
		}
		hs.Access = h
	})
}
func GetModel(h ModelHandler) Option {
	return OptionFunc(func(hs *Handler) {
		Model(hs)
		validateGetHandler(*hs)
		hs.Get = func(r GetRequest) { h(ModelRequest(r)) }
	})
}
func GetCollection(h CollectionHandler) Option {
	return OptionFunc(func(hs *Handler) {
		Collection(hs)
		validateGetHandler(*hs)
		hs.Get = func(r GetRequest) { h(CollectionRequest(r)) }
	})
}
func GetResource(h GetHandler) Option {
	return OptionFunc(func(hs *Handler) {
		validateGetHandler(*hs)
		hs.Get = h
	})
}
func Call(method string, h CallHandler) Option {
	if method != "*" && !isValidPart(method) {
		panic("res: invalid method name: " + method)
	}
	return OptionFunc(func(hs *Handler) {
		if hs.Call == nil {
			hs.Call = make(map[string]CallHandler)
		}
		if _, ok := hs.Call[method]; ok {
			panic("res: multiple call handlers for method " + method)
		}
		hs.Call[method] = h
	})
}
func Set(h CallHandler) Option {
	return Call("set", h)
}
func New(h NewHandler) Option {
	return OptionFunc(func(hs *Handler) {
		if hs.New != nil {
			panic("res: multiple new handlers")
		}
		hs.New = h
	})
}
func Auth(method string, h AuthHandler) Option {
	if method != "*" && !isValidPart(method) {
		panic("res: invalid method name: " + method)
	}
	return OptionFunc(func(hs *Handler) {
		if hs.Auth == nil {
			hs.Auth = make(map[string]AuthHandler)
		}
		if _, ok := hs.Auth[method]; ok {
			panic("res: multiple auth handlers for method " + method)
		}
		hs.Auth[method] = h
	})
}
func ApplyChange(h ApplyChangeHandler) Option {
	return OptionFunc(func(hs *Handler) {
		if hs.ApplyChange != nil {
			panic("res: multiple apply change handlers")
		}
		hs.ApplyChange = h
	})
}
func ApplyAdd(h ApplyAddHandler) Option {
	return OptionFunc(func(hs *Handler) {
		if hs.ApplyAdd != nil {
			panic("res: multiple apply add handlers")
		}
		hs.ApplyAdd = h
	})
}
func ApplyRemove(h ApplyRemoveHandler) Option {
	return OptionFunc(func(hs *Handler) {
		if hs.ApplyRemove != nil {
			panic("res: multiple apply remove handlers")
		}
		hs.ApplyRemove = h
	})
}
func ApplyCreate(h ApplyCreateHandler) Option {
	return OptionFunc(func(hs *Handler) {
		if hs.ApplyCreate != nil {
			panic("res: multiple apply create handlers")
		}
		hs.ApplyCreate = h
	})
}
func ApplyDelete(h ApplyDeleteHandler) Option {
	return OptionFunc(func(hs *Handler) {
		if hs.ApplyDelete != nil {
			panic("res: multiple apply delete handlers")
		}
		hs.ApplyDelete = h
	})
}
func Group(group string) Option {
	return OptionFunc(func(hs *Handler) {
		hs.Group = group
	})
}
func OnRegister(callback func(service *Service, pattern Pattern, rh Handler)) Option {
	return OptionFunc(func(hs *Handler) {
		if hs.OnRegister != nil {
			prevcb := hs.OnRegister
			hs.OnRegister = func(service *Service, pattern Pattern, rh Handler) {
				prevcb(service, pattern, rh)
				callback(service, pattern, rh)
			}
		} else {
			hs.OnRegister = callback
		}
	})
}
func (s *Service) SetReset(resources, access []string) *Service {
	return s.SetOwnedResources(resources, access)
}
func (s *Service) SetOwnedResources(resources, access []string) *Service {
	s.resetResources = resources
	s.resetAccess = access
	return s
}
func (s *Service) ListenAndServe(url string, options ...nats.Option) error {
	if !atomic.CompareAndSwapInt32(&s.state, stateStopped, stateStarting) {
		return errNotStopped
	}
	opts := []nats.Option{
		nats.MaxReconnects(-1),
		nats.ReconnectHandler(s.handleReconnect),
		nats.DisconnectHandler(s.handleDisconnect),
		nats.ClosedHandler(s.handleClosed),
	}
	if s.Mux.path != "" {
		opts = append(opts, nats.Name(s.Mux.path))
	}
	opts = append(opts, options...)
	s.infof("Connecting to NATS server")
	nc, err := nats.Connect(url, opts...)
	if err != nil {
		s.errorf("Failed to connect to NATS server: %s", err)
		return err
	}
	return s.serve(nc)
}
func (s *Service) Serve(conn Conn) error {
	if !atomic.CompareAndSwapInt32(&s.state, stateStopped, stateStarting) {
		return errNotStopped
	}
	if nc, ok := conn.(*nats.Conn); ok {
		nc.SetReconnectHandler(s.handleReconnect)
		nc.SetDisconnectHandler(s.handleDisconnect)
		nc.SetClosedHandler(s.handleClosed)
	}
	return s.serve(conn)
}
func (s *Service) serve(nc Conn) error {
	s.infof("Starting service")
	err := s.ValidateListeners()
	if err != nil {
		return err
	}
	inCh := make(chan *nats.Msg, inChannelSize)
	workCh := make(chan *work, 1)
	s.nc = nc
	s.inCh = inCh
	s.workCh = workCh
	s.rwork = make(map[string]*work)
	s.queryTQ = timerqueue.New(s.queryEventExpire, s.queryDuration)
	s.wg.Add(workerCount)
	for i := 0; i < workerCount; i++ {
		go s.startWorker(s.workCh)
	}
	atomic.StoreInt32(&s.state, stateStarted)
	err = s.subscribe()
	if err != nil {
		s.errorf("Failed to subscribe: %s", err)
		go s.Shutdown()
	} else {
		s.ResetAll()
		if s.onServe != nil {
			s.onServe(s)
		}
		s.infof("Listening for requests")
		s.startListener(inCh)
	}
	close(workCh)
	s.wg.Wait()
	return nil
}
func (s *Service) Shutdown() error {
	if !atomic.CompareAndSwapInt32(&s.state, stateStarted, stateStopping) {
		return errNotStarted
	}
	s.infof("Stopping service...")
	s.close()
	s.wg.Wait()
	s.inCh = nil
	s.nc = nil
	s.workCh = nil
	atomic.StoreInt32(&s.state, stateStopped)
	s.infof("Stopped")
	return nil
}
func (s *Service) close() {
	s.nc.Close()
	close(s.inCh)
}
func (s *Service) Reset(resources []string, access []string) {
	if atomic.LoadInt32(&s.state) != stateStarted {
		s.errorf("Failed to reset: service not started")
		return
	}
	s.reset(resources, access)
}
func (s *Service) reset(resources []string, access []string) {
	lr := len(resources)
	la := len(access)
	if lr == 0 && la == 0 {
		return
	}
	if lr == 0 {
		resources = nil
	}
	if la == 0 {
		access = nil
	}
	s.event("system.reset", resetEvent{
		Resources: resources,
		Access:    access,
	})
}
func (s *Service) ResetAll() {
	if atomic.LoadInt32(&s.state) != stateStarted {
		s.errorf("Failed to reset: service not started")
		return
	}
	s.setDefaultOwnership()
	s.reset(s.resetResources, s.resetAccess)
}