func (c *Client) Send(dests *StringArray, data interface{}, config *MessageConfig) (*OnMessage, error) {
	config, err := MergeMessageConfig(c.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 := c.send(dests.Elems(), payload, !config.Unencrypted, config.MaxHoldingSeconds); err != nil {
		return nil, err
	}
	onReply := NewOnMessage(1, nil)
	if !config.NoReply {
		c.responseChannels.Add(string(payload.MessageId), onReply, cache.DefaultExpiration)
	}
	return onReply, nil
}
Send sends bytes or string data to one or multiple destinations with an
optional config. Returned OnMessage channel will emit if a reply or ACK for
this message is received.
package nkn
import (
	"bytes"
	"compress/zlib"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"log"
	"net/url"
	"strings"
	"sync"
	"time"
	"github.com/golang/protobuf/proto"
	"github.com/gorilla/websocket"
	"github.com/nknorg/nkn-sdk-go/payloads"
	"github.com/nknorg/nkn/v2/api/common/errcode"
	"github.com/nknorg/nkn/v2/config"
	"github.com/nknorg/nkn/v2/crypto"
	"github.com/nknorg/nkn/v2/crypto/ed25519"
	"github.com/nknorg/nkn/v2/pb"
	"github.com/nknorg/nkn/v2/transaction"
	"github.com/nknorg/nkn/v2/util/address"
	"github.com/patrickmn/go-cache"
	"golang.org/x/crypto/nacl/box"
)
const (
	MessageIDSize = 8
	maxClientMessageSize = 4000000
	pingInterval = 8 * time.Second
	pongTimeout = 10 * time.Second
	setClientAction = "setClient"
)
type Client struct {
	OnConnect *OnConnect  
	OnMessage *OnMessage  
	config            *ClientConfig
	account           *Account
	publicKey         []byte
	curveSecretKey    *[sharedKeySize]byte
	address           string
	addressID         []byte
	sigChainBlockHash string
	reconnectChan     chan struct{}
	responseChannels  *cache.Cache
	lock       sync.RWMutex
	isClosed   bool
	conn       *websocket.Conn
	node       *Node
	sharedKeys map[string]*[sharedKeySize]byte
	wallet     *Wallet
}
type clientInterface interface {
	getConfig() *ClientConfig
	send(dests []string, payload *payloads.Payload, encrypted bool, maxHoldingSeconds int32) error
	GetSubscribers(topic string, offset, limit int, meta, txPool bool, subscriberHashPrefix []byte) (*Subscribers, error)
}
type setClientResult struct {
	Node              *Node  `json:"node"`
	SigChainBlockHash string `json:"sigChainBlockHash"`
}
func NewClient(account *Account, identifier string, config *ClientConfig) (*Client, error) {
	config, err := MergeClientConfig(config)
	if err != nil {
		return nil, err
	}
	pk := account.PubKey()
	addr := address.MakeAddressString(pk, identifier)
	var sk [ed25519.PrivateKeySize]byte
	copy(sk[:], account.PrivKey())
	curveSecretKey := ed25519.PrivateKeyToCurve25519PrivateKey(&sk)
	w, err := NewWallet(account, &WalletConfig{SeedRPCServerAddr: config.SeedRPCServerAddr})
	if err != nil {
		return nil, err
	}
	c := Client{
		config:           config,
		account:          account,
		publicKey:        pk,
		curveSecretKey:   curveSecretKey,
		address:          addr,
		addressID:        addressToID(addr),
		OnConnect:        NewOnConnect(1, nil),
		OnMessage:        NewOnMessage(int(config.MsgChanLen), nil),
		reconnectChan:    make(chan struct{}),
		responseChannels: cache.New(time.Duration(config.MsgCacheExpiration)*time.Millisecond, time.Duration(config.MsgCacheCleanupInterval)*time.Millisecond),
		sharedKeys:       make(map[string]*[sharedKeySize]byte),
		wallet:           w,
	}
	go c.handleReconnect()
	err = c.connect(int(c.config.ConnectRetries))
	if err != nil {
		return nil, err
	}
	return &c, nil
}
func (c *Client) Account() *Account {
	return c.account
}
func (c *Client) Seed() []byte {
	return c.account.Seed()
}
func (c *Client) PubKey() []byte {
	return c.account.PubKey()
}
func (c *Client) Address() string {
	return c.address
}
func (c *Client) IsClosed() bool {
	c.lock.RLock()
	defer c.lock.RUnlock()
	return c.isClosed
}
func (c *Client) Close() error {
	c.lock.Lock()
	defer c.lock.Unlock()
	if c.isClosed {
		return nil
	}
	c.isClosed = true
	c.OnConnect.close()
	c.OnMessage.close()
	close(c.reconnectChan)
	c.conn.Close()
	return nil
}
func (c *Client) GetNode() *Node {
	c.lock.RLock()
	defer c.lock.RUnlock()
	return c.node
}
func (c *Client) GetConn() *websocket.Conn {
	c.lock.RLock()
	defer c.lock.RUnlock()
	return c.conn
}
func (c *Client) getOrComputeSharedKey(remotePublicKey []byte) (*[sharedKeySize]byte, error) {
	c.lock.RLock()
	sharedKey, ok := c.sharedKeys[string(remotePublicKey)]
	c.lock.RUnlock()
	if ok && sharedKey != nil {
		return sharedKey, nil
	}
	if len(remotePublicKey) != ed25519.PublicKeySize {
		return nil, ErrInvalidPubkeySize
	}
	var pk [ed25519.PublicKeySize]byte
	copy(pk[:], remotePublicKey)
	curve25519PublicKey, ok := ed25519.PublicKeyToCurve25519PublicKey(&pk)
	if !ok {
		return nil, ErrInvalidPubkey
	}
	sharedKey = new([sharedKeySize]byte)
	box.Precompute(sharedKey, curve25519PublicKey, c.curveSecretKey)
	c.lock.Lock()
	c.sharedKeys[string(remotePublicKey)] = sharedKey
	c.lock.Unlock()
	return sharedKey, nil
}
func (c *Client) encryptPayload(msg *payloads.Payload, dests []string) ([][]byte, error) {
	rawPayload, err := proto.Marshal(msg)
	if err != nil {
		return nil, err
	}
	if len(dests) > 1 {
		var key [sharedKeySize]byte
		if _, err := rand.Read(key[:]); err != nil {
			return nil, err
		}
		encrypted, msgNonce, err := encrypt(rawPayload, &key)
		if err != nil {
			return nil, err
		}
		msgs := make([][]byte, len(dests))
		for i, dest := range dests {
			_, destPubkey, _, err := address.ParseClientAddress(dest)
			if err != nil {
				return nil, err
			}
			sharedKey, err := c.getOrComputeSharedKey(destPubkey)
			if err != nil {
				return nil, err
			}
			encryptedKey, keyNonce, err := encrypt(key[:], sharedKey)
			if err != nil {
				return nil, err
			}
			nonce := append(keyNonce, msgNonce...)
			msgs[i], err = proto.Marshal(&payloads.Message{
				Payload:      encrypted,
				Encrypted:    true,
				Nonce:        nonce,
				EncryptedKey: encryptedKey,
			})
			if err != nil {
				return nil, err
			}
		}
		return msgs, nil
	}
	_, destPubkey, _, err := address.ParseClientAddress(dests[0])
	if err != nil {
		return nil, err
	}
	sharedKey, err := c.getOrComputeSharedKey(destPubkey)
	if err != nil {
		return nil, err
	}
	encrypted, nonce, err := encrypt(rawPayload, sharedKey)
	if err != nil {
		return nil, err
	}
	data, err := proto.Marshal(&payloads.Message{
		Payload:      encrypted,
		Encrypted:    true,
		Nonce:        nonce,
		EncryptedKey: nil,
	})
	if err != nil {
		return nil, err
	}
	return [][]byte{data}, nil
}
func (c *Client) decryptPayload(msg *payloads.Message, srcAddr string) ([]byte, error) {
	rawPayload := msg.Payload
	_, srcPubkey, _, err := address.ParseClientAddress(srcAddr)
	if err != nil {
		return nil, err
	}
	encryptedKey := msg.EncryptedKey
	var decrypted []byte
	if encryptedKey != nil && len(encryptedKey) > 0 {
		var keyNonce, msgNonce [nonceSize]byte
		copy(keyNonce[:], msg.Nonce[:nonceSize])
		copy(msgNonce[:], msg.Nonce[nonceSize:])
		sharedKey, err := c.getOrComputeSharedKey(srcPubkey)
		if err != nil {
			return nil, err
		}
		decryptedKeySlice, err := decrypt(encryptedKey, keyNonce, sharedKey)
		if err != nil {
			return nil, err
		}
		var decryptedKey [sharedKeySize]byte
		copy(decryptedKey[:], decryptedKeySlice)
		decrypted, err = decrypt(rawPayload, msgNonce, &decryptedKey)
		if err != nil {
			return nil, err
		}
	} else {
		var nonce [nonceSize]byte
		copy(nonce[:], msg.Nonce)
		sharedKey, err := c.getOrComputeSharedKey(srcPubkey)
		if err != nil {
			return nil, err
		}
		decrypted, err = decrypt(rawPayload, nonce, sharedKey)
		if err != nil {
			return nil, err
		}
	}
	return decrypted, nil
}
func (c *Client) handleMessage(msgType int, data []byte) error {
	if c.IsClosed() {
		return nil
	}
	switch msgType {
	case websocket.TextMessage:
		msg := make(map[string]*json.RawMessage)
		if err := json.Unmarshal(data, &msg); err != nil {
			return err
		}
		var action string
		if err := json.Unmarshal(*msg["Action"], &action); err != nil {
			return err
		}
		var errCode errcode.ErrCode
		if err := json.Unmarshal(*msg["Error"], &errCode); err != nil {
			return err
		}
		if errCode != errcode.SUCCESS {
			if errCode == errcode.WRONG_NODE {
				var node Node
				if err := json.Unmarshal(*msg["Result"], &node); err != nil {
					return err
				}
				go func() {
					err := c.connectToNode(&node)
					if err != nil {
						c.Reconnect()
					}
				}()
			} else if action == setClientAction {
				c.Close()
			}
			return errorWithCode{
				err:  errors.New(errcode.ErrMessage[errCode]),
				code: int32(errCode),
			}
		}
		switch action {
		case setClientAction:
			var result setClientResult
			if err := json.Unmarshal(*msg["Result"], &result); err != nil {
				return err
			}
			c.sigChainBlockHash = result.SigChainBlockHash
			c.lock.RLock()
			defer c.lock.RUnlock()
			if c.isClosed {
				return nil
			}
			node := c.GetNode()
			c.OnConnect.receive(node)
		case "updateSigChainBlockHash":
			var sigChainBlockHash string
			if err := json.Unmarshal(*msg["Result"], &sigChainBlockHash); err != nil {
				return err
			}
			c.sigChainBlockHash = sigChainBlockHash
		}
	case websocket.BinaryMessage:
		clientMsg := &pb.ClientMessage{}
		if err := proto.Unmarshal(data, clientMsg); err != nil {
			return err
		}
		switch clientMsg.MessageType {
		case pb.ClientMessageType_INBOUND_MESSAGE:
			inboundMsg := &pb.InboundMessage{}
			if err := proto.Unmarshal(clientMsg.Message, inboundMsg); err != nil {
				return err
			}
			if len(inboundMsg.PrevHash) > 0 {
				go func() {
					if err := c.sendReceipt(inboundMsg.PrevHash); err != nil {
						log.Println(err)
					}
				}()
			}
			payloadMsg := &payloads.Message{}
			if err := proto.Unmarshal(inboundMsg.Payload, payloadMsg); err != nil {
				return err
			}
			var payloadBytes []byte
			if payloadMsg.Encrypted {
				var err error
				payloadBytes, err = c.decryptPayload(payloadMsg, inboundMsg.Src)
				if err != nil {
					return err
				}
			} else {
				payloadBytes = payloadMsg.Payload
			}
			payload := &payloads.Payload{}
			if err := proto.Unmarshal(payloadBytes, payload); err != nil {
				return err
			}
			data := payload.Data
			switch payload.Type {
			case payloads.PayloadType_TEXT:
				textData := &payloads.TextData{}
				if err := proto.Unmarshal(data, textData); err != nil {
					return err
				}
				data = []byte(textData.Text)
			case payloads.PayloadType_ACK:
				data = nil
			}
			msg := &Message{
				Src:       inboundMsg.Src,
				Data:      data,
				Type:      int32(payload.Type),
				Encrypted: payloadMsg.Encrypted,
				MessageID: payload.MessageId,
				NoReply:   payload.NoReply,
			}
			if len(payload.ReplyToId) > 0 {
				msgIDString := string(payload.ReplyToId)
				onReply, ok := c.responseChannels.Get(msgIDString)
				if ok {
					c.responseChannels.Delete(msgIDString)
					onReply.(*OnMessage).receive(msg, false)
				}
				return nil
			}
			if msg == nil {
				return nil
			}
			if payload.NoReply {
				msg.reply = func(data interface{}) error {
					return nil
				}
			} else {
				msg.reply = func(data interface{}) error {
					payload, err := newReplyPayload(data, payload.MessageId)
					if err != nil {
						return err
					}
					if err := c.send([]string{inboundMsg.Src}, payload, payloadMsg.Encrypted, 0); err != nil {
						return err
					}
					return nil
				}
			}
			c.lock.RLock()
			defer c.lock.RUnlock()
			if c.isClosed {
				return nil
			}
			c.OnMessage.receive(msg, true)
		}
	}
	return nil
}
func (c *Client) connectToNode(node *Node) error {
	var rpcAddr string
	var wg sync.WaitGroup
	if len(node.RPCAddr) > 0 {
		wg.Add(1)
		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()
		go func() {
			defer wg.Done()
			addr := (&url.URL{Scheme: "http", Host: node.RPCAddr}).String()
			nodeState, err := GetNodeStateContext(ctx, &RPCConfig{
				SeedRPCServerAddr: NewStringArray(addr),
				RPCTimeout:        c.config.WsHandshakeTimeout,
			})
			if err != nil {
				return
			}
			if nodeState.SyncState != pb.SyncState_name[int32(pb.SyncState_PERSIST_FINISHED)] {
				return
			}
			rpcAddr = addr
		}()
	}
	wsAddr := (&url.URL{Scheme: "ws", Host: node.Addr}).String()
	dialer := websocket.DefaultDialer
	dialer.HandshakeTimeout = time.Duration(c.config.WsHandshakeTimeout) * time.Millisecond
	conn, _, err := dialer.Dial(wsAddr, nil)
	if err != nil {
		return err
	}
	wg.Wait()
	c.lock.Lock()
	prevConn := c.conn
	c.conn = conn
	c.node = node
	if len(rpcAddr) > 0 {
		c.wallet.config.SeedRPCServerAddr = NewStringArray(rpcAddr)
	} else {
		c.wallet.config.SeedRPCServerAddr = NewStringArray()
	}
	c.lock.Unlock()
	if prevConn != nil {
		prevConn.Close()
	}
	conn.SetReadLimit(config.MaxClientMessageSize)
	conn.SetReadDeadline(time.Now().Add(pongTimeout))
	conn.SetPongHandler(func(string) error {
		conn.SetReadDeadline(time.Now().Add(pongTimeout))
		return nil
	})
	done := make(chan struct{})
	go func() {
		ticker := time.NewTicker(pingInterval)
		defer ticker.Stop()
		var err error
		for {
			select {
			case <-ticker.C:
				c.lock.Lock()
				conn.SetWriteDeadline(time.Now().Add(pingInterval))
				err = conn.WriteMessage(websocket.PingMessage, nil)
				c.lock.Unlock()
				if err != nil {
					log.Println(err)
					c.Reconnect()
					return
				}
			case <-done:
				return
			}
		}
	}()
	go func() {
		req := make(map[string]interface{})
		req["Action"] = "setClient"
		req["Addr"] = c.Address()
		c.lock.Lock()
		err := conn.WriteJSON(req)
		c.lock.Unlock()
		if err != nil {
			log.Println(err)
			c.Reconnect()
			return
		}
	}()
	go func() {
		defer close(done)
		for {
			if c.IsClosed() {
				return
			}
			msgType, data, err := conn.ReadMessage()
			if err != nil {
				log.Println(err)
				c.Reconnect()
				return
			}
			conn.SetReadDeadline(time.Now().Add(pongTimeout))
			err = c.handleMessage(msgType, data)
			if err != nil {
				log.Println(err)
				continue
			}
		}
	}()
	return nil
}
func (c *Client) connect(maxRetries int) error {
	retryInterval := c.config.MinReconnectInterval
	for retry := 1; maxRetries == 0 || retry <= maxRetries; retry++ {
		if retry > 1 {
			log.Printf("Retry in %v ms...", retryInterval)
			time.Sleep(time.Duration(retryInterval) * time.Millisecond)
			retryInterval *= 2
			if retryInterval > c.config.MaxReconnectInterval {
				retryInterval = c.config.MaxReconnectInterval
			}
		}
		node, err := GetWsAddr(c.Address(), c.config)
		if err != nil {
			log.Println(err)
			continue
		}
		err = c.connectToNode(node)
		if err != nil {
			log.Println(err)
			continue
		}
		return nil
	}
	return ErrConnectFailed
}
func (c *Client) Reconnect() {
	if c.IsClosed() {
		return
	}
	select {
	case c.reconnectChan <- struct{}{}:
	default:
	}
}
func (c *Client) handleReconnect() {
	for range c.reconnectChan {
		if c.IsClosed() {
			return
		}
		log.Printf("Reconnect in %v ms...", c.config.MinReconnectInterval)
		time.Sleep(time.Duration(c.config.MinReconnectInterval) * time.Millisecond)
		err := c.connect(0)
		if err != nil {
			log.Println(err)
			c.Close()
			return
		}
	}
}
func (c *Client) writeMessage(buf []byte) error {
	c.lock.Lock()
	c.conn.SetWriteDeadline(time.Now().Add(time.Duration(c.config.WsWriteTimeout) * time.Millisecond))
	err := c.conn.WriteMessage(websocket.BinaryMessage, buf)
	c.lock.Unlock()
	if err != nil {
		c.Reconnect()
	}
	return err
}
func (c *Client) sendReceipt(prevSignature []byte) error {
	sigChainElem := &pb.SigChainElem{}
	buff := bytes.NewBuffer(nil)
	if err := sigChainElem.SerializationUnsigned(buff); err != nil {
		return err
	}
	sigChainElemSerialized := buff.Bytes()
	digest := sha256.Sum256(prevSignature)
	digest = sha256.Sum256(append(digest[:], sigChainElemSerialized...))
	signature, err := crypto.Sign(c.account.PrivateKey, digest[:])
	if err != nil {
		return err
	}
	receipt := &pb.Receipt{
		PrevHash:  prevSignature,
		Signature: signature,
	}
	receiptData, err := proto.Marshal(receipt)
	if err != nil {
		return err
	}
	clientMsg := &pb.ClientMessage{
		MessageType:     pb.ClientMessageType_RECEIPT,
		Message:         receiptData,
		CompressionType: pb.CompressionType_COMPRESSION_NONE,
	}
	buf, err := proto.Marshal(clientMsg)
	if err != nil {
		return err
	}
	return c.writeMessage(buf)
}