diff --git a/node/api.go b/node/api.go index b9d78d1e29..1b04b70938 100644 --- a/node/api.go +++ b/node/api.go @@ -17,6 +17,7 @@ package node import ( + "context" "fmt" "strings" "time" @@ -25,6 +26,7 @@ import ( "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/rpc" "github.com/rcrowley/go-metrics" ) @@ -73,6 +75,44 @@ func (api *PrivateAdminAPI) RemovePeer(url string) (bool, error) { return true, nil } +// PeerEvents creates an RPC subscription which receives peer events from the +// node's p2p.Server +func (api *PrivateAdminAPI) PeerEvents(ctx context.Context) (*rpc.Subscription, error) { + // Make sure the server is running, fail otherwise + server := api.node.Server() + if server == nil { + return nil, ErrNodeStopped + } + + // Create the subscription + notifier, supported := rpc.NotifierFromContext(ctx) + if !supported { + return nil, rpc.ErrNotificationsUnsupported + } + rpcSub := notifier.CreateSubscription() + + go func() { + events := make(chan *p2p.PeerEvent) + sub := server.SubscribeEvents(events) + defer sub.Unsubscribe() + + for { + select { + case event := <-events: + notifier.Notify(rpcSub.ID, event) + case <-sub.Err(): + return + case <-rpcSub.Err(): + return + case <-notifier.Closed(): + return + } + } + }() + + return rpcSub, nil +} + // StartRPC starts the HTTP RPC API server. func (api *PrivateAdminAPI) StartRPC(host *string, port *int, cors *string, apis *string) (bool, error) { api.node.lock.Lock() diff --git a/p2p/message.go b/p2p/message.go index 1292d21213..67c831a132 100644 --- a/p2p/message.go +++ b/p2p/message.go @@ -27,6 +27,8 @@ import ( "sync/atomic" "time" + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/rlp" ) @@ -271,3 +273,67 @@ func ExpectMsg(r MsgReader, code uint64, content interface{}) error { } return nil } + +// MsgEventer wraps a MsgReadWriter and sends events whenever a message is sent +// or received +type MsgEventer struct { + MsgReadWriter + + feed *event.Feed + peerID discover.NodeID + Protocol string +} + +// NewMsgEventer returns a MsgEventer which sends message events to the given +// feed +func NewMsgEventer(rw MsgReadWriter, feed *event.Feed, peerID discover.NodeID, proto string) *MsgEventer { + return &MsgEventer{ + MsgReadWriter: rw, + feed: feed, + peerID: peerID, + Protocol: proto, + } +} + +// ReadMsg reads a message from the underlying MsgReadWriter and emits a +// "message received" event +func (self *MsgEventer) ReadMsg() (Msg, error) { + msg, err := self.MsgReadWriter.ReadMsg() + if err != nil { + return msg, err + } + self.feed.Send(&PeerEvent{ + Type: PeerEventTypeMsgRecv, + Peer: self.peerID, + Protocol: self.Protocol, + MsgCode: &msg.Code, + MsgSize: &msg.Size, + }) + return msg, nil +} + +// WriteMsg writes a message to the underlying MsgReadWriter and emits a +// "message sent" event +func (self *MsgEventer) WriteMsg(msg Msg) error { + err := self.MsgReadWriter.WriteMsg(msg) + if err != nil { + return err + } + self.feed.Send(&PeerEvent{ + Type: PeerEventTypeMsgSend, + Peer: self.peerID, + Protocol: self.Protocol, + MsgCode: &msg.Code, + MsgSize: &msg.Size, + }) + return nil +} + +// Close closes the underlying MsgReadWriter if it implements the io.Closer +// interface +func (self *MsgEventer) Close() error { + if v, ok := self.MsgReadWriter.(io.Closer); ok { + return v.Close() + } + return nil +} diff --git a/p2p/peer.go b/p2p/peer.go index fb4b39e950..40694b2873 100644 --- a/p2p/peer.go +++ b/p2p/peer.go @@ -25,6 +25,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/rlp" @@ -60,6 +61,38 @@ type protoHandshake struct { Rest []rlp.RawValue `rlp:"tail"` } +// PeerEventType is the type of peer events emitted by a p2p.Server +type PeerEventType string + +const ( + // PeerEventTypeAdd is the type of event emitted when a peer is added + // to a p2p.Server + PeerEventTypeAdd PeerEventType = "add" + + // PeerEventTypeDrop is the type of event emitted when a peer is + // dropped from a p2p.Server + PeerEventTypeDrop PeerEventType = "drop" + + // PeerEventTypeMsgSend is the type of event emitted when a + // message is successfully sent to a peer + PeerEventTypeMsgSend PeerEventType = "msgsend" + + // PeerEventTypeMsgRecv is the type of event emitted when a + // message is received from a peer + PeerEventTypeMsgRecv PeerEventType = "msgrecv" +) + +// 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 +type PeerEvent struct { + Type PeerEventType `json:"type"` + Peer discover.NodeID `json:"peer"` + Error string `json:"error,omitempty"` + Protocol string `json:"protocol,omitempty"` + MsgCode *uint64 `json:"msg_code,omitempty"` + MsgSize *uint32 `json:"msg_size,omitempty"` +} + // Peer represents a connected remote node. type Peer struct { rw *conn @@ -71,6 +104,9 @@ type Peer struct { protoErr chan error closed chan struct{} disc chan DiscReason + + // events receives message send / receive events if set + events *event.Feed } // NewPeer returns a peer for testing purposes. @@ -297,9 +333,13 @@ func (p *Peer) startProtocols(writeStart <-chan struct{}, writeErr chan<- error) proto.closed = p.closed proto.wstart = writeStart proto.werr = writeErr + var rw MsgReadWriter = proto + if p.events != nil { + rw = NewMsgEventer(rw, p.events, p.ID(), proto.Name) + } p.log.Trace(fmt.Sprintf("Starting protocol %s/%d", proto.Name, proto.Version)) go func() { - err := proto.Run(p, proto) + err := proto.Run(p, rw) if err == nil { p.log.Trace(fmt.Sprintf("Protocol %s/%d returned", proto.Name, proto.Version)) err = errProtocolReturned diff --git a/p2p/server.go b/p2p/server.go index d7909d53a9..770577a1ad 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -27,6 +27,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/discv5" @@ -134,6 +135,10 @@ type Config struct { // If NoDial is true, the server will not dial any peers. NoDial bool `toml:",omitempty"` + + // If EnableMsgEvents is set then the server will emit PeerEvents + // whenever a message is sent to or received from a peer + EnableMsgEvents bool } // Server manages all peer connections. @@ -166,6 +171,7 @@ type Server struct { addpeer chan *conn delpeer chan peerDrop loopWG sync.WaitGroup // loop, listenLoop + peerFeed event.Feed } type peerOpFunc func(map[discover.NodeID]*Peer) @@ -291,6 +297,11 @@ func (srv *Server) RemovePeer(node *discover.Node) { } } +// SubscribePeers subscribes the given channel to peer events +func (srv *Server) SubscribeEvents(ch chan *PeerEvent) event.Subscription { + return srv.peerFeed.Subscribe(ch) +} + // Self returns the local node's endpoint information. func (srv *Server) Self() *discover.Node { srv.lock.Lock() @@ -544,6 +555,11 @@ running: if err == nil { // The handshakes are done and it passed all checks. p := newPeer(c, srv.Protocols) + // If message events are enabled, pass the peerFeed + // to the peer + if srv.EnableMsgEvents { + p.events = &srv.peerFeed + } name := truncateName(c.name) log.Debug("Adding p2p peer", "id", c.id, "name", name, "addr", c.fd.RemoteAddr(), "peers", len(peers)+1) peers[c.id] = p @@ -755,7 +771,23 @@ func (srv *Server) runPeer(p *Peer) { if srv.newPeerHook != nil { srv.newPeerHook(p) } + + // broadcast peer add + srv.peerFeed.Send(&PeerEvent{ + Type: PeerEventTypeAdd, + Peer: p.ID(), + }) + + // run the protocol remoteRequested, err := p.run() + + // broadcast peer drop + srv.peerFeed.Send(&PeerEvent{ + Type: PeerEventTypeDrop, + Peer: p.ID(), + Error: err.Error(), + }) + // Note: run waits for existing peers to be sent on srv.delpeer // before returning, so this send should not select on srv.quit. srv.delpeer <- peerDrop{p, err, remoteRequested}