func (p *peer) EnqueueSendTransactions(txs types.Transactions, queue chan broadcastItem) {
	p.enqueueSendNonEncodedItem(txs, EvmTxsMsg, queue)
	for _, tx := range txs {
		p.knownTxs.Add(tx.Hash())
	}
	for p.knownTxs.Cardinality() >= p.cfg.MaxKnownTxs {
		p.knownTxs.Pop()
	}
}
EnqueueSendTransactions queues list of transactions propagation to a remote
peer.
The method is blocking in a case if the peer's broadcast queue is full.
package gossip
import (
	"errors"
	"fmt"
	"sync"
	"time"
	"github.com/Fantom-foundation/lachesis-base/gossip/dagstream"
	"github.com/Fantom-foundation/lachesis-base/hash"
	"github.com/Fantom-foundation/lachesis-base/inter/dag"
	"github.com/Fantom-foundation/lachesis-base/inter/idx"
	"github.com/Fantom-foundation/lachesis-base/utils/datasemaphore"
	mapset "github.com/deckarep/golang-set"
	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/core/types"
	"github.com/ethereum/go-ethereum/log"
	"github.com/ethereum/go-ethereum/p2p"
	"github.com/ethereum/go-ethereum/rlp"
	"github.com/Fantom-foundation/go-opera/inter"
)
var (
	errClosed            = errors.New("peer set is closed")
	errAlreadyRegistered = errors.New("peer is already registered")
	errNotRegistered     = errors.New("peer is not registered")
)
const (
	handshakeTimeout = 5 * time.Second
)
type PeerInfo struct {
	Version     int       `json:"version"`  
	Epoch       idx.Epoch `json:"epoch"`
	NumOfBlocks idx.Block `json:"blocks"`
}
type broadcastItem struct {
	Code uint64
	Raw  rlp.RawValue
}
type peer struct {
	id string
	cfg PeerCacheConfig
	*p2p.Peer
	rw p2p.MsgReadWriter
	version int  
	knownTxs            mapset.Set          
	knownEvents         mapset.Set          
	queue               chan broadcastItem  
	queuedDataSemaphore *datasemaphore.DataSemaphore
	term                chan struct{}  
	progress PeerProgress
	sync.RWMutex
}
func (p *peer) SetProgress(x PeerProgress) {
	p.Lock()
	defer p.Unlock()
	p.progress = x
}
func (p *peer) InterestedIn(h hash.Event) bool {
	e := h.Epoch()
	p.RLock()
	defer p.RUnlock()
	return e != 0 &&
		p.progress.Epoch != 0 &&
		(e == p.progress.Epoch || e == p.progress.Epoch+1) &&
		!p.knownEvents.Contains(h)
}
func (a *PeerProgress) Less(b PeerProgress) bool {
	if a.Epoch != b.Epoch {
		return a.Epoch < b.Epoch
	}
	return a.LastBlockIdx < b.LastBlockIdx
}
func NewPeer(version int, p *p2p.Peer, rw p2p.MsgReadWriter, cfg PeerCacheConfig) *peer {
	warningFn := func(received dag.Metric, processing dag.Metric, releasing dag.Metric) {
		log.Warn("Peer queue semaphore inconsistency",
			"receivedNum", received.Num, "receivedSize", received.Size,
			"processingNum", processing.Num, "processingSize", processing.Size,
			"releasingNum", releasing.Num, "releasingSize", releasing.Size)
	}
	return &peer{
		cfg:                 cfg,
		Peer:                p,
		rw:                  rw,
		version:             version,
		id:                  fmt.Sprintf("%x", p.ID().Bytes()[:8]),
		knownTxs:            mapset.NewSet(),
		knownEvents:         mapset.NewSet(),
		queue:               make(chan broadcastItem, cfg.MaxQueuedItems),
		queuedDataSemaphore: datasemaphore.New(dag.Metric{cfg.MaxQueuedItems, cfg.MaxQueuedSize}, warningFn),
		term:                make(chan struct{}),
	}
}
func (p *peer) broadcast(queue chan broadcastItem) {
	for {
		select {
		case item := <-queue:
			_ = p2p.Send(p.rw, item.Code, item.Raw)
			p.queuedDataSemaphore.Release(memSize(item.Raw))
		case <-p.term:
			return
		}
	}
}
func (p *peer) close() {
	p.queuedDataSemaphore.Terminate()
	close(p.term)
}
func (p *peer) Info() *PeerInfo {
	return &PeerInfo{
		Version:     p.version,
		Epoch:       p.progress.Epoch,
		NumOfBlocks: p.progress.LastBlockIdx,
	}
}
func (p *peer) MarkEvent(hash hash.Event) {
	for p.knownEvents.Cardinality() >= p.cfg.MaxKnownEvents {
		p.knownEvents.Pop()
	}
	p.knownEvents.Add(hash)
}
func (p *peer) MarkTransaction(hash common.Hash) {
	for p.knownTxs.Cardinality() >= p.cfg.MaxKnownTxs {
		p.knownTxs.Pop()
	}
	p.knownTxs.Add(hash)
}
func (p *peer) SendTransactions(txs types.Transactions) error {
	for _, tx := range txs {
		p.knownTxs.Add(tx.Hash())
	}
	for p.knownTxs.Cardinality() >= p.cfg.MaxKnownTxs {
		p.knownTxs.Pop()
	}
	return p2p.Send(p.rw, EvmTxsMsg, txs)
}
func (p *peer) SendTransactionHashes(txids []common.Hash) error {
	for _, txid := range txids {
		p.knownTxs.Add(txid)
	}
	for p.knownTxs.Cardinality() >= p.cfg.MaxKnownTxs {
		p.knownTxs.Pop()
	}
	return p2p.Send(p.rw, NewEvmTxHashesMsg, txids)
}
func memSize(v rlp.RawValue) dag.Metric {
	return dag.Metric{1, uint64(len(v) + 1024)}
}
func (p *peer) asyncSendEncodedItem(raw rlp.RawValue, code uint64, queue chan broadcastItem) bool {
	if !p.queuedDataSemaphore.TryAcquire(memSize(raw)) {
		return false
	}
	item := broadcastItem{
		Code: code,
		Raw:  raw,
	}
	select {
	case queue <- item:
		return true
	case <-p.term:
	default:
	}
	p.queuedDataSemaphore.Release(memSize(raw))
	return false
}
func (p *peer) asyncSendNonEncodedItem(value interface{}, code uint64, queue chan broadcastItem) bool {
	raw, err := rlp.EncodeToBytes(value)
	if err != nil {
		return false
	}
	return p.asyncSendEncodedItem(raw, code, queue)
}
func (p *peer) enqueueSendEncodedItem(raw rlp.RawValue, code uint64, queue chan broadcastItem) {
	if !p.queuedDataSemaphore.Acquire(memSize(raw), 10*time.Second) {
		return
	}
	item := broadcastItem{
		Code: code,
		Raw:  raw,
	}
	select {
	case queue <- item:
		return
	case <-p.term:
	}
	p.queuedDataSemaphore.Release(memSize(raw))
}
func (p *peer) enqueueSendNonEncodedItem(value interface{}, code uint64, queue chan broadcastItem) {
	raw, err := rlp.EncodeToBytes(value)
	if err != nil {
		return
	}
	p.enqueueSendEncodedItem(raw, code, queue)
}
func SplitTransactions(txs types.Transactions, fn func(types.Transactions)) {
	for len(txs) > 0 {
		batchSize := 0
		var batch types.Transactions
		for i, tx := range txs {
			batchSize += int(tx.Size()) + 1024
			batch = txs[:i+1]
			if batchSize >= softResponseLimitSize || i+1 >= softLimitItems {
				break
			}
		}
		txs = txs[len(batch):]
		fn(batch)
	}
}
func (p *peer) AsyncSendTransactions(txs types.Transactions, queue chan broadcastItem) {
	if p.asyncSendNonEncodedItem(txs, EvmTxsMsg, queue) {
		for _, tx := range txs {
			p.knownTxs.Add(tx.Hash())
		}
		for p.knownTxs.Cardinality() >= p.cfg.MaxKnownTxs {
			p.knownTxs.Pop()
		}
	} else {
		p.Log().Debug("Dropping transactions propagation", "count", len(txs))
	}
}
func (p *peer) AsyncSendTransactionHashes(txids []common.Hash, queue chan broadcastItem) {
	if p.asyncSendNonEncodedItem(txids, NewEvmTxHashesMsg, queue) {
		for _, tx := range txids {
			p.knownTxs.Add(tx)
		}
		for p.knownTxs.Cardinality() >= p.cfg.MaxKnownTxs {
			p.knownTxs.Pop()
		}
	} else {
		p.Log().Debug("Dropping tx announcement", "count", len(txids))
	}
}