func (m *DefaultTxQueueManager) loadCandidateTx(candidateSize int) ([]Transaction, error) {
	candidate := make([]Transaction, candidateSize)
	var err error
	for i := 0; i < candidateSize; i++ {
		candidate[i], err = m.txQueue.Poll()
		if err != nil {
			return nil, err
		}
	}
	return candidate, nil
}
loadCandidateTx is a function that returns candidate transactions which could be
included into contribution from the queue
package cleisthenes
import (
	"errors"
	"fmt"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
	"github.com/DE-labtory/iLogger"
)
type TxQueue interface {
	Push(tx Transaction)
	Poll() (Transaction, error)
	Len() int
}
type MemTxQueue struct {
	txs []Transaction
	sync.RWMutex
}
type emptyQueErr struct{}
func (e *emptyQueErr) Error() string {
	return "memQueue is empty."
}
type indexBoundaryErr struct {
	queSize int
	want    int
}
func (e *indexBoundaryErr) Error() string {
	return fmt.Sprintf("index is larger than memQueue size.\n"+
		"memQueue size : %d, you want : %d", e.queSize, e.want)
}
func NewTxQueue() *MemTxQueue {
	return &MemTxQueue{
		txs: []Transaction{},
	}
}
func (q *MemTxQueue) empty() bool {
	return len(q.txs) == 0
}
func (q *MemTxQueue) peek() (Transaction, error) {
	if q.empty() {
		return nil, &emptyQueErr{}
	}
	return q.txs[0], nil
}
func (q *MemTxQueue) Poll() (Transaction, error) {
	q.Lock()
	defer q.Unlock()
	ret, err := q.peek()
	if err != nil {
		return nil, err
	}
	q.txs = q.txs[1:]
	return ret, nil
}
func (q *MemTxQueue) Len() int {
	q.Lock()
	defer q.Unlock()
	return len(q.txs)
}
func (q *MemTxQueue) at(index int) (Transaction, error) {
	if index >= q.Len() {
		return nil, &indexBoundaryErr{
			queSize: q.Len(),
			want:    index,
		}
	}
	return q.txs[index], nil
}
func (q *MemTxQueue) Push(tx Transaction) {
	q.Lock()
	defer q.Unlock()
	q.txs = append(q.txs, tx)
}
type TxValidator func(Transaction) bool
type TxQueueManager interface {
	AddTransaction(tx Transaction) error
}
type DefaultTxQueueManager struct {
	txQueue TxQueue
	hb      HoneyBadger
	stopFlag int32
	contributionSize int
	batchSize        int
	closeChan chan struct{}
	tryInterval time.Duration
	txValidator TxValidator
}
func NewDefaultTxQueueManager(
	txQueue TxQueue,
	hb HoneyBadger,
	contributionSize int,
	batchSize int,
	tryInterval time.Duration,
	txVerifier TxValidator,
) *DefaultTxQueueManager {
	m := &DefaultTxQueueManager{
		txQueue:          txQueue,
		hb:               hb,
		contributionSize: contributionSize,
		batchSize:        batchSize,
		closeChan:   make(chan struct{}),
		tryInterval: tryInterval,
		txValidator: txVerifier,
	}
	go m.runContributionProposeRoutine()
	return m
}
func (m *DefaultTxQueueManager) AddTransaction(tx Transaction) error {
	if !m.txValidator(tx) {
		return errors.New(fmt.Sprintf("error invalid transaction: %v", tx))
	}
	m.txQueue.Push(tx)
	return nil
}
func (m *DefaultTxQueueManager) Close() {
	if first := atomic.CompareAndSwapInt32(&m.stopFlag, int32(0), int32(1)); !first {
		return
	}
	m.closeChan <- struct{}{}
	<-m.closeChan
}
func (m *DefaultTxQueueManager) toDie() bool {
	return atomic.LoadInt32(&(m.stopFlag)) == int32(1)
}
func (m *DefaultTxQueueManager) runContributionProposeRoutine() {
	for !m.toDie() {
		if !m.hb.OnConsensus() {
			m.tryPropose()
		}
		iLogger.Debug(nil, "[DefaultTxQueueManager] try to propose contribution...")
		time.Sleep(m.tryInterval)
	}
}
func (m *DefaultTxQueueManager) tryPropose() error {
	if m.txQueue.Len() < int(m.contributionSize) {
		return nil
	}
	contribution, err := m.createContribution()
	if err != nil {
		return err
	}
	m.hb.HandleContribution(contribution)
	return nil
}
func (m *DefaultTxQueueManager) createContribution() (Contribution, error) {
	candidate, err := m.loadCandidateTx(min(m.batchSize, m.txQueue.Len()))
	if err != nil {
		return Contribution{}, err
	}
	return Contribution{
		TxList: m.selectRandomTx(candidate, m.contributionSize),
	}, nil
}