From cdab6654759a959ec78db05a2f6a04b5015b5570 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Sat, 6 May 2017 15:38:49 +0100 Subject: [PATCH] p2p/simulations: Move PeerEvents to admin RPC namespace Signed-off-by: Lewis Marshall --- node/api.go | 40 ++++++++++ p2p/simulations/adapters/exec.go | 77 +------------------ p2p/simulations/adapters/inproc.go | 40 ++++++++-- p2p/simulations/network.go | 2 +- .../simulations/discovery/discovery_test.go | 2 +- 5 files changed, 79 insertions(+), 82 deletions(-) diff --git a/node/api.go b/node/api.go index 70aa59d844..ba8cc1b8b7 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/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go index 13715cc100..fd31e5ef70 100644 --- a/p2p/simulations/adapters/exec.go +++ b/p2p/simulations/adapters/exec.go @@ -315,22 +315,10 @@ func startP2PNode(conf *node.Config, service node.Service) (*node.Node, error) { if err != nil { return nil, err } - - constructor := func(s node.Service) node.ServiceConstructor { - return func(ctx *node.ServiceContext) (node.Service, error) { - return s, nil - } + constructor := func(ctx *node.ServiceContext) (node.Service, error) { + return service, nil } - - // register the peer events API - // - // TODO: move this to node.PrivateAdminAPI once the following is merged: - // https://github.com/ethereum/go-ethereum/pull/13885 - if err := stack.Register(constructor(&PeerAPI{stack.Server})); err != nil { - return nil, err - } - - if err := stack.Register(constructor(service)); err != nil { + if err := stack.Register(constructor); err != nil { return nil, err } if err := stack.Start(); err != nil { @@ -339,65 +327,6 @@ func startP2PNode(conf *node.Config, service node.Service) (*node.Node, error) { return stack, nil } -// PeerAPI is used to expose peer events under the "eth" RPC namespace. -// -// TODO: move this to node.PrivateAdminAPI and expose under the "admin" -// namespace once the following is merged: -// https://github.com/ethereum/go-ethereum/pull/13885 -type PeerAPI struct { - server func() p2p.Server -} - -func (p *PeerAPI) Protocols() []p2p.Protocol { - return nil -} - -func (p *PeerAPI) APIs() []rpc.API { - return []rpc.API{{ - Namespace: "eth", - Version: "1.0", - Service: p, - }} -} - -func (p *PeerAPI) Start(p2p.Server) error { - return nil -} - -func (p *PeerAPI) Stop() error { - return nil -} - -// PeerEvents creates an RPC sunscription which receives peer events from the -// underlying p2p.Server -func (p *PeerAPI) PeerEvents(ctx context.Context) (*rpc.Subscription, error) { - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - - rpcSub := notifier.CreateSubscription() - - go func() { - events := make(chan *p2p.PeerEvent) - sub := p.server().SubscribeEvents(events) - defer sub.Unsubscribe() - - for { - select { - case event := <-events: - notifier.Notify(rpcSub.ID, event) - case <-rpcSub.Err(): - return - case <-notifier.Closed(): - return - } - } - }() - - return rpcSub, nil -} - // stdioConn wraps os.Stdin / os.Stdout with a no-op Close method so we can // use stdio for RPC messages type stdioConn struct { diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index 559e3ea234..fbabe9c5f7 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -17,6 +17,7 @@ package adapters import ( + "context" "errors" "fmt" "sync" @@ -184,7 +185,7 @@ func (self *SimNode) startRPC() error { return errors.New("RPC already started") } - // add SimAdminAPI and PeerAPI so that the network can call the + // add SimAdminAPI so that the network can call the // AddPeer, RemovePeer and PeerEvents RPC methods apis := append(self.service.APIs(), []rpc.API{ { @@ -192,11 +193,6 @@ func (self *SimNode) startRPC() error { Version: "1.0", Service: &SimAdminAPI{self}, }, - { - Namespace: "eth", - Version: "1.0", - Service: &PeerAPI{func() p2p.Server { return self }}, - }, }...) // start the RPC handler @@ -356,3 +352,35 @@ func (api *SimAdminAPI) RemovePeer(url string) (bool, error) { api.SimNode.RemovePeer(node) return true, nil } + +// PeerEvents creates an RPC subscription which receives peer events from the +// underlying p2p.Server +func (api *SimAdminAPI) PeerEvents(ctx context.Context) (*rpc.Subscription, error) { + notifier, supported := rpc.NotifierFromContext(ctx) + if !supported { + return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported + } + + rpcSub := notifier.CreateSubscription() + + go func() { + events := make(chan *p2p.PeerEvent) + sub := api.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 +} diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index 95a0198ee5..e24c9b5cef 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -426,7 +426,7 @@ func (self *Network) Start(id *adapters.NodeId) error { return fmt.Errorf("error getting rpc client for node %v: %s", id, err) } events := make(chan *p2p.PeerEvent) - sub, err := client.EthSubscribe(context.Background(), events, "peerEvents") + sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents") if err != nil { return fmt.Errorf("error getting peer events for node %v: %s", id, err) } diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index f905cb8ac5..5a5ec53f37 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -159,7 +159,7 @@ func triggerChecks(trigger chan *adapters.NodeId, net *simulations.Network, id * return err } events := make(chan *p2p.PeerEvent) - sub, err := client.EthSubscribe(context.Background(), events, "peerEvents") + sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents") if err != nil { return fmt.Errorf("error getting peer events for node %v: %s", id, err) }