func (s *Server) NumClients() int {
	s.mtx.RLock()
	defer s.mtx.RUnlock()
	return len(s.subscriptions)
}
NumClients returns the number of clients.
package pubsub
import (
	"context"
	"errors"
	"fmt"
	"github.com/tendermint/tendermint/abci/types"
	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
	"github.com/tendermint/tendermint/libs/pubsub/query"
	"github.com/tendermint/tendermint/libs/service"
)
type operation int
const (
	sub operation = iota
	pub
	unsub
	shutdown
)
var (
	ErrSubscriptionNotFound = errors.New("subscription not found")
	ErrAlreadySubscribed = errors.New("already subscribed")
)
type Query interface {
	Matches(events []types.Event) (bool, error)
	String() string
}
type UnsubscribeArgs struct {
	ID         string
	Subscriber string
	Query      Query
}
func (args UnsubscribeArgs) Validate() error {
	if args.Subscriber == "" {
		return errors.New("must specify a subscriber")
	}
	if args.ID == "" && args.Query == nil {
		return fmt.Errorf("subscription is not fully defined [subscriber=%q]", args.Subscriber)
	}
	return nil
}
type cmd struct {
	op operation
	query        Query
	subscription *Subscription
	clientID     string
	msg    interface{}
	events []types.Event
}
type Server struct {
	service.BaseService
	cmds    chan cmd
	cmdsCap int
	mtx tmsync.RWMutex
	subscriptions map[string]map[string]string
}
type Option func(*Server)
func NewServer(options ...Option) *Server {
	s := &Server{
		subscriptions: make(map[string]map[string]string),
	}
	s.BaseService = *service.NewBaseService(nil, "PubSub", s)
	for _, option := range options {
		option(s)
	}
	s.cmds = make(chan cmd, s.cmdsCap)
	return s
}
func BufferCapacity(cap int) Option {
	return func(s *Server) {
		if cap > 0 {
			s.cmdsCap = cap
		}
	}
}
func (s *Server) BufferCapacity() int {
	return s.cmdsCap
}
func (s *Server) Subscribe(
	ctx context.Context,
	clientID string,
	query Query,
	outCapacity ...int) (*Subscription, error) {
	outCap := 1
	if len(outCapacity) > 0 {
		if outCapacity[0] <= 0 {
			panic("Negative or zero capacity. Use SubscribeUnbuffered if you want an unbuffered channel")
		}
		outCap = outCapacity[0]
	}
	return s.subscribe(ctx, clientID, query, outCap)
}
func (s *Server) SubscribeUnbuffered(ctx context.Context, clientID string, query Query) (*Subscription, error) {
	return s.subscribe(ctx, clientID, query, 0)
}
func (s *Server) subscribe(ctx context.Context, clientID string, query Query, outCapacity int) (*Subscription, error) {
	s.mtx.RLock()
	clientSubscriptions, ok := s.subscriptions[clientID]
	if ok {
		_, ok = clientSubscriptions[query.String()]
	}
	s.mtx.RUnlock()
	if ok {
		return nil, ErrAlreadySubscribed
	}
	subscription := NewSubscription(outCapacity)
	select {
	case s.cmds <- cmd{op: sub, clientID: clientID, query: query, subscription: subscription}:
		s.mtx.Lock()
		if _, ok = s.subscriptions[clientID]; !ok {
			s.subscriptions[clientID] = make(map[string]string)
		}
		s.subscriptions[clientID][query.String()] = subscription.id
		s.subscriptions[clientID][subscription.id] = query.String()
		s.mtx.Unlock()
		return subscription, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	case <-s.Quit():
		return nil, nil
	}
}
func (s *Server) Unsubscribe(ctx context.Context, args UnsubscribeArgs) error {
	if err := args.Validate(); err != nil {
		return err
	}
	var qs string
	if args.Query != nil {
		qs = args.Query.String()
	}
	clientSubscriptions, err := func() (map[string]string, error) {
		s.mtx.RLock()
		defer s.mtx.RUnlock()
		clientSubscriptions, ok := s.subscriptions[args.Subscriber]
		if args.ID != "" {
			qs, ok = clientSubscriptions[args.ID]
			if ok && args.Query == nil {
				var err error
				args.Query, err = query.New(qs)
				if err != nil {
					return nil, err
				}
			}
		} else if qs != "" {
			args.ID, ok = clientSubscriptions[qs]
		}
		if !ok {
			return nil, ErrSubscriptionNotFound
		}
		return clientSubscriptions, nil
	}()
	if err != nil {
		return err
	}
	select {
	case s.cmds <- cmd{op: unsub, clientID: args.Subscriber, query: args.Query, subscription: &Subscription{id: args.ID}}:
		s.mtx.Lock()
		defer s.mtx.Unlock()
		delete(clientSubscriptions, args.ID)
		delete(clientSubscriptions, qs)
		if len(clientSubscriptions) == 0 {
			delete(s.subscriptions, args.Subscriber)
		}
		return nil
	case <-ctx.Done():
		return ctx.Err()
	case <-s.Quit():
		return nil
	}
}
func (s *Server) UnsubscribeAll(ctx context.Context, clientID string) error {
	s.mtx.RLock()
	_, ok := s.subscriptions[clientID]
	s.mtx.RUnlock()
	if !ok {
		return ErrSubscriptionNotFound
	}
	select {
	case s.cmds <- cmd{op: unsub, clientID: clientID}:
		s.mtx.Lock()
		defer s.mtx.Unlock()
		delete(s.subscriptions, clientID)
		return nil
	case <-ctx.Done():
		return ctx.Err()
	case <-s.Quit():
		return nil
	}
}