func (mq *MessageQueue) transferRebroadcastWants() bool {
	mq.wllock.Lock()
	defer mq.wllock.Unlock()
	if mq.bcstWants.sent.Len() == 0 && mq.peerWants.sent.Len() == 0 {
		return false
	}
	mq.bcstWants.pending.Absorb(mq.bcstWants.sent)
	mq.peerWants.pending.Absorb(mq.peerWants.sent)
	return true
}
Transfer wants from the rebroadcast lists into the pending lists.
package messagequeue
import (
	"context"
	"math"
	"sync"
	"time"
	"github.com/benbjohnson/clock"
	bsmsg "github.com/ipfs/go-bitswap/message"
	pb "github.com/ipfs/go-bitswap/message/pb"
	bsnet "github.com/ipfs/go-bitswap/network"
	bswl "github.com/ipfs/go-bitswap/wantlist"
	cid "github.com/ipfs/go-cid"
	logging "github.com/ipfs/go-log"
	peer "github.com/libp2p/go-libp2p-core/peer"
	"github.com/libp2p/go-libp2p/p2p/protocol/ping"
	"go.uber.org/zap"
)
var log = logging.Logger("bitswap")
var sflog = log.Desugar()
const (
	defaultRebroadcastInterval = 30 * time.Second
	maxRetries  = 3
	sendTimeout = 30 * time.Second
	maxMessageSize = 1024 * 1024 * 2
	sendErrorBackoff = 100 * time.Millisecond
	maxPriority = math.MaxInt32
	sendMessageDebounce = time.Millisecond
	sendMessageCutoff = 256
	sendMessageMaxDelay = 20 * time.Millisecond
	maxValidLatency = 30 * time.Second
)
type MessageNetwork interface {
	ConnectTo(context.Context, peer.ID) error
	NewMessageSender(context.Context, peer.ID, *bsnet.MessageSenderOpts) (bsnet.MessageSender, error)
	Latency(peer.ID) time.Duration
	Ping(context.Context, peer.ID) ping.Result
	Self() peer.ID
}
type MessageQueue struct {
	ctx          context.Context
	shutdown     func()
	p            peer.ID
	network      MessageNetwork
	dhTimeoutMgr DontHaveTimeoutManager
	maxMessageSize int
	sendErrorBackoff time.Duration
	maxValidLatency time.Duration
	outgoingWork chan time.Time
	responses chan []cid.Cid
	wllock    sync.Mutex
	bcstWants recallWantlist
	peerWants recallWantlist
	cancels   *cid.Set
	priority  int32
	sender                bsnet.MessageSender
	rebroadcastIntervalLk sync.RWMutex
	rebroadcastInterval   time.Duration
	rebroadcastTimer      *clock.Timer
	msg bsmsg.BitSwapMessage
	clock clock.Clock
	events chan messageEvent
}
type recallWantlist struct {
	pending *bswl.Wantlist
	sent *bswl.Wantlist
	sentAt map[cid.Cid]time.Time
}
func newRecallWantList() recallWantlist {
	return recallWantlist{
		pending: bswl.New(),
		sent:    bswl.New(),
		sentAt:  make(map[cid.Cid]time.Time),
	}
}
func (r *recallWantlist) Add(c cid.Cid, priority int32, wtype pb.Message_Wantlist_WantType) {
	r.pending.Add(c, priority, wtype)
}
func (r *recallWantlist) Remove(c cid.Cid) {
	r.pending.Remove(c)
	r.sent.Remove(c)
	delete(r.sentAt, c)
}
func (r *recallWantlist) RemoveType(c cid.Cid, wtype pb.Message_Wantlist_WantType) {
	r.pending.RemoveType(c, wtype)
	r.sent.RemoveType(c, wtype)
	if _, ok := r.sent.Contains(c); !ok {
		delete(r.sentAt, c)
	}
}
func (r *recallWantlist) MarkSent(e bswl.Entry) bool {
	if !r.pending.RemoveType(e.Cid, e.WantType) {
		return false
	}
	r.sent.Add(e.Cid, e.Priority, e.WantType)
	return true
}
func (r *recallWantlist) SentAt(c cid.Cid, at time.Time) {
	if _, ok := r.sent.Contains(c); ok {
		if _, ok := r.sentAt[c]; !ok {
			r.sentAt[c] = at
		}
	}
}
func (r *recallWantlist) ClearSentAt(c cid.Cid) {
	delete(r.sentAt, c)
}
type peerConn struct {
	p       peer.ID
	network MessageNetwork
}
func newPeerConnection(p peer.ID, network MessageNetwork) *peerConn {
	return &peerConn{p, network}
}
func (pc *peerConn) Ping(ctx context.Context) ping.Result {
	return pc.network.Ping(ctx, pc.p)
}
func (pc *peerConn) Latency() time.Duration {
	return pc.network.Latency(pc.p)
}
type OnDontHaveTimeout func(peer.ID, []cid.Cid)
type DontHaveTimeoutManager interface {
	Start()
	Shutdown()
	AddPending([]cid.Cid)
	CancelPending([]cid.Cid)
	UpdateMessageLatency(time.Duration)
}
func New(ctx context.Context, p peer.ID, network MessageNetwork, onDontHaveTimeout OnDontHaveTimeout) *MessageQueue {
	onTimeout := func(ks []cid.Cid) {
		log.Infow("Bitswap: timeout waiting for blocks", "cids", ks, "peer", p)
		onDontHaveTimeout(p, ks)
	}
	clock := clock.New()
	dhTimeoutMgr := newDontHaveTimeoutMgr(newPeerConnection(p, network), onTimeout, clock)
	return newMessageQueue(ctx, p, network, maxMessageSize, sendErrorBackoff, maxValidLatency, dhTimeoutMgr, clock, nil)
}
type messageEvent int
const (
	messageQueued messageEvent = iota
	messageFinishedSending
	latenciesRecorded
)
func newMessageQueue(
	ctx context.Context,
	p peer.ID,
	network MessageNetwork,
	maxMsgSize int,
	sendErrorBackoff time.Duration,
	maxValidLatency time.Duration,
	dhTimeoutMgr DontHaveTimeoutManager,
	clock clock.Clock,
	events chan messageEvent) *MessageQueue {
	ctx, cancel := context.WithCancel(ctx)
	return &MessageQueue{
		ctx:                 ctx,
		shutdown:            cancel,
		p:                   p,
		network:             network,
		dhTimeoutMgr:        dhTimeoutMgr,
		maxMessageSize:      maxMsgSize,
		bcstWants:           newRecallWantList(),
		peerWants:           newRecallWantList(),
		cancels:             cid.NewSet(),
		outgoingWork:        make(chan time.Time, 1),
		responses:           make(chan []cid.Cid, 8),
		rebroadcastInterval: defaultRebroadcastInterval,
		sendErrorBackoff:    sendErrorBackoff,
		maxValidLatency:     maxValidLatency,
		priority:            maxPriority,
		msg:    bsmsg.New(false),
		clock:  clock,
		events: events,
	}
}
func (mq *MessageQueue) AddBroadcastWantHaves(wantHaves []cid.Cid) {
	if len(wantHaves) == 0 {
		return
	}
	mq.wllock.Lock()
	defer mq.wllock.Unlock()
	for _, c := range wantHaves {
		mq.bcstWants.Add(c, mq.priority, pb.Message_Wantlist_Have)
		mq.priority--
		mq.cancels.Remove(c)
	}
	mq.signalWorkReady()
}
func (mq *MessageQueue) AddWants(wantBlocks []cid.Cid, wantHaves []cid.Cid) {
	if len(wantBlocks) == 0 && len(wantHaves) == 0 {
		return
	}
	mq.wllock.Lock()
	defer mq.wllock.Unlock()
	for _, c := range wantHaves {
		mq.peerWants.Add(c, mq.priority, pb.Message_Wantlist_Have)
		mq.priority--
		mq.cancels.Remove(c)
	}
	for _, c := range wantBlocks {
		mq.peerWants.Add(c, mq.priority, pb.Message_Wantlist_Block)
		mq.priority--
		mq.cancels.Remove(c)
	}
	mq.signalWorkReady()
}
func (mq *MessageQueue) AddCancels(cancelKs []cid.Cid) {
	if len(cancelKs) == 0 {
		return
	}
	mq.dhTimeoutMgr.CancelPending(cancelKs)
	mq.wllock.Lock()
	workReady := false
	for _, c := range cancelKs {
		_, wasSentBcst := mq.bcstWants.sent.Contains(c)
		_, wasSentPeer := mq.peerWants.sent.Contains(c)
		mq.bcstWants.Remove(c)
		mq.peerWants.Remove(c)
		if wasSentBcst || wasSentPeer {
			mq.cancels.Add(c)
			workReady = true
		}
	}
	mq.wllock.Unlock()
	if workReady {
		mq.signalWorkReady()
	}
}
func (mq *MessageQueue) ResponseReceived(ks []cid.Cid) {
	if len(ks) == 0 {
		return
	}
	select {
	case mq.responses <- ks:
	default:
	}
}
func (mq *MessageQueue) SetRebroadcastInterval(delay time.Duration) {
	mq.rebroadcastIntervalLk.Lock()
	mq.rebroadcastInterval = delay
	if mq.rebroadcastTimer != nil {
		mq.rebroadcastTimer.Reset(delay)
	}
	mq.rebroadcastIntervalLk.Unlock()
}
func (mq *MessageQueue) Startup() {
	mq.rebroadcastIntervalLk.RLock()
	mq.rebroadcastTimer = mq.clock.Timer(mq.rebroadcastInterval)
	mq.rebroadcastIntervalLk.RUnlock()
	go mq.runQueue()
}
func (mq *MessageQueue) Shutdown() {
	mq.shutdown()
}
func (mq *MessageQueue) onShutdown() {
	mq.dhTimeoutMgr.Shutdown()
	if mq.sender != nil {
		_ = mq.sender.Reset()
	}
}
func (mq *MessageQueue) runQueue() {
	defer mq.onShutdown()
	scheduleWork := mq.clock.Timer(0)
	if !scheduleWork.Stop() {
		<-scheduleWork.C
	}
	var workScheduled time.Time
	for mq.ctx.Err() == nil {
		select {
		case <-mq.rebroadcastTimer.C:
			mq.rebroadcastWantlist()
		case when := <-mq.outgoingWork:
			if workScheduled.IsZero() {
				workScheduled = when
			} else if !scheduleWork.Stop() {
				<-scheduleWork.C
			}
			if mq.pendingWorkCount() > sendMessageCutoff ||
				mq.clock.Since(workScheduled) >= sendMessageMaxDelay {
				mq.sendIfReady()
				workScheduled = time.Time{}
			} else {
				scheduleWork.Reset(sendMessageDebounce)
				if mq.events != nil {
					mq.events <- messageQueued
				}
			}
		case <-scheduleWork.C:
			workScheduled = time.Time{}
			mq.sendIfReady()
		case res := <-mq.responses:
			mq.handleResponse(res)
		case <-mq.ctx.Done():
			return
		}
	}
}
func (mq *MessageQueue) rebroadcastWantlist() {
	mq.rebroadcastIntervalLk.RLock()
	mq.rebroadcastTimer.Reset(mq.rebroadcastInterval)
	mq.rebroadcastIntervalLk.RUnlock()
	if mq.transferRebroadcastWants() {
		mq.sendMessage()
	}
}