func (mem *Mempool) extractTxs() []types.BlockTx {
	if mem.txs == nil {
		return nil
	}
	var ret []types.BlockTx
	for e := mem.txs.Front(); e != nil; e = e.Next() {
		tx := e.Value.(*mempoolTx).tx
		var blockTx types.BlockTx
		err := rlp.DecodeBytes([]byte(tx), &blockTx)
		if err != nil {
			continue
		}
		ret = append(ret, blockTx)
	}
	return ret
}
extractTxs extracts BlockTxs from Mempool
package mempool
import (
	"container/list"
	"errors"
	"sync"
	"sync/atomic"
	"github.com/spf13/viper"
	"go.uber.org/zap"
	"time"
	agtypes "github.com/annchain/annchain/angine/types"
	auto "github.com/annchain/annchain/module/lib/go-autofile"
	"github.com/annchain/annchain/module/lib/go-clist"
	cmn "github.com/annchain/annchain/module/lib/go-common"
	"github.com/annchain/annchain/module/xlib/def"
	"github.com/annchain/annchain/types"
	"github.com/annchain/anth/rlp"
)
const cacheSize = 100000
type IFilter interface {
	CheckTx(agtypes.Tx) (bool, error)
}
type Mempool struct {
	config  *viper.Viper
	mtx     sync.Mutex
	txs     *clist.CList  
	counter def.INT       
	height  def.INT       
	cache *txCache
	wal *auto.AutoFile  
	txLimit int
	txFilters []IFilter
	logger *zap.Logger
}
func NewMempool(logger *zap.Logger, config *viper.Viper) *Mempool {
	mempool := &Mempool{
		config:  config,
		txs:     clist.New(),
		counter: 0,
		height:  0,
		cache:   newTxCache(cacheSize),
		txLimit: config.GetInt("block_max_txs") * 2,
		logger:  logger,
	}
	mempool.initWAL()
	blockSortInterval := config.GetInt("mempool_block_sort_interval")
	if blockSortInterval > 0 {
		go func() {
			t := time.Tick(time.Millisecond * time.Duration(blockSortInterval))
			for {
				select {
				case <-t:
					mempool.sortTxs()
				}
			}
		}()
	}
	return mempool
}
func (mem *Mempool) RegisterFilter(filter IFilter) {
	mem.txFilters = append(mem.txFilters, filter)
}
func (mem *Mempool) Lock() {
	mem.mtx.Lock()
}
func (mem *Mempool) Unlock() {
	mem.mtx.Unlock()
}
func (mem *Mempool) Size() int {
	return mem.txs.Len()
}
func (mem *Mempool) Flush() {
	mem.Lock()
	mem.cache.Reset()
	for e := mem.txs.Front(); e != nil; e = e.Next() {
		mem.txs.Remove(e)
		e.DetachPrev()
	}
	mem.Unlock()
}
func (mem *Mempool) TxsFrontWait() *clist.CElement {
	return mem.txs.FrontWait()
}
func (mem *Mempool) CheckTx(tx agtypes.Tx) (err error) {
	if mem.cache.Exists(tx) {
		return errors.New("Duplicate transaction (ignored)")
	}
	if mem.config.GetBool("mempool_enable_txs_limits") && mem.txs.Len() > mem.txLimit {
		return errors.New("Too many unsolved TX (rejected)")
	}
	if err := mem.checkTxWithFilters(tx); err != nil {
		return errors.New("plugin checktx failed with error: " + err.Error())
	}
	if mem.wal != nil {
		mem.wal.Write([]byte(tx))
		mem.wal.Write([]byte("\n"))
	}
	mem.cache.Push(tx)
	nc := atomic.AddInt64(&mem.counter, 1)
	memTx := &mempoolTx{
		counter: nc,
		height:  atomic.LoadInt64(&mem.height),
		tx:      tx,
	}
	mem.txs.PushBack(memTx)
	return nil
}
func (mem *Mempool) Reap(maxTxs int) []agtypes.Tx {
	mem.Lock()
	txs := mem.collectTxs(maxTxs)
	mem.Unlock()
	return txs
}
func (mem *Mempool) Update(height def.INT, txs []agtypes.Tx) {
	txsMap := make(map[string]struct{})
	for _, tx := range txs {
		txsMap[string(tx)] = struct{}{}
	}
	atomic.StoreInt64(&mem.height, height)
	mem.Lock()
	mem.refreshMempoolTxs(txsMap)
	mem.Unlock()
}
func (mem *Mempool) collectTxs(maxTxs int) []agtypes.Tx {
	if maxTxs == 0 {
		return []agtypes.Tx{}
	} else if maxTxs < 0 {
		maxTxs = mem.txs.Len()
	} else {
		maxTxs = cmn.MinInt(mem.txs.Len(), maxTxs)
	}
	totalSize := 0
	maxSize := mem.config.GetInt("block_max_size")
	txs := make([]agtypes.Tx, 0, maxTxs)
	for e := mem.txs.Front(); e != nil && len(txs) < maxTxs; e = e.Next() {
		memTx := e.Value.(*mempoolTx)
		totalSize += len(memTx.tx)
		if totalSize >= maxSize {
			break
		}
		txs = append(txs, memTx.tx)
	}
	return txs
}
func (mem *Mempool) refreshMempoolTxs(blockTxsMap map[string]struct{}) {
	txsLen := mem.txs.Len()
	index := 0
	for e := mem.txs.Front(); e != nil && index < txsLen; e = e.Next() {
		index++
		memTx := e.Value.(*mempoolTx)
		if _, ok := blockTxsMap[string(memTx.tx)]; ok {
			mem.txs.Remove(e)
			e.DetachPrev()
		} else if err := mem.recheckTx(memTx.tx); err != nil {
			mem.txs.Remove(e)
			e.DetachPrev()
		}
	}
}
func (mem *Mempool) recheckTx(tx agtypes.Tx) error {
	return mem.checkTxWithFilters(tx)
}
func (mem *Mempool) checkTxWithFilters(tx agtypes.Tx) error {
	if mem.txFilters == nil || len(mem.txFilters) == 0 {
		return nil
	}
	for _, p := range mem.txFilters {
		_, err := p.CheckTx(tx)
		if err != nil {
			return err
		}
	}
	return nil
}
func (mem *Mempool) initWAL() {
	walDir := mem.config.GetString("mempool_wal_dir")
	if walDir != "" {
		err := cmn.EnsureDir(walDir, 0700)
		if err != nil {
			cmn.PanicSanity(err)
		}
		af, err := auto.OpenAutoFile(walDir + "/wal")
		if err != nil {
			cmn.PanicSanity(err)
		}
		mem.wal = af
	}
}