func (m *MultiClient) GetSubscribersCount(topic string, subscriberHashPrefix []byte) (int, error) {
	return m.GetSubscribersCountContext(context.Background(), topic, subscriberHashPrefix)
}
GetSubscribersCount wraps GetSubscribersCountContext with background context.
package nkn
import (
	"context"
	"log"
	"net"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
	multierror "github.com/hashicorp/go-multierror"
	ncp "github.com/nknorg/ncp-go"
	"github.com/nknorg/nkn-sdk-go/payloads"
	"github.com/nknorg/nkn/v2/transaction"
	"github.com/nknorg/nkn/v2/util/address"
	"github.com/patrickmn/go-cache"
)
const (
	MultiClientIdentifierRe = "^__\\d+__$"
	DefaultSessionAllowAddr = ".*"
	SessionIDSize = MessageIDSize
	acceptSessionBufSize = 128
)
var (
	multiClientIdentifierRe = regexp.MustCompile(MultiClientIdentifierRe)
)
func addMultiClientPrefix(dest []string, clientID int) []string {
	result := make([]string, len(dest))
	for i, addr := range dest {
		result[i] = addIdentifier(addr, clientID)
	}
	return result
}
func addIdentifier(addr string, id int) string {
	if id < 0 {
		return addr
	}
	return addIdentifierPrefix(addr, "__"+strconv.Itoa(id)+"__")
}
func removeIdentifier(src string) (string, string) {
	s := strings.SplitN(src, ".", 2)
	if len(s) > 1 {
		if multiClientIdentifierRe.MatchString(s[0]) {
			return s[1], s[0]
		}
	}
	return src, ""
}
type MultiClient struct {
	OnConnect *OnConnect  
	OnMessage *OnMessage  
	config        *ClientConfig
	offset        int
	addr          *ClientAddr
	acceptSession chan *ncp.Session
	onClose       chan struct{}
	msgCache      *cache.Cache
	lock          sync.RWMutex
	clients       map[int]*Client
	defaultClient *Client
	acceptAddrs   []*regexp.Regexp
	isClosed      bool
	createDone    bool
	sessionLock sync.Mutex
	sessions    map[string]*ncp.Session
}
func NewMultiClient(account *Account, baseIdentifier string, numSubClients int, originalClient bool, config *ClientConfig) (*MultiClient, error) {
	config, err := MergeClientConfig(config)
	if err != nil {
		return nil, err
	}
	numClients := numSubClients
	offset := 0
	if originalClient {
		numClients++
		offset = 1
	}
	addr := address.MakeAddressString(account.PublicKey, baseIdentifier)
	m := &MultiClient{
		config:        config,
		offset:        offset,
		addr:          NewClientAddr(addr),
		OnConnect:     NewOnConnect(1, nil),
		OnMessage:     NewOnMessage(int(config.MsgChanLen), nil),
		acceptSession: make(chan *ncp.Session, acceptSessionBufSize),
		onClose:       make(chan struct{}, 0),
		msgCache:      cache.New(time.Duration(config.MsgCacheExpiration)*time.Millisecond, time.Duration(config.MsgCacheCleanupInterval)*time.Millisecond),
		clients:       make(map[int]*Client, numClients),
		defaultClient: nil,
		sessions:      make(map[string]*ncp.Session, 0),
	}
	var wg sync.WaitGroup
	defaultClientIdx := numSubClients
	success := make(chan struct{}, 1)
	fail := make(chan struct{}, 1)
	for i := -offset; i < numSubClients; i++ {
		wg.Add(1)
		go func(i int) {
			client, err := NewClient(account, addIdentifier(baseIdentifier, i), config)
			if err != nil {
				log.Println(err)
				wg.Done()
				return
			}
			m.lock.Lock()
			if m.isClosed {
				err := client.Close()
				if err != nil {
					log.Println(err)
				}
				wg.Done()
				m.lock.Unlock()
				return
			}
			m.clients[i] = client
			if i < defaultClientIdx {
				m.defaultClient = client
				defaultClientIdx = i
			}
			m.lock.Unlock()
			select {
			case success <- struct{}{}:
			default:
			}
			wg.Done()
			node, ok := <-client.OnConnect.C
			if !ok {
				return
			}
			m.lock.RLock()
			if m.isClosed {
				m.lock.RUnlock()
				return
			}
			m.OnConnect.receive(node)
			m.lock.RUnlock()
			for {
				select {
				case msg, ok := <-client.OnMessage.C:
					if !ok {
						return
					}
					if msg.Type == SessionType {
						if !msg.Encrypted {
							continue
						}
						err := m.handleSessionMsg(addIdentifier("", i-offset), msg.Src, msg.MessageID, msg.Data)
						if err != nil {
							if err != ncp.ErrSessionClosed && err != ErrAddrNotAllowed {
								log.Println(err)
							}
							continue
						}
					} else {
						cacheKey := string(msg.MessageID)
						if _, ok := m.msgCache.Get(cacheKey); ok {
							continue
						}
						m.msgCache.Set(cacheKey, nil, cache.DefaultExpiration)
						msg.Src, _ = removeIdentifier(msg.Src)
						if msg.NoReply {
							msg.reply = func(data interface{}) error {
								return nil
							}
						} else {
							msg.reply = func(data interface{}) error {
								payload, err := newReplyPayload(data, msg.MessageID)
								if err != nil {
									return err
								}
								if err := m.send([]string{msg.Src}, payload, msg.Encrypted, 0); err != nil {
									return err
								}
								return nil
							}
						}
						m.lock.RLock()
						if m.isClosed {
							m.lock.RUnlock()
							return
						}
						m.OnMessage.receive(msg, true)
						m.lock.RUnlock()
					}
				case <-m.onClose:
					return
				}
			}
		}(i)
	}
	go func() {
		wg.Wait()
		m.lock.Lock()
		m.createDone = true
		m.lock.Unlock()
		select {
		case fail <- struct{}{}:
		default:
		}
	}()
	select {
	case <-success:
		return m, nil
	case <-fail:
		return nil, ErrCreateClientFailed
	}
}
func (m *MultiClient) Account() *Account {
	return m.GetDefaultClient().account
}
func (m *MultiClient) Seed() []byte {
	return m.GetDefaultClient().Seed()
}
func (m *MultiClient) PubKey() []byte {
	return m.GetDefaultClient().PubKey()
}
func (m *MultiClient) Address() string {
	return m.addr.String()
}
func (m *MultiClient) Addr() net.Addr {
	return m.addr
}
func (m *MultiClient) GetClients() map[int]*Client {
	m.lock.RLock()
	defer m.lock.RUnlock()
	if m.createDone {
		return m.clients
	}
	clients := make(map[int]*Client, len(m.clients))
	for i, client := range m.clients {
		clients[i] = client
	}
	return clients
}
func (m *MultiClient) GetClient(i int) *Client {
	m.lock.RLock()
	defer m.lock.RUnlock()
	return m.clients[i]
}
func (m *MultiClient) GetDefaultClient() *Client {
	m.lock.RLock()
	defer m.lock.RUnlock()
	return m.defaultClient
}
func (m *MultiClient) SendWithClient(clientID int, dests *StringArray, data interface{}, config *MessageConfig) (*OnMessage, error) {
	client := m.GetClient(clientID)
	if client == nil {
		return nil, ErrNilClient
	}
	config, err := MergeMessageConfig(m.config.MessageConfig, config)
	if err != nil {
		return nil, err
	}
	payload, err := newMessagePayload(data, config.MessageID, config.NoReply)
	if err != nil {
		return nil, err
	}
	if err := m.sendWithClient(clientID, dests.Elems(), payload, !config.Unencrypted, config.MaxHoldingSeconds); err != nil {
		return nil, err
	}
	onReply := NewOnMessage(1, nil)
	if !config.NoReply {
		client.responseChannels.Add(string(payload.MessageId), onReply, cache.DefaultExpiration)
	}
	return onReply, nil
}
func (m *MultiClient) SendBinaryWithClient(clientID int, dests *StringArray, data []byte, config *MessageConfig) (*OnMessage, error) {
	return m.SendWithClient(clientID, dests, data, config)
}
func (m *MultiClient) SendTextWithClient(clientID int, dests *StringArray, data string, config *MessageConfig) (*OnMessage, error) {
	return m.SendWithClient(clientID, dests, data, config)
}
func (m *MultiClient) sendWithClient(clientID int, dests []string, payload *payloads.Payload, encrypted bool, maxHoldingSeconds int32) error {
	client := m.GetClient(clientID)
	if client == nil {
		return ErrNilClient
	}
	return client.send(addMultiClientPrefix(dests, clientID), payload, encrypted, maxHoldingSeconds)
}
func (m *MultiClient) Send(dests *StringArray, data interface{}, config *MessageConfig) (*OnMessage, error) {
	config, err := MergeMessageConfig(m.config.MessageConfig, config)
	if err != nil {
		return nil, err
	}
	payload, err := newMessagePayload(data, config.MessageID, config.NoReply)
	if err != nil {
		return nil, err
	}
	var lock sync.Mutex
	var errs error
	var onRawReply *OnMessage
	onReply := NewOnMessage(1, nil)
	clients := m.GetClients()
	if !config.NoReply {
		onRawReply = NewOnMessage(1, nil)
		for _, client := range clients {
			client.responseChannels.Add(string(payload.MessageId), onRawReply, cache.DefaultExpiration)
		}
	}
	success := make(chan struct{}, 1)
	fail := make(chan struct{}, 1)
	go func() {
		sent := 0
		for clientID := range clients {
			err := m.sendWithClient(clientID, dests.Elems(), payload, !config.Unencrypted, config.MaxHoldingSeconds)
			if err == nil {
				select {
				case success <- struct{}{}:
				default:
				}
				sent++
			} else {
				lock.Lock()
				errs = multierror.Append(errs, err)
				lock.Unlock()
			}
		}
		if sent == 0 {
			select {
			case fail <- struct{}{}:
			default:
			}
		}
		if !config.NoReply {
			msg := <-onRawReply.C
			msg.Src, _ = removeIdentifier(msg.Src)
			onReply.receive(msg, false)
		}
	}()
	select {
	case <-success:
		return onReply, nil
	case <-fail:
		return nil, errs
	}
}
func (m *MultiClient) SendBinary(dests *StringArray, data []byte, config *MessageConfig) (*OnMessage, error) {
	return m.Send(dests, data, config)
}
func (m *MultiClient) SendText(dests *StringArray, data string, config *MessageConfig) (*OnMessage, error) {
	return m.Send(dests, data, config)
}
func (m *MultiClient) send(dests []string, payload *payloads.Payload, encrypted bool, maxHoldingSeconds int32) error {
	var lock sync.Mutex
	var errs error
	success := make(chan struct{}, 1)
	fail := make(chan struct{}, 1)
	go func() {
		sent := 0
		for clientID := range m.GetClients() {
			err := m.sendWithClient(clientID, dests, payload, encrypted, maxHoldingSeconds)
			if err == nil {
				select {
				case success <- struct{}{}:
				default:
				}
				sent++
			} else {
				lock.Lock()
				errs = multierror.Append(errs, err)
				lock.Unlock()
			}
		}
		if sent == 0 {
			select {
			case fail <- struct{}{}:
			default:
			}
		}
	}()
	select {
	case <-success:
		return nil
	case <-fail:
		return errs
	}
}
func (m *MultiClient) Publish(topic string, data interface{}, config *MessageConfig) error {
	return publish(m, topic, data, config)
}
func (m *MultiClient) PublishBinary(topic string, data []byte, config *MessageConfig) error {
	return m.Publish(topic, data, config)
}
func (m *MultiClient) PublishText(topic string, data string, config *MessageConfig) error {
	return m.Publish(topic, data, config)
}
func (m *MultiClient) newSession(remoteAddr string, sessionID []byte, config *ncp.Config) (*ncp.Session, error) {
	rawClients := m.GetClients()
	clientIDs := make([]string, 0, len(rawClients))
	clients := make(map[string]*Client, len(rawClients))
	for id, client := range rawClients {
		clientID := addIdentifier("", id)
		clientIDs = append(clientIDs, clientID)
		clients[clientID] = client
	}
	sort.Strings(clientIDs)
	return ncp.NewSession(m.addr, NewClientAddr(remoteAddr), clientIDs, nil, (func(localClientID, remoteClientID string, buf []byte, writeTimeout time.Duration) error {
		payload := &payloads.Payload{
			Type:      payloads.PayloadType_SESSION,
			MessageId: sessionID,
			Data:      buf,
		}
		client := clients[localClientID]
		err := client.sendTimeout([]string{addIdentifierPrefix(remoteAddr, remoteClientID)}, payload, true, 0, writeTimeout)
		if err != nil {
			return err
		}
		return nil
	}), config)
}
func (m *MultiClient) shouldAcceptAddr(addr string) bool {
	for _, allowAddr := range m.acceptAddrs {
		if allowAddr.MatchString(addr) {
			return true
		}
	}
	return false
}
func (m *MultiClient) handleSessionMsg(localClientID, src string, sessionID, data []byte) error {
	remoteAddr, remoteClientID := removeIdentifier(src)
	sessionKey := sessionKey(remoteAddr, sessionID)
	var err error
	m.sessionLock.Lock()
	session, ok := m.sessions[sessionKey]
	if !ok {
		if !m.shouldAcceptAddr(remoteAddr) {
			m.sessionLock.Unlock()
			return ErrAddrNotAllowed
		}
		session, err = m.newSession(remoteAddr, sessionID, m.config.SessionConfig)
		if err != nil {
			m.sessionLock.Unlock()
			return err
		}
		m.sessions[sessionKey] = session
	}
	m.sessionLock.Unlock()
	err = session.ReceiveWith(localClientID, remoteClientID, data)
	if err != nil {
		return err
	}
	if !ok {
		select {
		case m.acceptSession <- session:
		default:
			log.Println("Accept session channel full, discard request...")
		}
	}
	return nil
}
func (m *MultiClient) Listen(addrsRe *StringArray) error {
	var addrs []string
	if addrsRe == nil {
		addrs = []string{DefaultSessionAllowAddr}
	} else {
		addrs = addrsRe.Elems()
	}
	var err error
	acceptAddrs := make([]*regexp.Regexp, len(addrs))
	for i := 0; i < len(acceptAddrs); i++ {
		acceptAddrs[i], err = regexp.Compile(addrs[i])
		if err != nil {
			return err
		}
	}
	m.lock.Lock()
	m.acceptAddrs = acceptAddrs
	m.lock.Unlock()
	return nil
}
func (m *MultiClient) Dial(remoteAddr string) (net.Conn, error) {
	return m.DialSession(remoteAddr)
}
func (m *MultiClient) DialSession(remoteAddr string) (*ncp.Session, error) {
	return m.DialWithConfig(remoteAddr, nil)
}
func (m *MultiClient) DialWithConfig(remoteAddr string, config *DialConfig) (*ncp.Session, error) {
	config, err := MergeDialConfig(m.config.SessionConfig, config)
	if err != nil {
		return nil, err
	}
	sessionID, err := RandomBytes(SessionIDSize)
	if err != nil {
		return nil, err
	}
	sessionKey := sessionKey(remoteAddr, sessionID)
	session, err := m.newSession(remoteAddr, sessionID, config.SessionConfig)
	if err != nil {
		return nil, err
	}
	m.sessionLock.Lock()
	m.sessions[sessionKey] = session
	m.sessionLock.Unlock()
	ctx := context.Background()
	var cancel context.CancelFunc
	if config.DialTimeout > 0 {
		ctx, cancel = context.WithTimeout(ctx, time.Duration(config.DialTimeout)*time.Millisecond)
		defer cancel()
	}
	err = session.Dial(ctx)
	if err != nil {
		return nil, err
	}
	return session, nil
}
func (m *MultiClient) AcceptSession() (*ncp.Session, error) {
	for {
		select {
		case session := <-m.acceptSession:
			err := session.Accept()
			if err != nil {
				log.Println(err)
				continue
			}
			return session, nil
		case <-m.onClose:
			return nil, ErrClosed
		}
	}
}
func (m *MultiClient) Accept() (net.Conn, error) {
	return m.AcceptSession()
}
func (m *MultiClient) Close() error {
	m.lock.Lock()
	defer m.lock.Unlock()
	if m.isClosed {
		return nil
	}
	m.sessionLock.Lock()
	for _, session := range m.sessions {
		err := session.Close()
		if err != nil {
			log.Println(err)
			continue
		}
	}
	m.sessionLock.Unlock()
	time.AfterFunc(time.Duration(m.config.SessionConfig.Linger)*time.Millisecond, func() {
		for _, client := range m.GetClients() {
			err := client.Close()
			if err != nil {
				log.Println(err)
				continue
			}
		}
	})
	m.isClosed = true
	m.OnConnect.close()
	m.OnMessage.close()
	close(m.onClose)
	return nil
}
func (m *MultiClient) IsClosed() bool {
	m.lock.RLock()
	defer m.lock.RUnlock()
	return m.isClosed
}
func (m *MultiClient) Reconnect() {
	m.lock.RLock()
	defer m.lock.RUnlock()
	if m.isClosed {
		return
	}
	for _, client := range m.clients {
		client.Reconnect()
	}
}
func (m *MultiClient) getConfig() *ClientConfig {
	return m.config
}
func (m *MultiClient) SignTransaction(tx *transaction.Transaction) error {
	return m.GetDefaultClient().SignTransaction(tx)
}
func (m *MultiClient) NewNanoPay(recipientAddress, fee string, duration int) (*NanoPay, error) {
	return NewNanoPay(m, m.GetDefaultClient().wallet, recipientAddress, fee, duration)
}
func (m *MultiClient) NewNanoPayClaimer(recipientAddress string, claimIntervalMs int32, minFlushAmount string, onError *OnError) (*NanoPayClaimer, error) {
	if len(recipientAddress) == 0 {
		recipientAddress = m.GetDefaultClient().wallet.address
	}
	return NewNanoPayClaimer(m, recipientAddress, claimIntervalMs, minFlushAmount, onError)
}
func (m *MultiClient) GetNonce(txPool bool) (int64, error) {
	return m.GetNonceContext(context.Background(), txPool)
}
func (m *MultiClient) GetNonceContext(ctx context.Context, txPool bool) (int64, error) {
	return m.GetNonceByAddress(m.GetDefaultClient().wallet.address, txPool)
}
func (m *MultiClient) GetNonceByAddress(address string, txPool bool) (int64, error) {
	return m.GetNonceByAddressContext(context.Background(), address, txPool)
}
func (m *MultiClient) GetNonceByAddressContext(ctx context.Context, address string, txPool bool) (int64, error) {
	for _, c := range m.GetClients() {
		if c.wallet.config.SeedRPCServerAddr.Len() > 0 {
			res, err := GetNonce(address, txPool, c.wallet.config)
			if err == nil {
				return res, err
			}
		}
	}
	return GetNonce(address, txPool, m.config)
}
func (m *MultiClient) GetHeight() (int32, error) {
	return m.GetHeightContext(context.Background())
}
func (m *MultiClient) GetHeightContext(ctx context.Context) (int32, error) {
	for _, c := range m.GetClients() {
		if c.wallet.config.SeedRPCServerAddr.Len() > 0 {
			res, err := GetHeight(c.wallet.config)
			if err == nil {
				return res, err
			}
		}
	}
	return GetHeight(m.config)
}
func (m *MultiClient) Balance() (*Amount, error) {
	return m.BalanceContext(context.Background())
}
func (m *MultiClient) BalanceContext(ctx context.Context) (*Amount, error) {
	return m.BalanceByAddress(m.GetDefaultClient().wallet.address)
}
func (m *MultiClient) BalanceByAddress(address string) (*Amount, error) {
	return m.BalanceByAddressContext(context.Background(), address)
}
func (m *MultiClient) BalanceByAddressContext(ctx context.Context, address string) (*Amount, error) {
	for _, c := range m.GetClients() {
		if c.wallet.config.SeedRPCServerAddr.Len() > 0 {
			res, err := GetBalance(address, c.wallet.config)
			if err == nil {
				return res, err
			}
		}
	}
	return GetBalance(address, m.config)
}
func (m *MultiClient) GetSubscribers(topic string, offset, limit int, meta, txPool bool, subscriberHashPrefix []byte) (*Subscribers, error) {
	return m.GetSubscribersContext(context.Background(), topic, offset, limit, meta, txPool, subscriberHashPrefix)
}
func (m *MultiClient) GetSubscribersContext(ctx context.Context, topic string, offset, limit int, meta, txPool bool, subscriberHashPrefix []byte) (*Subscribers, error) {
	for _, c := range m.GetClients() {
		if c.wallet.config.SeedRPCServerAddr.Len() > 0 {
			res, err := GetSubscribers(topic, offset, limit, meta, txPool, subscriberHashPrefix, c.wallet.config)
			if err == nil {
				return res, err
			}
		}
	}
	return GetSubscribers(topic, offset, limit, meta, txPool, subscriberHashPrefix, m.config)
}
func (m *MultiClient) GetSubscription(topic string, subscriber string) (*Subscription, error) {
	return m.GetSubscriptionContext(context.Background(), topic, subscriber)
}
func (m *MultiClient) GetSubscriptionContext(ctx context.Context, topic string, subscriber string) (*Subscription, error) {
	for _, c := range m.GetClients() {
		if c.wallet.config.SeedRPCServerAddr.Len() > 0 {
			res, err := GetSubscription(topic, subscriber, c.wallet.config)
			if err == nil {
				return res, err
			}
		}
	}
	return GetSubscription(topic, subscriber, m.config)
}