roysc/plugeth
1
0
Fork 0
forked from cerc-io/plugeth
Code Pull Requests Activity

p2p: add address info to peer event reporting (#19716)

Browse Source
This commit is contained in:
Martin Holst Swende 2019-07-05 20:27:13 +02:00 committed by Felix Lange
parent dcc4adfcd7
commit 7fd82a0e3e
3 changed files with 40 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
p2p/message.go
View File

@ -252,19 +252,23 @@ func ExpectMsg(r MsgReader, code uint64, content interface{}) error {
type msgEventer struct { type msgEventer struct {
MsgReadWriter MsgReadWriter
feed *event.Feed feed *event.Feed
peerID enode.ID peerID enode.ID
Protocol string Protocol string
localAddress string
remoteAddress string
} }
// newMsgEventer returns a msgEventer which sends message events to the given // newMsgEventer returns a msgEventer which sends message events to the given
// feed // feed
func newMsgEventer(rw MsgReadWriter, feed *event.Feed, peerID enode.ID, proto string) *msgEventer { func newMsgEventer(rw MsgReadWriter, feed *event.Feed, peerID enode.ID, proto, remote, local string) *msgEventer {
return &msgEventer{ return &msgEventer{
MsgReadWriter: rw, MsgReadWriter: rw,
feed: feed, feed: feed,
peerID: peerID, peerID: peerID,
Protocol: proto, Protocol: proto,
remoteAddress: remote,
localAddress: local,
} }
} }
@ -276,11 +280,13 @@ func (ev *msgEventer) ReadMsg() (Msg, error) {
return msg, err return msg, err
} }
ev.feed.Send(&PeerEvent{ ev.feed.Send(&PeerEvent{
Type: PeerEventTypeMsgRecv, Type: PeerEventTypeMsgRecv,
Peer: ev.peerID, Peer: ev.peerID,
Protocol: ev.Protocol, Protocol: ev.Protocol,
MsgCode: &msg.Code, MsgCode: &msg.Code,
MsgSize: &msg.Size, MsgSize: &msg.Size,
LocalAddress: ev.localAddress,
RemoteAddress: ev.remoteAddress,
}) })
return msg, nil return msg, nil
} }
@ -293,11 +299,13 @@ func (ev *msgEventer) WriteMsg(msg Msg) error {
return err return err
} }
ev.feed.Send(&PeerEvent{ ev.feed.Send(&PeerEvent{
Type: PeerEventTypeMsgSend, Type: PeerEventTypeMsgSend,
Peer: ev.peerID, Peer: ev.peerID,
Protocol: ev.Protocol, Protocol: ev.Protocol,
MsgCode: &msg.Code, MsgCode: &msg.Code,
MsgSize: &msg.Size, MsgSize: &msg.Size,
LocalAddress: ev.localAddress,
RemoteAddress: ev.remoteAddress,
}) })
return nil return nil
} }

16
p2p/peer.go
View File

@ -91,12 +91,14 @@ const (
// PeerEvent is an event emitted when peers are either added or dropped from // PeerEvent is an event emitted when peers are either added or dropped from
// a p2p.Server or when a message is sent or received on a peer connection // a p2p.Server or when a message is sent or received on a peer connection
type PeerEvent struct { type PeerEvent struct {
Type PeerEventType `json:"type"` Type PeerEventType `json:"type"`
Peer enode.ID `json:"peer"` Peer enode.ID `json:"peer"`
Error string `json:"error,omitempty"` Error string `json:"error,omitempty"`
Protocol string `json:"protocol,omitempty"` Protocol string `json:"protocol,omitempty"`
MsgCode *uint64 `json:"msg_code,omitempty"` MsgCode *uint64 `json:"msg_code,omitempty"`
MsgSize *uint32 `json:"msg_size,omitempty"` MsgSize *uint32 `json:"msg_size,omitempty"`
LocalAddress string `json:"local,omitempty"`
RemoteAddress string `json:"remote,omitempty"`
} }
// Peer represents a connected remote node. // Peer represents a connected remote node.
@ -354,7 +356,7 @@ func (p *Peer) startProtocols(writeStart <-chan struct{}, writeErr chan<- error)
proto.werr = writeErr proto.werr = writeErr
var rw MsgReadWriter = proto var rw MsgReadWriter = proto
if p.events != nil { if p.events != nil {
rw = newMsgEventer(rw, p.events, p.ID(), proto.Name) rw = newMsgEventer(rw, p.events, p.ID(), proto.Name, p.Info().Network.RemoteAddress, p.Info().Network.LocalAddress)
} }
p.log.Trace(fmt.Sprintf("Starting protocol %s/%d", proto.Name, proto.Version)) p.log.Trace(fmt.Sprintf("Starting protocol %s/%d", proto.Name, proto.Version))
go func() { go func() {

14
p2p/server.go
View File

@ -1016,8 +1016,10 @@ func (srv *Server) runPeer(p *Peer) {
// broadcast peer add // broadcast peer add
srv.peerFeed.Send(&PeerEvent{ srv.peerFeed.Send(&PeerEvent{
Type: PeerEventTypeAdd, Type: PeerEventTypeAdd,
Peer: p.ID(), Peer: p.ID(),
RemoteAddress: p.RemoteAddr().String(),
LocalAddress: p.LocalAddr().String(),
}) })
// run the protocol // run the protocol
@ -1025,9 +1027,11 @@ func (srv *Server) runPeer(p *Peer) {
// broadcast peer drop // broadcast peer drop
srv.peerFeed.Send(&PeerEvent{ srv.peerFeed.Send(&PeerEvent{
Type: PeerEventTypeDrop, Type: PeerEventTypeDrop,
Peer: p.ID(), Peer: p.ID(),
Error: err.Error(), Error: err.Error(),
RemoteAddress: p.RemoteAddr().String(),
LocalAddress: p.LocalAddr().String(),
}) })
// Note: run waits for existing peers to be sent on srv.delpeer // Note: run waits for existing peers to be sent on srv.delpeer