fluencelabs/tendermint
1
0
Fork 0
mirror of https://github.com/fluencelabs/tendermint synced 2025-04-25 14:52:17 +00:00
Code Issues Projects Releases Wiki Activity

Fix mempool

Browse Source
This commit is contained in:
Jae Kwon 2018-03-31 11:51:32 +02:00
parent 34974e3932
commit 35a1d747b0
2 changed files with 20 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
mempool/reactor.go
View File

@ -1,13 +1,12 @@
package mempool package mempool
import ( import (
"bytes"
"fmt" "fmt"
"reflect" "reflect"
"time" "time"
abci "github.com/tendermint/abci/types" abci "github.com/tendermint/abci/types"
wire "github.com/tendermint/go-wire" "github.com/tendermint/go-amino"
"github.com/tendermint/tmlibs/clist" "github.com/tendermint/tmlibs/clist"
"github.com/tendermint/tmlibs/log" "github.com/tendermint/tmlibs/log"
@ -71,7 +70,7 @@ func (memR *MempoolReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
// Receive implements Reactor. // Receive implements Reactor.
// It adds any received transactions to the mempool. // It adds any received transactions to the mempool.
func (memR *MempoolReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) { func (memR *MempoolReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
_, msg, err := DecodeMessage(msgBytes) msg, err := DecodeMessage(msgBytes)
if err != nil { if err != nil {
memR.Logger.Error("Error decoding message", "src", src, "chId", chID, "msg", msg, "err", err, "bytes", msgBytes) memR.Logger.Error("Error decoding message", "src", src, "chId", chID, "msg", msg, "err", err, "bytes", msgBytes)
memR.Switch.StopPeerForError(src, err) memR.Switch.StopPeerForError(src, err)
@ -137,7 +136,7 @@ func (memR *MempoolReactor) broadcastTxRoutine(peer p2p.Peer) {
} }
// send memTx // send memTx
msg := &TxMessage{Tx: memTx.tx} msg := &TxMessage{Tx: memTx.tx}
success := peer.Send(MempoolChannel, struct{ MempoolMessage }{msg}) success := peer.Send(MempoolChannel, cdc.MustMarshalBinaryBare(msg))
if !success { if !success {
time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond) time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
continue continue
@ -158,24 +157,17 @@ func (memR *MempoolReactor) broadcastTxRoutine(peer p2p.Peer) {
//----------------------------------------------------------------------------- //-----------------------------------------------------------------------------
// Messages // Messages
const (
msgTypeTx = byte(0x01)
)
// MempoolMessage is a message sent or received by the MempoolReactor. // MempoolMessage is a message sent or received by the MempoolReactor.
type MempoolMessage interface{} type MempoolMessage interface{}
var _ = wire.RegisterInterface( func RegisterMempoolMessages(cdc *amino.Codec) {
struct{ MempoolMessage }{}, cdc.RegisterInterface((*MempoolMessage)(nil), nil)
wire.ConcreteType{&TxMessage{}, msgTypeTx}, cdc.RegisterConcrete(&TxMessage{}, "tendermint/mempool/TxMessage", nil)
) }
// DecodeMessage decodes a byte-array into a MempoolMessage. // DecodeMessage decodes a byte-array into a MempoolMessage.
func DecodeMessage(bz []byte) (msgType byte, msg MempoolMessage, err error) { func DecodeMessage(bz []byte) (msg MempoolMessage, err error) {
msgType = bz[0] err = cdc.UnmarshalBinaryBare(bz, &msg)
n := new(int)
r := bytes.NewReader(bz)
msg = wire.ReadBinary(struct{ MempoolMessage }{}, r, maxMempoolMessageSize, n, &err).(struct{ MempoolMessage }).MempoolMessage
return return
} }

11
mempool/wire.go Normal file
View File

@ -0,0 +1,11 @@
package mempool
import (
"github.com/tendermint/go-amino"
)
var cdc = amino.NewCodec()
func init() {
RegisterMempoolMessages(cdc)
}