func (sm *SyncManager) handleTxMsg(tmsg *txMsg) {
	peer := tmsg.peer
	state, exists := sm.peerStates[peer]
	if !exists {
		log.Warnf("Received tx message from unknown peer %s", peer)
		return
	}
	txHash := tmsg.tx.Hash()
	if _, exists = sm.rejectedTxns[txHash]; exists {
		log.Debugf("Ignoring unsolicited previously rejected "+
			"transaction %v from %s", txHash, peer)
		return
	}
	delete(state.requestedTxns, txHash)
	delete(sm.requestedTxns, txHash)
	err := sm.txMemPool.AppendToTxPool(tmsg.tx)
	if err != nil {
		sm.rejectedTxns[txHash] = struct{}{}
		sm.limitMap(sm.rejectedTxns, maxRejectedTxns)
		elaErr := errors.SimpleWithMessage(errors.ErrTxPoolFailure, err,
			fmt.Sprintf("AppendToTxPool fail tx hash %s", tmsg.tx.Hash()))
		peer.PushRejectMsg(p2p.CmdTx, elaErr, &txHash, false)
		return
	}
	iv := msg.NewInvVect(msg.InvTypeTx, &txHash)
	sm.peerNotifier.RelayInventory(iv, tmsg.tx)
}
handleTxMsg handles transaction messages from all peers.
package netsync
import (
	"fmt"
	"sync"
	"sync/atomic"
	"github.com/elastos/Elastos.ELA.SideChain/blockchain"
	"github.com/elastos/Elastos.ELA.SideChain/events"
	"github.com/elastos/Elastos.ELA.SideChain/mempool"
	"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/errors"
	"github.com/elastos/Elastos.ELA/common"
	"github.com/elastos/Elastos.ELA/p2p"
	"github.com/elastos/Elastos.ELA/p2p/msg"
)
const (
	maxRejectedTxns = 1000
	maxRequestedBlocks = msg.MaxInvPerMsg
	maxRequestedTxns = msg.MaxInvPerMsg
)
var zeroHash common.Uint256
type newPeerMsg struct {
	peer *peer.Peer
}
type blockMsg struct {
	block *types.Block
	peer  *peer.Peer
	reply chan struct{}
}
type invMsg struct {
	inv  *msg.Inv
	peer *peer.Peer
}
type donePeerMsg struct {
	peer *peer.Peer
}
type txMsg struct {
	tx    *types.Transaction
	peer  *peer.Peer
	reply chan struct{}
}
type getSyncPeerMsg struct {
	reply chan uint64
}
type processBlockResponse struct {
	isOrphan bool
	err      error
}
type processBlockMsg struct {
	block *types.Block
	reply chan processBlockResponse
}
type isCurrentMsg struct {
	reply chan bool
}
type pauseMsg struct {
	unpause <-chan struct{}
}
type peerSyncState struct {
	syncCandidate   bool
	requestQueue    []*msg.InvVect
	requestedTxns   map[common.Uint256]struct{}
	requestedBlocks map[common.Uint256]struct{}
}
type SyncManager struct {
	peerNotifier PeerNotifier
	started      int32
	shutdown     int32
	chain        *blockchain.BlockChain
	txMemPool    *mempool.TxPool
	msgChan      chan interface{}
	wg           sync.WaitGroup
	quit         chan struct{}
	rejectedTxns    map[common.Uint256]struct{}
	requestedTxns   map[common.Uint256]struct{}
	requestedBlocks map[common.Uint256]struct{}
	syncPeer        *peer.Peer
	peerStates      map[*peer.Peer]*peerSyncState
}
func (sm *SyncManager) startSync() {
	if sm.syncPeer != nil {
		return
	}
	bestHeight := sm.chain.GetBestHeight()
	var bestPeer *peer.Peer
	for peer, state := range sm.peerStates {
		if !state.syncCandidate {
			continue
		}
		if peer.Height() < bestHeight {
			state.syncCandidate = false
			continue
		}
		bestPeer = peer
		break
	}
	if bestPeer != nil {
		if bestPeer.Height() == bestHeight {
			return
		}
		sm.requestedBlocks = make(map[common.Uint256]struct{})
		locator, err := sm.chain.LatestBlockLocator()
		if err != nil {
			log.Errorf("Failed to get block locator for the "+
				"latest block: %v", err)
			return
		}
		log.Infof("Syncing to block height %d from peer %v",
			bestPeer.Height(), bestPeer.Addr())
		bestPeer.PushGetBlocksMsg(locator, &zeroHash)
		sm.syncPeer = bestPeer
	} else {
		log.Warnf("No sync peer candidates available")
	}
}
func (sm *SyncManager) isSyncCandidate(peer *peer.Peer) bool {
	if peer.Services()&pact.SFNodeNetwork != pact.SFNodeNetwork {
		return false
	}
	return true
}
func (sm *SyncManager) handleNewPeerMsg(peer *peer.Peer) {
	if atomic.LoadInt32(&sm.shutdown) != 0 {
		return
	}
	log.Infof("New valid peer %s", peer)
	isSyncCandidate := sm.isSyncCandidate(peer)
	sm.peerStates[peer] = &peerSyncState{
		syncCandidate:   isSyncCandidate,
		requestedTxns:   make(map[common.Uint256]struct{}),
		requestedBlocks: make(map[common.Uint256]struct{}),
	}
	if isSyncCandidate && sm.syncPeer == nil {
		sm.startSync()
	}
}
func (sm *SyncManager) handleDonePeerMsg(peer *peer.Peer) {
	state, exists := sm.peerStates[peer]
	if !exists {
		log.Warnf("Received done peer message for unknown peer %s", peer)
		return
	}
	delete(sm.peerStates, peer)
	log.Infof("Lost peer %s", peer)
	for txHash := range state.requestedTxns {
		delete(sm.requestedTxns, txHash)
	}
	for blockHash := range state.requestedBlocks {
		delete(sm.requestedBlocks, blockHash)
	}
	if sm.syncPeer == peer {
		sm.syncPeer = nil
		sm.startSync()
	}
}