func (s *server) pushTxMsg(sp *serverPeer, hash *common.Uint256, doneChan chan<- struct{},
	waitChan <-chan struct{}) error {
	tx := s.txMemPool.GetTransaction(*hash)
	if tx == nil {
		if doneChan != nil {
			doneChan <- struct{}{}
		}
		return fmt.Errorf("unable to fetch tx %v from transaction pool", hash)
	}
	if waitChan != nil {
		<-waitChan
	}
	sp.QueueMessage(msg.NewTx(tx), doneChan)
	return nil
}
pushTxMsg sends a tx message for the provided transaction hash to the
connected peer.  An error is returned if the transaction hash is not known.
package server
import (
	"bytes"
	"fmt"
	"github.com/elastos/Elastos.ELA.SideChain/blockchain"
	"github.com/elastos/Elastos.ELA.SideChain/bloom"
	"github.com/elastos/Elastos.ELA.SideChain/config"
	"github.com/elastos/Elastos.ELA.SideChain/filter"
	"github.com/elastos/Elastos.ELA.SideChain/mempool"
	"github.com/elastos/Elastos.ELA.SideChain/netsync"
	"github.com/elastos/Elastos.ELA.SideChain/pact"
	"github.com/elastos/Elastos.ELA.SideChain/peer"
	"github.com/elastos/Elastos.ELA.SideChain/types"
	"github.com/elastos/Elastos.ELA/common"
	"github.com/elastos/Elastos.ELA/p2p"
	"github.com/elastos/Elastos.ELA/p2p/msg"
	p2psvr "github.com/elastos/Elastos.ELA/p2p/server"
)
const (
	defaultServices = pact.SFNodeNetwork | pact.SFNodeBloom
	MaxBlocksPerMsg = 500
	InvTypeFilteredSideChainBlock msg.InvType = 6
)
type Config struct {
	DataDir string
	Chain *blockchain.BlockChain
	TxMemPool *mempool.TxPool
	ChainParams *config.Params
	PermanentPeers []string
	NewTxFilter func(filter.TxFilterType) filter.TxFilter
	NodeVersion string
}
type naFilter struct{}
func (f *naFilter) Filter(na *p2p.NetAddress) bool {
	service := pact.ServiceFlag(na.Services)
	return service&pact.SFNodeNetwork == pact.SFNodeNetwork
}
type newPeerMsg struct {
	p2psvr.IPeer
	reply chan bool
}
type donePeerMsg struct {
	p2psvr.IPeer
	reply chan struct{}
}
type relayMsg struct {
	invVect *msg.InvVect
	data    interface{}
}
type server struct {
	p2psvr.IServer
	syncManager *netsync.SyncManager
	chain       *blockchain.BlockChain
	txMemPool   *mempool.TxPool
	newTxFiler  func(filter.TxFilterType) filter.TxFilter
	peerQueue chan interface{}
	relayInv  chan relayMsg
	quit      chan struct{}
	services  pact.ServiceFlag
}
type serverPeer struct {
	*peer.Peer
	server        *server
	continueHash  *common.Uint256
	isWhitelisted bool
	filter        *filter.Filter
	quit          chan struct{}
	txProcessed    chan struct{}
	blockProcessed chan struct{}
}
func newServerPeer(s *server) *serverPeer {
	return &serverPeer{
		server:         s,
		filter:         filter.New(s.newTxFiler),
		quit:           make(chan struct{}),
		txProcessed:    make(chan struct{}, 1),
		blockProcessed: make(chan struct{}, 1),
	}
}
func (sp *serverPeer) handleDisconnect() {
	sp.WaitForDisconnect()
	sp.server.syncManager.DonePeer(sp.Peer)
}
func (sp *serverPeer) OnVersion(_ *peer.Peer, m *msg.Version) {
	sp.server.chain.TimeSource.AddTimeSample(sp.Addr(), m.Timestamp)
	sp.server.syncManager.NewPeer(sp.Peer)
	sp.SetDisableRelayTx(!m.Relay)
	go sp.handleDisconnect()
}
func (sp *serverPeer) OnMemPool(_ *peer.Peer, _ *msg.MemPool) {
	if sp.server.services&pact.SFNodeBloom != pact.SFNodeBloom {
		log.Debugf("peer %v sent mempool request with bloom "+
			"filtering disabled -- disconnecting", sp)
		sp.Disconnect()
		return
	}
	sp.AddBanScore(0, 33, "mempool")
	txs := sp.server.txMemPool.GetTxsInPool()
	invMsg := msg.NewInvSize(uint(len(txs)))
	for _, tx := range txs {
		txId := tx.Hash()
		if !sp.filter.IsLoaded() || sp.filter.Match(tx) {
			iv := msg.NewInvVect(msg.InvTypeTx, &txId)
			invMsg.AddInvVect(iv)
			if len(invMsg.InvList)+1 > msg.MaxInvPerMsg {
				break
			}
		}
	}
	if len(invMsg.InvList) > 0 {
		sp.QueueMessage(invMsg, nil)
	}
}
func (sp *serverPeer) OnTx(_ *peer.Peer, msgTx *msg.Tx) {
	tx := msgTx.Serializable.(*types.Transaction)
	txId := tx.Hash()
	iv := msg.NewInvVect(msg.InvTypeTx, &txId)
	sp.AddKnownInventory(iv)
	sp.server.syncManager.QueueTx(tx, sp.Peer, sp.txProcessed)
	<-sp.txProcessed
}
func (sp *serverPeer) OnBlock(_ *peer.Peer, msgBlock *msg.Block) {
	block := msgBlock.Serializable.(*types.Block)
	blockHash := block.Hash()
	iv := msg.NewInvVect(msg.InvTypeBlock, &blockHash)
	sp.AddKnownInventory(iv)
	sp.server.syncManager.QueueBlock(block, sp.Peer, sp.blockProcessed)
	<-sp.blockProcessed
}
func (sp *serverPeer) OnInv(_ *peer.Peer, msg *msg.Inv) {
	if len(msg.InvList) > 0 {
		sp.server.syncManager.QueueInv(msg, sp.Peer)
	}
}
func (sp *serverPeer) OnNotFound(_ *peer.Peer, notFound *msg.NotFound) {
	for _, i := range notFound.InvList {
		if i.Type == msg.InvTypeTx {
			continue
		}
		log.Debugf("%s sent us notfound message --  disconnecting", sp)
		sp.AddBanScore(100, 0, notFound.CMD())
		sp.Disconnect()
		return
	}
}
func (sp *serverPeer) OnGetData(_ *peer.Peer, getData *msg.GetData) {
	numAdded := 0
	notFound := msg.NewNotFound()
	length := len(getData.InvList)
	sp.AddBanScore(0, uint32(length)*99/msg.MaxInvPerMsg, "getdata")
	var waitChan chan struct{}
	doneChan := make(chan struct{}, 1)
	for i, iv := range getData.InvList {
		var c chan struct{}
		if i == length-1 && len(notFound.InvList) == 0 {
			c = doneChan
		} else if (i+1)%5 == 0 {
			c = make(chan struct{}, 1)
		}
		var err error
		switch iv.Type {
		case msg.InvTypeTx:
			err = sp.server.pushTxMsg(sp, &iv.Hash, c, waitChan)
		case msg.InvTypeBlock:
			err = sp.server.pushBlockMsg(sp, &iv.Hash, c, waitChan)
		case msg.InvTypeFilteredBlock:
			err = sp.server.pushMerkleBlockMsg(sp, &iv.Hash, c, waitChan)
		case InvTypeFilteredSideChainBlock:
			err = sp.server.pushSideChainMerkleBlockMsg(sp, &iv.Hash, c, waitChan)
		default:
			continue
		}
		if err != nil {
			notFound.AddInvVect(iv)
			if i == length-1 && c != nil {
				<-c
			}
		}
		numAdded++
		waitChan = c
	}
	if len(notFound.InvList) != 0 {
		sp.QueueMessage(notFound, doneChan)
	}
	if numAdded > 0 {
		<-doneChan
	}
}
func (sp *serverPeer) OnGetBlocks(_ *peer.Peer, m *msg.GetBlocks) {
	chain := sp.server.chain
	hashList := chain.LocateBlocks(m.Locator, &m.HashStop, MaxBlocksPerMsg)
	invMsg := msg.NewInv()
	for i := range hashList {
		iv := msg.NewInvVect(msg.InvTypeBlock, hashList[i])
		invMsg.AddInvVect(iv)
	}
	if len(invMsg.InvList) > 0 {
		invListLen := len(invMsg.InvList)
		if invListLen == MaxBlocksPerMsg {
			continueHash := invMsg.InvList[invListLen-1].Hash
			sp.continueHash = &continueHash
		}
		sp.QueueMessage(invMsg, nil)
	}
}
func (sp *serverPeer) enforceNodeBloomFlag(cmd string) bool {
	if sp.server.services&pact.SFNodeBloom != pact.SFNodeBloom {
		log.Debugf("%s sent an unsupported %s request -- "+
			"disconnecting", sp, cmd)
		sp.AddBanScore(100, 0, cmd)
		sp.Disconnect()
		return false
	}
	return true
}
func (sp *serverPeer) OnFilterAdd(_ *peer.Peer, filterAdd *msg.FilterAdd) {
	if !sp.enforceNodeBloomFlag(filterAdd.CMD()) {
		return
	}
	if !sp.filter.IsLoaded() {
		log.Debugf("%s sent a filteradd request with no filter "+
			"loaded -- disconnecting", sp)
		sp.Disconnect()
		return
	}
	err := sp.filter.Add(filterAdd.Data)
	if err != nil {
		log.Debugf("%s sent invalid filteradd request with error %s"+
			" -- disconnecting", sp, err)
		sp.Disconnect()
	}
}
func (sp *serverPeer) OnFilterClear(_ *peer.Peer, filterClear *msg.FilterClear) {
	if !sp.enforceNodeBloomFlag(filterClear.CMD()) {
		return
	}
	if !sp.filter.IsLoaded() {
		log.Debugf("%s sent a filterclear request with no "+
			"filter loaded -- disconnecting", sp)
		sp.Disconnect()
		return
	}
	sp.filter.Clear()
	sp.SetDisableRelayTx(true)
}
func (sp *serverPeer) OnFilterLoad(_ *peer.Peer, filterLoad *msg.FilterLoad) {
	if !sp.enforceNodeBloomFlag(filterLoad.CMD()) {
		return
	}
	sp.SetDisableRelayTx(false)
	buf := new(bytes.Buffer)
	filterLoad.Serialize(buf)
	err := sp.filter.Load(&msg.TxFilterLoad{
		Type: filter.FTBloom,
		Data: buf.Bytes(),
	})
	if err != nil {
		log.Debugf("%s sent invalid filterload request with error %s"+
			" -- disconnecting", sp, err)
		sp.Disconnect()
	}
}
func (sp *serverPeer) enforceTxFilterFlag(cmd string) bool {
	if sp.server.services&pact.SFTxFiltering != pact.SFTxFiltering {
		log.Debugf("%s sent an unsupported %s request -- "+
			"disconnecting", sp, cmd)
		sp.AddBanScore(100, 0, cmd)
		sp.Disconnect()
		return false
	}
	return true
}
func (sp *serverPeer) OnTxFilterLoad(_ *peer.Peer, tf *msg.TxFilterLoad) {
	if !sp.enforceTxFilterFlag(tf.CMD()) {
		return
	}
	sp.SetDisableRelayTx(false)
	err := sp.filter.Load(tf)
	if err != nil {
		log.Debugf("%s sent invalid txfilter request with error %s"+
			" -- disconnecting", sp, err)
		sp.Disconnect()
		return
	}
}
func (sp *serverPeer) OnReject(_ *peer.Peer, msg *msg.Reject) {
	log.Infof("%s sent a reject message Code: %s, Hash %s, Reason: %s",
		sp, msg.RejectCode.String(), msg.Hash.String(), msg.Reason)
}