aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--eth/backend.go4
-rw-r--r--eth/protocol.go95
-rw-r--r--eth/protocol_test.go189
-rw-r--r--p2p/handshake.go8
-rw-r--r--p2p/message.go77
-rw-r--r--p2p/message_test.go24
-rw-r--r--p2p/peer.go7
-rw-r--r--p2p/peer_test.go73
-rw-r--r--p2p/rlpx_test.go4
-rw-r--r--p2p/server.go11
-rw-r--r--p2p/server_test.go2
-rw-r--r--whisper/envelope.go2
-rw-r--r--whisper/peer.go9
13 files changed, 227 insertions, 278 deletions
diff --git a/eth/backend.go b/eth/backend.go
index 9e5324688..afe314d74 100644
--- a/eth/backend.go
+++ b/eth/backend.go
@@ -405,7 +405,7 @@ func (self *Ethereum) txBroadcastLoop() {
// automatically stops if unsubscribe
for obj := range self.txSub.Chan() {
event := obj.(core.TxPreEvent)
- self.net.Broadcast("eth", TxMsg, event.Tx.RlpData())
+ self.net.Broadcast("eth", TxMsg, []*types.Transaction{event.Tx})
}
}
@@ -414,7 +414,7 @@ func (self *Ethereum) blockBroadcastLoop() {
for obj := range self.blockSub.Chan() {
switch ev := obj.(type) {
case core.NewMinedBlockEvent:
- self.net.Broadcast("eth", NewBlockMsg, ev.Block.RlpData(), ev.Block.Td)
+ self.net.Broadcast("eth", NewBlockMsg, []interface{}{ev.Block, ev.Block.Td})
}
}
}
diff --git a/eth/protocol.go b/eth/protocol.go
index a1c2ae688..6d610a663 100644
--- a/eth/protocol.go
+++ b/eth/protocol.go
@@ -89,7 +89,7 @@ type blockPool interface {
RemovePeer(peerId string)
}
-// message structs used for rlp decoding
+// message structs used for RLP serialization
type newBlockMsgData struct {
Block *types.Block
TD *big.Int
@@ -100,6 +100,14 @@ type getBlockHashesMsgData struct {
Amount uint64
}
+type statusMsgData struct {
+ ProtocolVersion uint32
+ NetworkId uint32
+ TD *big.Int
+ CurrentBlock common.Hash
+ GenesisBlock common.Hash
+}
+
// main entrypoint, wrappers starting a server running the eth protocol
// use this constructor to attach the protocol ("class") to server caps
// the Dev p2p layer then runs the protocol instance on each peer
@@ -132,18 +140,25 @@ func runEthProtocol(protocolVersion, networkId int, txPool txPool, chainManager
},
id: fmt.Sprintf("%x", id[:8]),
}
- err = self.handleStatus()
- if err == nil {
- self.propagateTxs()
- for {
- err = self.handle()
- if err != nil {
- self.blockPool.RemovePeer(self.id)
- break
- }
+
+ // handshake.
+ if err := self.handleStatus(); err != nil {
+ return err
+ }
+ defer self.blockPool.RemovePeer(self.id)
+
+ // propagate existing transactions. new transactions appearing
+ // after this will be sent via broadcasts.
+ if err := p2p.Send(rw, TxMsg, txPool.GetTransactions()); err != nil {
+ return err
+ }
+
+ // main loop. handle incoming messages.
+ for {
+ if err := self.handle(); err != nil {
+ return err
}
}
- return
}
func (self *ethProtocol) handle() error {
@@ -186,7 +201,7 @@ func (self *ethProtocol) handle() error {
request.Amount = maxHashes
}
hashes := self.chainManager.GetBlockHashesFromHash(request.Hash, request.Amount)
- return p2p.EncodeMsg(self.rw, BlockHashesMsg, rlp.Flat(hashes))
+ return p2p.Send(self.rw, BlockHashesMsg, hashes)
case BlockHashesMsg:
msgStream := rlp.NewStream(msg.Payload)
@@ -216,7 +231,7 @@ func (self *ethProtocol) handle() error {
return err
}
- var blocks []interface{}
+ var blocks []*types.Block
var i int
for {
i++
@@ -224,7 +239,7 @@ func (self *ethProtocol) handle() error {
err := msgStream.Decode(&hash)
if err == rlp.EOL {
break
- } else {
+ } else if err != nil {
return self.protoError(ErrDecode, "msg %v: %v", msg, err)
}
@@ -236,7 +251,7 @@ func (self *ethProtocol) handle() error {
break
}
}
- return p2p.EncodeMsg(self.rw, BlocksMsg, blocks...)
+ return p2p.Send(self.rw, BlocksMsg, blocks)
case BlocksMsg:
msgStream := rlp.NewStream(msg.Payload)
@@ -283,29 +298,8 @@ func (self *ethProtocol) handle() error {
return nil
}
-type statusMsgData struct {
- ProtocolVersion uint32
- NetworkId uint32
- TD *big.Int
- CurrentBlock common.Hash
- GenesisBlock common.Hash
-}
-
-func (self *ethProtocol) statusMsg() p2p.Msg {
- td, currentBlock, genesisBlock := self.chainManager.Status()
-
- return p2p.NewMsg(StatusMsg,
- uint32(self.protocolVersion),
- uint32(self.networkId),
- td,
- currentBlock,
- genesisBlock,
- )
-}
-
func (self *ethProtocol) handleStatus() error {
- // send precanned status message
- if err := self.rw.WriteMsg(self.statusMsg()); err != nil {
+ if err := self.sendStatus(); err != nil {
return err
}
@@ -314,11 +308,9 @@ func (self *ethProtocol) handleStatus() error {
if err != nil {
return err
}
-
if msg.Code != StatusMsg {
return self.protoError(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg)
}
-
if msg.Size > ProtocolMaxMsgSize {
return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
}
@@ -351,12 +343,12 @@ func (self *ethProtocol) handleStatus() error {
func (self *ethProtocol) requestBlockHashes(from common.Hash) error {
self.peer.Debugf("fetching hashes (%d) %x...\n", maxHashes, from[0:4])
- return p2p.EncodeMsg(self.rw, GetBlockHashesMsg, interface{}(from), uint64(maxHashes))
+ return p2p.Send(self.rw, GetBlockHashesMsg, getBlockHashesMsgData{from, maxHashes})
}
func (self *ethProtocol) requestBlocks(hashes []common.Hash) error {
self.peer.Debugf("fetching %v blocks", len(hashes))
- return p2p.EncodeMsg(self.rw, GetBlocksMsg, rlp.Flat(hashes))
+ return p2p.Send(self.rw, GetBlocksMsg, hashes)
}
func (self *ethProtocol) protoError(code int, format string, params ...interface{}) (err *errs.Error) {
@@ -365,19 +357,20 @@ func (self *ethProtocol) protoError(code int, format string, params ...interface
return
}
+func (self *ethProtocol) sendStatus() error {
+ td, currentBlock, genesisBlock := self.chainManager.Status()
+ return p2p.Send(self.rw, StatusMsg, &statusMsgData{
+ ProtocolVersion: uint32(self.protocolVersion),
+ NetworkId: uint32(self.networkId),
+ TD: td,
+ CurrentBlock: currentBlock,
+ GenesisBlock: genesisBlock,
+ })
+}
+
func (self *ethProtocol) protoErrorDisconnect(err *errs.Error) {
err.Log(self.peer.Logger)
if err.Fatal() {
self.peer.Disconnect(p2p.DiscSubprotocolError)
}
}
-
-func (self *ethProtocol) propagateTxs() {
- transactions := self.txPool.GetTransactions()
- iface := make([]interface{}, len(transactions))
- for i, transaction := range transactions {
- iface[i] = transaction
- }
-
- self.rw.WriteMsg(p2p.NewMsg(TxMsg, iface...))
-}
diff --git a/eth/protocol_test.go b/eth/protocol_test.go
index 108fb4475..7620b3854 100644
--- a/eth/protocol_test.go
+++ b/eth/protocol_test.go
@@ -1,8 +1,6 @@
package eth
import (
- "bytes"
- "io"
"log"
"math/big"
"os"
@@ -29,52 +27,21 @@ func logInit() {
}
}
-type testMsgReadWriter struct {
- in chan p2p.Msg
- out []p2p.Msg
-}
-
-func (self *testMsgReadWriter) In(msg p2p.Msg) {
- self.in <- msg
-}
-
-func (self *testMsgReadWriter) Out() (msg p2p.Msg, ok bool) {
- if len(self.out) > 0 {
- msg = self.out[0]
- self.out = self.out[1:]
- ok = true
- }
- return
-}
-
-func (self *testMsgReadWriter) WriteMsg(msg p2p.Msg) error {
- self.out = append(self.out, msg)
- return nil
-}
-
-func (self *testMsgReadWriter) ReadMsg() (p2p.Msg, error) {
- msg, ok := <-self.in
- if !ok {
- return msg, io.EOF
- }
- return msg, nil
-}
-
type testTxPool struct {
getTransactions func() []*types.Transaction
addTransactions func(txs []*types.Transaction)
}
type testChainManager struct {
- getBlockHashes func(hash []byte, amount uint64) (hashes [][]byte)
- getBlock func(hash []byte) *types.Block
- status func() (td *big.Int, currentBlock []byte, genesisBlock []byte)
+ getBlockHashes func(hash common.Hash, amount uint64) (hashes []common.Hash)
+ getBlock func(hash common.Hash) *types.Block
+ status func() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash)
}
type testBlockPool struct {
- addBlockHashes func(next func() ([]byte, bool), peerId string)
+ addBlockHashes func(next func() (common.Hash, bool), peerId string)
addBlock func(block *types.Block, peerId string) (err error)
- addPeer func(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(*errs.Error)) (best bool)
+ addPeer func(td *big.Int, currentBlock common.Hash, peerId string, requestHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool)
removePeer func(peerId string)
}
@@ -93,28 +60,28 @@ func (self *testTxPool) AddTransactions(txs []*types.Transaction) {
func (self *testTxPool) GetTransactions() types.Transactions { return nil }
-func (self *testChainManager) GetBlockHashesFromHash(hash []byte, amount uint64) (hashes [][]byte) {
+func (self *testChainManager) GetBlockHashesFromHash(hash common.Hash, amount uint64) (hashes []common.Hash) {
if self.getBlockHashes != nil {
hashes = self.getBlockHashes(hash, amount)
}
return
}
-func (self *testChainManager) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) {
+func (self *testChainManager) Status() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash) {
if self.status != nil {
td, currentBlock, genesisBlock = self.status()
}
return
}
-func (self *testChainManager) GetBlock(hash []byte) (block *types.Block) {
+func (self *testChainManager) GetBlock(hash common.Hash) (block *types.Block) {
if self.getBlock != nil {
block = self.getBlock(hash)
}
return
}
-func (self *testBlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) {
+func (self *testBlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId string) {
if self.addBlockHashes != nil {
self.addBlockHashes(next, peerId)
}
@@ -126,7 +93,7 @@ func (self *testBlockPool) AddBlock(block *types.Block, peerId string) {
}
}
-func (self *testBlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(*errs.Error)) (best bool) {
+func (self *testBlockPool) AddPeer(td *big.Int, currentBlock common.Hash, peerId string, requestBlockHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool) {
if self.addPeer != nil {
best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, peerError)
}
@@ -147,28 +114,36 @@ func testPeer() *p2p.Peer {
}
type ethProtocolTester struct {
+ p2p.MsgReadWriter // writing to the tester feeds the protocol
+
quit chan error
- rw *testMsgReadWriter // p2p.MsgReadWriter
- txPool *testTxPool // txPool
- chainManager *testChainManager // chainManager
- blockPool *testBlockPool // blockPool
+ pipe *p2p.MsgPipeRW // the protocol read/writes on this end
+ txPool *testTxPool // txPool
+ chainManager *testChainManager // chainManager
+ blockPool *testBlockPool // blockPool
t *testing.T
}
func newEth(t *testing.T) *ethProtocolTester {
+ p1, p2 := p2p.MsgPipe()
return &ethProtocolTester{
- quit: make(chan error),
- rw: &testMsgReadWriter{in: make(chan p2p.Msg, 10)},
- txPool: &testTxPool{},
- chainManager: &testChainManager{},
- blockPool: &testBlockPool{},
- t: t,
+ MsgReadWriter: p1,
+ quit: make(chan error, 1),
+ pipe: p2,
+ txPool: &testTxPool{},
+ chainManager: &testChainManager{},
+ blockPool: &testBlockPool{},
+ t: t,
}
}
func (self *ethProtocolTester) reset() {
- self.rw = &testMsgReadWriter{in: make(chan p2p.Msg, 10)}
- self.quit = make(chan error)
+ self.pipe.Close()
+
+ p1, p2 := p2p.MsgPipe()
+ self.MsgReadWriter = p1
+ self.pipe = p2
+ self.quit = make(chan error, 1)
}
func (self *ethProtocolTester) checkError(expCode int, delay time.Duration) (err error) {
@@ -190,33 +165,8 @@ func (self *ethProtocolTester) checkError(expCode int, delay time.Duration) (err
return
}
-func (self *ethProtocolTester) In(msg p2p.Msg) {
- self.rw.In(msg)
-}
-
-func (self *ethProtocolTester) Out() (p2p.Msg, bool) {
- return self.rw.Out()
-}
-
-func (self *ethProtocolTester) checkMsg(i int, code uint64, val interface{}) (msg p2p.Msg) {
- if i >= len(self.rw.out) {
- self.t.Errorf("expected at least %v msgs, got %v", i, len(self.rw.out))
- return
- }
- msg = self.rw.out[i]
- if msg.Code != code {
- self.t.Errorf("expected msg code %v, got %v", code, msg.Code)
- }
- if val != nil {
- if err := msg.Decode(val); err != nil {
- self.t.Errorf("rlp encoding error: %v", err)
- }
- }
- return
-}
-
func (self *ethProtocolTester) run() {
- err := runEthProtocol(ProtocolVersion, NetworkId, self.txPool, self.chainManager, self.blockPool, testPeer(), self.rw)
+ err := runEthProtocol(ProtocolVersion, NetworkId, self.txPool, self.chainManager, self.blockPool, testPeer(), self.pipe)
self.quit <- err
}
@@ -224,41 +174,52 @@ func TestStatusMsgErrors(t *testing.T) {
logInit()
eth := newEth(t)
td := common.Big1
- currentBlock := []byte{1}
- genesis := []byte{2}
- eth.chainManager.status = func() (*big.Int, []byte, []byte) { return td, currentBlock, genesis }
- go eth.run()
- statusMsg := p2p.NewMsg(4)
- eth.In(statusMsg)
- delay := 1 * time.Second
- eth.checkError(ErrNoStatusMsg, delay)
- var status statusMsgData
- eth.checkMsg(0, StatusMsg, &status) // first outgoing msg should be StatusMsg
- if status.TD.Cmp(td) != 0 ||
- status.ProtocolVersion != ProtocolVersion ||
- status.NetworkId != NetworkId ||
- status.TD.Cmp(td) != 0 ||
- bytes.Compare(status.CurrentBlock, currentBlock) != 0 ||
- bytes.Compare(status.GenesisBlock, genesis) != 0 {
- t.Errorf("incorrect outgoing status")
- }
-
- eth.reset()
+ currentBlock := common.Hash{1}
+ genesis := common.Hash{2}
+ eth.chainManager.status = func() (*big.Int, common.Hash, common.Hash) { return td, currentBlock, genesis }
go eth.run()
- statusMsg = p2p.NewMsg(0, uint32(48), uint32(0), td, currentBlock, genesis)
- eth.In(statusMsg)
- eth.checkError(ErrProtocolVersionMismatch, delay)
- eth.reset()
- go eth.run()
- statusMsg = p2p.NewMsg(0, uint32(49), uint32(1), td, currentBlock, genesis)
- eth.In(statusMsg)
- eth.checkError(ErrNetworkIdMismatch, delay)
+ tests := []struct {
+ code uint64
+ data interface{}
+ wantErrorCode int
+ }{
+ {
+ code: TxMsg, data: []interface{}{},
+ wantErrorCode: ErrNoStatusMsg,
+ },
+ {
+ code: StatusMsg, data: statusMsgData{10, NetworkId, td, currentBlock, genesis},
+ wantErrorCode: ErrProtocolVersionMismatch,
+ },
+ {
+ code: StatusMsg, data: statusMsgData{ProtocolVersion, 999, td, currentBlock, genesis},
+ wantErrorCode: ErrNetworkIdMismatch,
+ },
+ {
+ code: StatusMsg, data: statusMsgData{ProtocolVersion, NetworkId, td, currentBlock, common.Hash{3}},
+ wantErrorCode: ErrGenesisBlockMismatch,
+ },
+ }
+ for _, test := range tests {
+ // first outgoing msg should be StatusMsg.
+ err := p2p.ExpectMsg(eth, StatusMsg, &statusMsgData{
+ ProtocolVersion: ProtocolVersion,
+ NetworkId: NetworkId,
+ TD: td,
+ CurrentBlock: currentBlock,
+ GenesisBlock: genesis,
+ })
+ if err != nil {
+ t.Fatalf("incorrect outgoing status: %v", err)
+ }
- eth.reset()
- go eth.run()
- statusMsg = p2p.NewMsg(0, uint32(49), uint32(0), td, currentBlock, []byte{3})
- eth.In(statusMsg)
- eth.checkError(ErrGenesisBlockMismatch, delay)
+ // the send call might hang until reset because
+ // the protocol might not read the payload.
+ go p2p.Send(eth, test.code, test.data)
+ eth.checkError(test.wantErrorCode, 1*time.Second)
+ eth.reset()
+ go eth.run()
+ }
}
diff --git a/p2p/handshake.go b/p2p/handshake.go
index 7fc497517..031064407 100644
--- a/p2p/handshake.go
+++ b/p2p/handshake.go
@@ -92,7 +92,7 @@ func setupInboundConn(fd net.Conn, prv *ecdsa.PrivateKey, our *protoHandshake) (
return nil, errors.New("node ID in protocol handshake does not match encryption handshake")
}
// TODO: validate that handshake node ID matches
- if err := writeProtocolHandshake(rw, our); err != nil {
+ if err := Send(rw, handshakeMsg, our); err != nil {
return nil, fmt.Errorf("protocol write error: %v", err)
}
return &conn{rw, rhs}, nil
@@ -106,7 +106,7 @@ func setupOutboundConn(fd net.Conn, prv *ecdsa.PrivateKey, our *protoHandshake,
// Run the protocol handshake using authenticated messages.
rw := newRlpxFrameRW(fd, secrets)
- if err := writeProtocolHandshake(rw, our); err != nil {
+ if err := Send(rw, handshakeMsg, our); err != nil {
return nil, fmt.Errorf("protocol write error: %v", err)
}
rhs, err := readProtocolHandshake(rw, our)
@@ -398,10 +398,6 @@ func xor(one, other []byte) (xor []byte) {
return xor
}
-func writeProtocolHandshake(w MsgWriter, our *protoHandshake) error {
- return EncodeMsg(w, handshakeMsg, our.Version, our.Name, our.Caps, our.ListenPort, our.ID[:])
-}
-
func readProtocolHandshake(r MsgReader, our *protoHandshake) (*protoHandshake, error) {
// read and handle remote handshake
msg, err := r.ReadMsg()
diff --git a/p2p/message.go b/p2p/message.go
index 14e4404c9..ef3630a90 100644
--- a/p2p/message.go
+++ b/p2p/message.go
@@ -11,7 +11,6 @@ import (
"sync/atomic"
"time"
- "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/rlp"
)
@@ -28,13 +27,7 @@ type Msg struct {
Payload io.Reader
}
-// NewMsg creates an RLP-encoded message with the given code.
-func NewMsg(code uint64, params ...interface{}) Msg {
- p := bytes.NewReader(common.Encode(params))
- return Msg{Code: code, Size: uint32(p.Len()), Payload: p}
-}
-
-// Decode parse the RLP content of a message into
+// Decode parses the RLP content of a message into
// the given value, which must be a pointer.
//
// For the decoding rules, please see package rlp.
@@ -76,13 +69,30 @@ type MsgReadWriter interface {
MsgWriter
}
-// EncodeMsg writes an RLP-encoded message with the given code and
-// data elements.
-func EncodeMsg(w MsgWriter, code uint64, data ...interface{}) error {
- return w.WriteMsg(NewMsg(code, data...))
+// Send writes an RLP-encoded message with the given code.
+// data should encode as an RLP list.
+func Send(w MsgWriter, msgcode uint64, data interface{}) error {
+ size, r, err := rlp.EncodeToReader(data)
+ if err != nil {
+ return err
+ }
+ return w.WriteMsg(Msg{Code: msgcode, Size: uint32(size), Payload: r})
+}
+
+// SendItems writes an RLP with the given code and data elements.
+// For a call such as:
+//
+// SendItems(w, code, e1, e2, e3)
+//
+// the message payload will be an RLP list containing the items:
+//
+// [e1, e2, e3]
+//
+func SendItems(w MsgWriter, msgcode uint64, elems ...interface{}) error {
+ return Send(w, msgcode, elems)
}
-// netWrapper wrapsa MsgReadWriter with locks around
+// netWrapper wraps a MsgReadWriter with locks around
// ReadMsg/WriteMsg and applies read/write deadlines.
type netWrapper struct {
rmu, wmu sync.Mutex
@@ -175,7 +185,10 @@ func (p *MsgPipeRW) WriteMsg(msg Msg) error {
case p.w <- msg:
if msg.Size > 0 {
// wait for payload read or discard
- <-consumed
+ select {
+ case <-consumed:
+ case <-p.closing:
+ }
}
return nil
case <-p.closing:
@@ -197,8 +210,8 @@ func (p *MsgPipeRW) ReadMsg() (Msg, error) {
}
// Close unblocks any pending ReadMsg and WriteMsg calls on both ends
-// of the pipe. They will return ErrPipeClosed. Note that Close does
-// not interrupt any reads from a message payload.
+// of the pipe. They will return ErrPipeClosed. Close also
+// interrupts any reads from a message payload.
func (p *MsgPipeRW) Close() error {
if atomic.AddInt32(p.closed, 1) != 1 {
// someone else is already closing
@@ -208,3 +221,35 @@ func (p *MsgPipeRW) Close() error {
close(p.closing)
return nil
}
+
+// ExpectMsg reads a message from r and verifies that its
+// code and encoded RLP content match the provided values.
+// If content is nil, the payload is discarded and not verified.
+func ExpectMsg(r MsgReader, code uint64, content interface{}) error {
+ msg, err := r.ReadMsg()
+ if err != nil {
+ return err
+ }
+ if msg.Code != code {
+ return fmt.Errorf("message code mismatch: got %d, expected %d", msg.Code, code)
+ }
+ if content == nil {
+ return msg.Discard()
+ } else {
+ contentEnc, err := rlp.EncodeToBytes(content)
+ if err != nil {
+ panic("content encode error: " + err.Error())
+ }
+ if int(msg.Size) != len(contentEnc) {
+ return fmt.Errorf("message size mismatch: got %d, want %d", msg.Size, len(contentEnc))
+ }
+ actualContent, err := ioutil.ReadAll(msg.Payload)
+ if err != nil {
+ return err
+ }
+ if !bytes.Equal(actualContent, contentEnc) {
+ return fmt.Errorf("message payload mismatch:\ngot: %x\nwant: %x", actualContent, contentEnc)
+ }
+ }
+ return nil
+}
diff --git a/p2p/message_test.go b/p2p/message_test.go
index 31ed61d87..9a93dd347 100644
--- a/p2p/message_test.go
+++ b/p2p/message_test.go
@@ -5,33 +5,17 @@ import (
"encoding/hex"
"fmt"
"io"
- "io/ioutil"
"runtime"
"strings"
"testing"
"time"
)
-func TestNewMsg(t *testing.T) {
- msg := NewMsg(3, 1, "000")
- if msg.Code != 3 {
- t.Errorf("incorrect code %d, want %d", msg.Code)
- }
- expect := unhex("c50183303030")
- if msg.Size != uint32(len(expect)) {
- t.Errorf("incorrect size %d, want %d", msg.Size, len(expect))
- }
- pl, _ := ioutil.ReadAll(msg.Payload)
- if !bytes.Equal(pl, expect) {
- t.Errorf("incorrect payload content, got %x, want %x", pl, expect)
- }
-}
-
func ExampleMsgPipe() {
rw1, rw2 := MsgPipe()
go func() {
- EncodeMsg(rw1, 8, []byte{0, 0})
- EncodeMsg(rw1, 5, []byte{1, 1})
+ Send(rw1, 8, [][]byte{{0, 0}})
+ Send(rw1, 5, [][]byte{{1, 1}})
rw1.Close()
}()
@@ -40,7 +24,7 @@ func ExampleMsgPipe() {
if err != nil {
break
}
- var data [1][]byte
+ var data [][]byte
msg.Decode(&data)
fmt.Printf("msg: %d, %x\n", msg.Code, data[0])
}
@@ -55,7 +39,7 @@ loop:
rw1, rw2 := MsgPipe()
done := make(chan struct{})
go func() {
- if err := EncodeMsg(rw1, 1); err == nil {
+ if err := SendItems(rw1, 1); err == nil {
t.Error("EncodeMsg returned nil error")
} else if err != ErrPipeClosed {
t.Error("EncodeMsg returned wrong error: got %v, want %v", err, ErrPipeClosed)
diff --git a/p2p/peer.go b/p2p/peer.go
index c2c83abfc..6b97ea58d 100644
--- a/p2p/peer.go
+++ b/p2p/peer.go
@@ -132,7 +132,7 @@ loop:
select {
case <-ping.C:
go func() {
- if err := EncodeMsg(p.rw, pingMsg, nil); err != nil {
+ if err := SendItems(p.rw, pingMsg); err != nil {
p.protoErr <- err
return
}
@@ -161,7 +161,7 @@ loop:
func (p *Peer) politeDisconnect(reason DiscReason) {
done := make(chan struct{})
go func() {
- EncodeMsg(p.rw, discMsg, uint(reason))
+ SendItems(p.rw, discMsg, uint(reason))
// Wait for the other side to close the connection.
// Discard any data that they send until then.
io.Copy(ioutil.Discard, p.conn)
@@ -192,12 +192,13 @@ func (p *Peer) handle(msg Msg) error {
switch {
case msg.Code == pingMsg:
msg.Discard()
- go EncodeMsg(p.rw, pongMsg)
+ go SendItems(p.rw, pongMsg)
case msg.Code == discMsg:
var reason [1]DiscReason
// no need to discard or for error checking, we'll close the
// connection after this.
rlp.Decode(msg.Payload, &reason)
+ p.Debugf("Disconnect requested: %v\n", reason[0])
p.Disconnect(DiscRequested)
return discRequestedError(reason[0])
case msg.Code < baseProtocolLength:
diff --git a/p2p/peer_test.go b/p2p/peer_test.go
index cc9f1f0cd..3c4c71c0c 100644
--- a/p2p/peer_test.go
+++ b/p2p/peer_test.go
@@ -4,13 +4,10 @@ import (
"bytes"
"fmt"
"io"
- "io/ioutil"
"net"
"reflect"
"testing"
"time"
-
- "github.com/ethereum/go-ethereum/rlp"
)
var discard = Protocol{
@@ -55,13 +52,13 @@ func TestPeerProtoReadMsg(t *testing.T) {
Name: "a",
Length: 5,
Run: func(peer *Peer, rw MsgReadWriter) error {
- if err := expectMsg(rw, 2, []uint{1}); err != nil {
+ if err := ExpectMsg(rw, 2, []uint{1}); err != nil {
t.Error(err)
}
- if err := expectMsg(rw, 3, []uint{2}); err != nil {
+ if err := ExpectMsg(rw, 3, []uint{2}); err != nil {
t.Error(err)
}
- if err := expectMsg(rw, 4, []uint{3}); err != nil {
+ if err := ExpectMsg(rw, 4, []uint{3}); err != nil {
t.Error(err)
}
close(done)
@@ -72,9 +69,9 @@ func TestPeerProtoReadMsg(t *testing.T) {
closer, rw, _, errc := testPeer([]Protocol{proto})
defer closer.Close()
- EncodeMsg(rw, baseProtocolLength+2, 1)
- EncodeMsg(rw, baseProtocolLength+3, 2)
- EncodeMsg(rw, baseProtocolLength+4, 3)
+ Send(rw, baseProtocolLength+2, []uint{1})
+ Send(rw, baseProtocolLength+3, []uint{2})
+ Send(rw, baseProtocolLength+4, []uint{3})
select {
case <-done:
@@ -92,10 +89,10 @@ func TestPeerProtoEncodeMsg(t *testing.T) {
Name: "a",
Length: 2,
Run: func(peer *Peer, rw MsgReadWriter) error {
- if err := EncodeMsg(rw, 2); err == nil {
+ if err := SendItems(rw, 2); err == nil {
t.Error("expected error for out-of-range msg code, got nil")
}
- if err := EncodeMsg(rw, 1, "foo", "bar"); err != nil {
+ if err := SendItems(rw, 1, "foo", "bar"); err != nil {
t.Errorf("write error: %v", err)
}
return nil
@@ -104,7 +101,7 @@ func TestPeerProtoEncodeMsg(t *testing.T) {
closer, rw, _, _ := testPeer([]Protocol{proto})
defer closer.Close()
- if err := expectMsg(rw, 17, []string{"foo", "bar"}); err != nil {
+ if err := ExpectMsg(rw, 17, []string{"foo", "bar"}); err != nil {
t.Error(err)
}
}
@@ -115,11 +112,15 @@ func TestPeerWriteForBroadcast(t *testing.T) {
closer, rw, peer, peerErr := testPeer([]Protocol{discard})
defer closer.Close()
+ emptymsg := func(code uint64) Msg {
+ return Msg{Code: code, Size: 0, Payload: bytes.NewReader(nil)}
+ }
+
// test write errors
- if err := peer.writeProtoMsg("b", NewMsg(3)); err == nil {
+ if err := peer.writeProtoMsg("b", emptymsg(3)); err == nil {
t.Errorf("expected error for unknown protocol, got nil")
}
- if err := peer.writeProtoMsg("discard", NewMsg(8)); err == nil {
+ if err := peer.writeProtoMsg("discard", emptymsg(8)); err == nil {
t.Errorf("expected error for out-of-range msg code, got nil")
} else if perr, ok := err.(*peerError); !ok || perr.Code != errInvalidMsgCode {
t.Errorf("wrong error for out-of-range msg code, got %#v", err)
@@ -128,14 +129,14 @@ func TestPeerWriteForBroadcast(t *testing.T) {
// setup for reading the message on the other end
read := make(chan struct{})
go func() {
- if err := expectMsg(rw, 16, nil); err != nil {
+ if err := ExpectMsg(rw, 16, nil); err != nil {
t.Error(err)
}
close(read)
}()
// test successful write
- if err := peer.writeProtoMsg("discard", NewMsg(0)); err != nil {
+ if err := peer.writeProtoMsg("discard", emptymsg(0)); err != nil {
t.Errorf("expect no error for known protocol: %v", err)
}
select {
@@ -150,10 +151,10 @@ func TestPeerPing(t *testing.T) {
closer, rw, _, _ := testPeer(nil)
defer closer.Close()
- if err := EncodeMsg(rw, pingMsg); err != nil {
+ if err := SendItems(rw, pingMsg); err != nil {
t.Fatal(err)
}
- if err := expectMsg(rw, pongMsg, nil); err != nil {
+ if err := ExpectMsg(rw, pongMsg, nil); err != nil {
t.Error(err)
}
}
@@ -163,10 +164,10 @@ func TestPeerDisconnect(t *testing.T) {
closer, rw, _, disc := testPeer(nil)
defer closer.Close()
- if err := EncodeMsg(rw, discMsg, DiscQuitting); err != nil {
+ if err := SendItems(rw, discMsg, DiscQuitting); err != nil {
t.Fatal(err)
}
- if err := expectMsg(rw, discMsg, []interface{}{DiscRequested}); err != nil {
+ if err := ExpectMsg(rw, discMsg, []interface{}{DiscRequested}); err != nil {
t.Error(err)
}
closer.Close() // make test end faster
@@ -192,35 +193,3 @@ func TestNewPeer(t *testing.T) {
p.Disconnect(DiscAlreadyConnected) // Should not hang
}
-
-// expectMsg reads a message from r and verifies that its
-// code and encoded RLP content match the provided values.
-// If content is nil, the payload is discarded and not verified.
-func expectMsg(r MsgReader, code uint64, content interface{}) error {
- msg, err := r.ReadMsg()
- if err != nil {
- return err
- }
- if msg.Code != code {
- return fmt.Errorf("message code mismatch: got %d, expected %d", msg.Code, code)
- }
- if content == nil {
- return msg.Discard()
- } else {
- contentEnc, err := rlp.EncodeToBytes(content)
- if err != nil {
- panic("content encode error: " + err.Error())
- }
- if int(msg.Size) != len(contentEnc) {
- return fmt.Errorf("message size mismatch: got %d, want %d", msg.Size, len(contentEnc))
- }
- actualContent, err := ioutil.ReadAll(msg.Payload)
- if err != nil {
- return err
- }
- if !bytes.Equal(actualContent, contentEnc) {
- return fmt.Errorf("message payload mismatch:\ngot: %x\nwant: %x", actualContent, contentEnc)
- }
- }
- return nil
-}
diff --git a/p2p/rlpx_test.go b/p2p/rlpx_test.go
index 49354c7ed..d98f1c2cd 100644
--- a/p2p/rlpx_test.go
+++ b/p2p/rlpx_test.go
@@ -30,7 +30,7 @@ ba628a4ba590cb43f7848f41c4382885
`)
// Check WriteMsg. This puts a message into the buffer.
- if err := EncodeMsg(rw, 8, 1, 2, 3, 4); err != nil {
+ if err := Send(rw, 8, []uint{1, 2, 3, 4}); err != nil {
t.Fatalf("WriteMsg error: %v", err)
}
written := buf.Bytes()
@@ -102,7 +102,7 @@ func TestRlpxFrameRW(t *testing.T) {
for i := 0; i < 10; i++ {
// write message into conn buffer
wmsg := []interface{}{"foo", "bar", strings.Repeat("test", i)}
- err := EncodeMsg(rw1, uint64(i), wmsg...)
+ err := Send(rw1, uint64(i), wmsg)
if err != nil {
t.Fatalf("WriteMsg error (i=%d): %v", i, err)
}
diff --git a/p2p/server.go b/p2p/server.go
index 9c148ba39..813b0676f 100644
--- a/p2p/server.go
+++ b/p2p/server.go
@@ -9,10 +9,10 @@ import (
"sync"
"time"
- "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/nat"
+ "github.com/ethereum/go-ethereum/rlp"
)
const (
@@ -129,10 +129,14 @@ func (srv *Server) SuggestPeer(n *discover.Node) {
// Broadcast sends an RLP-encoded message to all connected peers.
// This method is deprecated and will be removed later.
-func (srv *Server) Broadcast(protocol string, code uint64, data ...interface{}) {
+func (srv *Server) Broadcast(protocol string, code uint64, data interface{}) error {
var payload []byte
if data != nil {
- payload = common.Encode(data)
+ var err error
+ payload, err = rlp.EncodeToBytes(data)
+ if err != nil {
+ return err
+ }
}
srv.lock.RLock()
defer srv.lock.RUnlock()
@@ -146,6 +150,7 @@ func (srv *Server) Broadcast(protocol string, code uint64, data ...interface{})
peer.writeProtoMsg(protocol, msg)
}
}
+ return nil
}
// Start starts running the server.
diff --git a/p2p/server_test.go b/p2p/server_test.go
index 30447050c..14e7c7de2 100644
--- a/p2p/server_test.go
+++ b/p2p/server_test.go
@@ -149,7 +149,7 @@ func TestServerBroadcast(t *testing.T) {
connected.Wait()
// broadcast one message
- srv.Broadcast("discard", 0, "foo")
+ srv.Broadcast("discard", 0, []string{"foo"})
golden := unhex("66e94d166f0a2c3b884cfa59ca34")
// check that the message has been written everywhere
diff --git a/whisper/envelope.go b/whisper/envelope.go
index 577638046..9ec9fc318 100644
--- a/whisper/envelope.go
+++ b/whisper/envelope.go
@@ -6,9 +6,9 @@ import (
"fmt"
"time"
+ "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/crypto/ecies"
- "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/rlp"
)
diff --git a/whisper/peer.go b/whisper/peer.go
index 4bd9428f5..ee5bffd0b 100644
--- a/whisper/peer.go
+++ b/whisper/peer.go
@@ -77,12 +77,11 @@ func (self *peer) broadcast(envelopes []*Envelope) error {
}
if i > 0 {
- if err := p2p.EncodeMsg(self.ws, envelopesMsg, envs[:i]...); err != nil {
+ if err := p2p.Send(self.ws, envelopesMsg, envs[:i]); err != nil {
return err
}
self.peer.DebugDetailln("broadcasted", i, "message(s)")
}
-
return nil
}
@@ -92,7 +91,7 @@ func (self *peer) addKnown(envelope *Envelope) {
func (self *peer) handleStatus() error {
ws := self.ws
- if err := ws.WriteMsg(self.statusMsg()); err != nil {
+ if err := p2p.SendItems(ws, statusMsg, protocolVersion); err != nil {
return err
}
msg, err := ws.ReadMsg()
@@ -115,7 +114,3 @@ func (self *peer) handleStatus() error {
}
return msg.Discard() // ignore anything after protocol version
}
-
-func (self *peer) statusMsg() p2p.Msg {
- return p2p.NewMsg(statusMsg, protocolVersion)
-}
generated by cgit v1.2.3 (git 2.25.1) at 2025-04-26 02:22:14 +0800