forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-27 15:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

p2p/simulations: Drop adapters.NodeId in favour of discover.NodeID

Browse source 
Signed-off-by: Lewis Marshall <lewis@lmars.net>
This commit is contained in:
Lewis Marshall 2017-05-27 16:11:03 +01:00
parent bdf4ba039c
commit da6d15da4c
28 changed files with 406 additions and 427 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
p2p/discover/node.go
View file

@ -225,6 +225,11 @@ func (n *Node) UnmarshalText(text []byte) error {
// The node identifier is a marshaled elliptic curve public key.
type NodeID [NodeIDBits / 8]byte
// Bytes returns a byte slice representation of the NodeID
func (n NodeID) Bytes() []byte {
return n[:]
}
// NodeID prints as a long hexadecimal number.
func (n NodeID) String() string {
return fmt.Sprintf("%x", n[:])
@ -240,6 +245,26 @@ func (n NodeID) TerminalString() string {
return hex.EncodeToString(n[:8])
}
// BytesID converts a byte slice to a NodeID
func BytesID(b []byte) (NodeID, error) {
var id NodeID
if len(b) != len(id) {
return id, fmt.Errorf("wrong length, want %d bytes", len(id))
}
copy(id[:], b)
return id, nil
}
// MustBytesID converts a byte slice to a NodeID.
// It panics if the byte slice is not a valid NodeID.
func MustBytesID(b []byte) NodeID {
id, err := BytesID(b)
if err != nil {
panic(err)
}
return id
}
// HexID converts a hex string to a NodeID.
// The string may be prefixed with 0x.
func HexID(in string) (NodeID, error) {

47
p2p/protocols/protocol_test.go
View file

@ -10,6 +10,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
)
@ -23,9 +24,9 @@ type hs0 struct {
C uint
}
// message to kill/drop the peer with nodeId
// message to kill/drop the peer with nodeID
type kill struct {
C *adapters.NodeId
C discover.NodeID
}
// message to drop connection
@ -139,10 +140,10 @@ func newProtocol(pp *p2ptest.TestPeerPool) func(*p2p.Peer, p2p.MsgReadWriter) er
func protocolTester(t *testing.T, pp *p2ptest.TestPeerPool) *p2ptest.ProtocolTester {
conf := adapters.RandomNodeConfig()
return p2ptest.NewProtocolTester(t, conf.Id, 2, newProtocol(pp))
return p2ptest.NewProtocolTester(t, conf.ID, 2, newProtocol(pp))
}
func protoHandshakeExchange(id *adapters.NodeId, proto *protoHandshake) []p2ptest.Exchange {
func protoHandshakeExchange(id discover.NodeID, proto *protoHandshake) []p2ptest.Exchange {
return []p2ptest.Exchange{
p2ptest.Exchange{
@ -170,13 +171,13 @@ func runProtoHandshake(t *testing.T, proto *protoHandshake, errs ...error) {
pp := p2ptest.NewTestPeerPool()
s := protocolTester(t, pp)
// TODO: make this more than one handshake
id := s.Ids[0]
id := s.IDs[0]
if err := s.TestExchanges(protoHandshakeExchange(id, proto)...); err != nil {
t.Fatal(err)
}
var disconnects []*p2ptest.Disconnect
for i, err := range errs {
disconnects = append(disconnects, &p2ptest.Disconnect{Peer: s.Ids[i], Error: err})
disconnects = append(disconnects, &p2ptest.Disconnect{Peer: s.IDs[i], Error: err})
}
if err := s.TestDisconnected(disconnects...); err != nil {
t.Fatal(err)
@ -195,7 +196,7 @@ func TestProtoHandshakeSuccess(t *testing.T) {
runProtoHandshake(t, &protoHandshake{42, "420"})
}
func moduleHandshakeExchange(id *adapters.NodeId, resp uint) []p2ptest.Exchange {
func moduleHandshakeExchange(id discover.NodeID, resp uint) []p2ptest.Exchange {
return []p2ptest.Exchange{
p2ptest.Exchange{
@ -222,12 +223,12 @@ func moduleHandshakeExchange(id *adapters.NodeId, resp uint) []p2ptest.Exchange
func runModuleHandshake(t *testing.T, resp uint, errs ...error) {
pp := p2ptest.NewTestPeerPool()
s := protocolTester(t, pp)
id := s.Ids[0]
id := s.IDs[0]
s.TestExchanges(protoHandshakeExchange(id, &protoHandshake{42, "420"})...)
s.TestExchanges(moduleHandshakeExchange(id, resp)...)
var disconnects []*p2ptest.Disconnect
for i, err := range errs {
disconnects = append(disconnects, &p2ptest.Disconnect{Peer: s.Ids[i], Error: err})
disconnects = append(disconnects, &p2ptest.Disconnect{Peer: s.IDs[i], Error: err})
}
s.TestDisconnected(disconnects...)
}
@ -241,7 +242,7 @@ func TestModuleHandshakeSuccess(t *testing.T) {
}
// testing complex interactions over multiple peers, relaying, dropping
func testMultiPeerSetup(a, b *adapters.NodeId) []p2ptest.Exchange {
func testMultiPeerSetup(a, b discover.NodeID) []p2ptest.Exchange {
return []p2ptest.Exchange{
p2ptest.Exchange{
@ -297,20 +298,20 @@ func runMultiplePeers(t *testing.T, peer int, errs ...error) {
pp := p2ptest.NewTestPeerPool()
s := protocolTester(t, pp)
s.TestExchanges(testMultiPeerSetup(s.Ids[0], s.Ids[1])...)
s.TestExchanges(testMultiPeerSetup(s.IDs[0], s.IDs[1])...)
// after some exchanges of messages, we can test state changes
// here this is simply demonstrated by the peerPool
// after the handshake negotiations peers must be added to the pool
// time.Sleep(1)
for !pp.Has(s.Ids[0]) {
for !pp.Has(s.IDs[0]) {
time.Sleep(1)
log.Trace(fmt.Sprintf("missing peer test-0: %v (%v)", pp, s.Ids))
log.Trace(fmt.Sprintf("missing peer test-0: %v (%v)", pp, s.IDs))
}
// if !pp.Has(s.Ids[0]) {
// t.Fatalf("missing peer test-0: %v (%v)", pp, s.Ids)
// if !pp.Has(s.IDs[0]) {
// t.Fatalf("missing peer test-0: %v (%v)", pp, s.IDs)
// }
if !pp.Has(s.Ids[1]) {
t.Fatalf("missing peer test-1: %v (%v)", pp, s.Ids)
if !pp.Has(s.IDs[1]) {
t.Fatalf("missing peer test-1: %v (%v)", pp, s.IDs)
}
// sending kill request for peer with index <peer>
@ -318,8 +319,8 @@ func runMultiplePeers(t *testing.T, peer int, errs ...error) {
Triggers: []p2ptest.Trigger{
p2ptest.Trigger{
Code: 2,
Msg: &kill{s.Ids[peer]},
Peer: s.Ids[0],
Msg: &kill{s.IDs[peer]},
Peer: s.IDs[0],
},
},
})
@ -330,19 +331,19 @@ func runMultiplePeers(t *testing.T, peer int, errs ...error) {
p2ptest.Trigger{
Code: 3,
Msg: &drop{},
Peer: s.Ids[(peer+1)%2],
Peer: s.IDs[(peer+1)%2],
},
},
})
// check the actual discconnect errors on the individual peers
var disconnects []*p2ptest.Disconnect
for i, err := range errs {
disconnects = append(disconnects, &p2ptest.Disconnect{Peer: s.Ids[i], Error: err})
disconnects = append(disconnects, &p2ptest.Disconnect{Peer: s.IDs[i], Error: err})
}
s.TestDisconnected(disconnects...)
// test if disconnected peers have been removed from peerPool
if pp.Has(s.Ids[peer]) {
t.Fatalf("peer test-%v not dropped: %v (%v)", peer, pp, s.Ids)
if pp.Has(s.IDs[peer]) {
t.Fatalf("peer test-%v not dropped: %v (%v)", peer, pp, s.IDs)
}
}

2
p2p/simulations/adapters/docker.go
View file

@ -64,7 +64,7 @@ func (d *DockerAdapter) NewNode(config *NodeConfig) (Node, error) {
node := &DockerNode{
ExecNode: ExecNode{
ID: config.Id,
ID: config.ID,
Config: conf,
},
}

9
p2p/simulations/adapters/exec.go
View file

@ -20,6 +20,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/rpc"
)
@ -57,7 +58,7 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
// create the node directory using the first 12 characters of the ID
// as Unix socket paths cannot be longer than 256 characters
dir := filepath.Join(e.BaseDir, config.Id.String()[:12])
dir := filepath.Join(e.BaseDir, config.ID.String()[:12])
if err := os.Mkdir(dir, 0755); err != nil {
return nil, fmt.Errorf("error creating node directory: %s", err)
}
@ -77,7 +78,7 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
conf.Stack.P2P.ListenAddr = "127.0.0.1:0"
node := &ExecNode{
ID: config.Id,
ID: config.ID,
Dir: dir,
Config: conf,
}
@ -93,7 +94,7 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
// (so for example we can run the node in a remote Docker container and
// still communicate with it).
type ExecNode struct {
ID *NodeId
ID discover.NodeID
Dir string
Config *execNodeConfig
Cmd *exec.Cmd
@ -265,7 +266,7 @@ func execP2PNode() {
// read the services and ID from argv
serviceNames := strings.Split(os.Args[1], ",")
id := NewNodeIdFromHex(os.Args[2])
id := discover.MustHexID(os.Args[2])
// decode the config
confEnv := os.Getenv("_P2P_NODE_CONFIG")

16
p2p/simulations/adapters/inproc.go
View file

@ -59,8 +59,8 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
defer s.mtx.Unlock()
// check a node with the ID doesn't already exist
id := config.Id
if _, exists := s.nodes[id.NodeID]; exists {
id := config.ID
if _, exists := s.nodes[id]; exists {
return nil, fmt.Errorf("node already exists: %s", id)
}
@ -75,11 +75,11 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
}
node := &SimNode{
Id: id,
ID: id,
config: config,
adapter: s,
}
s.nodes[id.NodeID] = node
s.nodes[id] = node
return node, nil
}
@ -113,7 +113,7 @@ func (s *SimAdapter) GetNode(id discover.NodeID) (*SimNode, bool) {
// by the underlying service.
type SimNode struct {
lock sync.RWMutex
Id *NodeId
ID discover.NodeID
config *NodeConfig
adapter *SimAdapter
node *node.Node
@ -129,7 +129,7 @@ func (self *SimNode) Addr() []byte {
// Node returns a discover.Node representing the SimNode
func (self *SimNode) Node() *discover.Node {
return discover.NewNode(self.Id.NodeID, net.IP{127, 0, 0, 1}, 30303, 30303)
return discover.NewNode(self.ID, net.IP{127, 0, 0, 1}, 30303, 30303)
}
// Client returns an rpc.Client which can be used to communicate with the
@ -183,7 +183,7 @@ func (self *SimNode) Start(snapshots map[string][]byte) error {
snapshot = snapshots[name]
}
serviceFunc := self.adapter.services[name]
service := serviceFunc(self.Id, snapshot)
service := serviceFunc(self.ID, snapshot)
self.running = append(self.running, service)
return service, nil
}
@ -272,7 +272,7 @@ func (self *SimNode) NodeInfo() *p2p.NodeInfo {
server := self.Server()
if server == nil {
return &p2p.NodeInfo{
ID: self.Id.String(),
ID: self.ID.String(),
Enode: self.Node().String(),
}
}

10
p2p/simulations/adapters/rpc_mux.go
View file

@ -44,7 +44,7 @@ type rpcMux struct {
type rpcMsg struct {
Method string `json:"method,omitempty"`
Version string `json:"jsonrpc,omitempty"`
Id json.RawMessage `json:"id,omitempty"`
ID json.RawMessage `json:"id,omitempty"`
Payload json.RawMessage `json:"params,omitempty"`
Result json.RawMessage `json:"result,omitempty"`
Error json.RawMessage `json:"error,omitempty"`
@ -174,7 +174,7 @@ func (mux *rpcMux) newMsg(msg *rpcMsg) *rpcMsg {
mux.idCounter++
mux.msgMap[id] = msg
newMsg := *msg
newMsg.Id = json.RawMessage(strconv.FormatUint(id, 10))
newMsg.ID = json.RawMessage(strconv.FormatUint(id, 10))
return &newMsg
}
@ -185,7 +185,7 @@ func (mux *rpcMux) lookup(msg *rpcMsg) *rpcReply {
// if the message has no ID, it is a subscription notification so
// lookup the original subscribe message
if msg.Id == nil {
if msg.ID == nil {
sub := &rpcSub{}
if err := json.Unmarshal(msg.Payload, sub); err != nil {
return nil
@ -194,7 +194,7 @@ func (mux *rpcMux) lookup(msg *rpcMsg) *rpcReply {
}
// lookup the original message and restore the ID
id, err := strconv.ParseUint(string(msg.Id), 10, 64)
id, err := strconv.ParseUint(string(msg.ID), 10, 64)
if err != nil {
return nil
}
@ -203,7 +203,7 @@ func (mux *rpcMux) lookup(msg *rpcMsg) *rpcReply {
return nil
}
delete(mux.msgMap, id)
msg.Id = origMsg.Id
msg.ID = origMsg.ID
// if the original message was a subscription, store the subscription
// ID so we can detect notifications

59
p2p/simulations/adapters/types.go
View file

@ -77,51 +77,10 @@ type NodeAdapter interface {
// RunProtocol is a function which runs a p2p protocol (see p2p.Protocol.Run)
type RunProtocol func(*p2p.Peer, p2p.MsgReadWriter) error
// NodeId wraps a discover.NodeID with some convenience methods
type NodeId struct {
discover.NodeID
}
func NewNodeId(id []byte) *NodeId {
var n discover.NodeID
copy(n[:], id)
return &NodeId{n}
}
func NewNodeIdFromHex(s string) *NodeId {
id := discover.MustHexID(s)
return &NodeId{id}
}
func (self *NodeId) Bytes() []byte {
return self.NodeID[:]
}
func (self *NodeId) Label() string {
return self.String()[:4]
}
func (self *NodeId) MarshalJSON() ([]byte, error) {
return json.Marshal(hex.EncodeToString(self.NodeID[:]))
}
func (self *NodeId) UnmarshalJSON(data []byte) error {
var s string
if err := json.Unmarshal(data, &s); err != nil {
return err
}
id, err := discover.HexID(s)
if err != nil {
return err
}
self.NodeID = id
return nil
}
// NodeConfig is the configuration used to start a node in a simulation
// network
type NodeConfig struct {
Id *NodeId
ID discover.NodeID
PrivateKey *ecdsa.PrivateKey
// Name is a human friendly name for the node like "node01"
@ -137,7 +96,7 @@ type NodeConfig struct {
// nodeConfigJSON is used to encode and decode NodeConfig as JSON by converting
// all fields to strings
type nodeConfigJSON struct {
Id string `json:"id"`
ID string `json:"id"`
PrivateKey string `json:"private_key"`
Name string `json:"name"`
Services []string `json:"services"`
@ -145,12 +104,10 @@ type nodeConfigJSON struct {
func (n *NodeConfig) MarshalJSON() ([]byte, error) {
confJSON := nodeConfigJSON{
ID: n.ID.String(),
Name: n.Name,
Services: n.Services,
}
if n.Id != nil {
confJSON.Id = n.Id.String()
}
if n.PrivateKey != nil {
confJSON.PrivateKey = hex.EncodeToString(crypto.FromECDSA(n.PrivateKey))
}
@ -163,12 +120,12 @@ func (n *NodeConfig) UnmarshalJSON(data []byte) error {
return err
}
if confJSON.Id != "" {
nodeID, err := discover.HexID(confJSON.Id)
if confJSON.ID != "" {
nodeID, err := discover.HexID(confJSON.ID)
if err != nil {
return err
}
n.Id = &NodeId{NodeID: nodeID}
n.ID = nodeID
}
if confJSON.PrivateKey != "" {
@ -196,7 +153,7 @@ func RandomNodeConfig() *NodeConfig {
pubkey := crypto.FromECDSAPub(&key.PublicKey)
copy(id[:], pubkey[1:])
return &NodeConfig{
Id: &NodeId{NodeID: id},
ID: id,
PrivateKey: key,
}
}
@ -205,7 +162,7 @@ func RandomNodeConfig() *NodeConfig {
type Services map[string]ServiceFunc
// ServiceFunc returns a node.Service which can be used to boot devp2p nodes
type ServiceFunc func(id *NodeId, snapshot []byte) node.Service
type ServiceFunc func(id discover.NodeID, snapshot []byte) node.Service
// serviceFuncs is a map of registered services which are used to boot devp2p
// nodes

6
p2p/simulations/cmd/p2psim/main.go
View file

@ -189,7 +189,7 @@ func listNetworks(ctx *cli.Context) error {
defer w.Flush()
fmt.Fprintf(w, "ID\tNODES\tCONNS\n")
for _, network := range networks {
fmt.Fprintf(w, "%s\t%d\t%d\n", network.Id, len(network.Nodes), len(network.Conns))
fmt.Fprintf(w, "%s\t%d\t%d\n", network.ID, len(network.Nodes), len(network.Conns))
}
return nil
}
@ -206,7 +206,7 @@ func createNetwork(ctx *cli.Context) error {
if err != nil {
return err
}
fmt.Fprintln(ctx.App.Writer, "Created network", network.Id)
fmt.Fprintln(ctx.App.Writer, "Created network", network.ID)
return nil
}
@ -222,7 +222,7 @@ func showNetwork(ctx *cli.Context) error {
}
w := tabwriter.NewWriter(ctx.App.Writer, 1, 2, 2, ' ', 0)
defer w.Flush()
fmt.Fprintf(w, "ID\t%s\n", network.Id)
fmt.Fprintf(w, "ID\t%s\n", network.ID)
fmt.Fprintf(w, "NODES\t%d\n", len(network.Nodes))
fmt.Fprintf(w, "CONNS\t%d\n", len(network.Conns))
return nil

6
p2p/simulations/events.go
View file

@ -81,11 +81,11 @@ func ControlEvent(v interface{}) *Event {
func (e *Event) String() string {
switch e.Type {
case EventTypeNode:
return fmt.Sprintf("<node-event> id: %s up: %t", e.Node.ID().Label(), e.Node.Up)
return fmt.Sprintf("<node-event> id: %s up: %t", e.Node.ID().TerminalString(), e.Node.Up)
case EventTypeConn:
return fmt.Sprintf("<conn-event> nodes: %s->%s up: %t", e.Conn.One.Label(), e.Conn.Other.Label(), e.Conn.Up)
return fmt.Sprintf("<conn-event> nodes: %s->%s up: %t", e.Conn.One.TerminalString(), e.Conn.Other.TerminalString(), e.Conn.Up)
case EventTypeMsg:
return fmt.Sprintf("<msg-event> nodes: %s->%s code: %d, received: %t", e.Msg.One.Label(), e.Msg.Other.Label(), e.Msg.Code, e.Msg.Received)
return fmt.Sprintf("<msg-event> nodes: %s->%s code: %d, received: %t", e.Msg.One.TerminalString(), e.Msg.Other.TerminalString(), e.Msg.Code, e.Msg.Received)
default:
return ""
}

7
p2p/simulations/examples/connectivity.go
View file

@ -13,6 +13,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rpc"
@ -26,7 +27,7 @@ func main() {
log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
services := map[string]adapters.ServiceFunc{
"ping-pong": func(id *adapters.NodeId, snapshot []byte) node.Service {
"ping-pong": func(id discover.NodeID, snapshot []byte) node.Service {
return newPingPongService(id)
},
}
@ -73,12 +74,12 @@ func main() {
// sends a ping to all its connected peers every 10s and receives a pong in
// return
type pingPongService struct {
id *adapters.NodeId
id discover.NodeID
log log.Logger
received int64
}
func newPingPongService(id *adapters.NodeId) *pingPongService {
func newPingPongService(id discover.NodeID) *pingPongService {
return &pingPongService{
id: id,
log: log.New("node.id", id),

20
p2p/simulations/http.go
View file

@ -260,7 +260,7 @@ type ServerConfig struct {
// generate some mock events in the network
Mocker func(*Network)
// In case of multiple mockers, set the default here
DefaultMockerId string
DefaultMockerID string
// map of Mockers
Mockers map[string]*MockerConfig
}
@ -322,14 +322,14 @@ func (s *Server) CreateNetwork(w http.ResponseWriter, req *http.Request) {
network, err := func() (*Network, error) {
s.mtx.Lock()
defer s.mtx.Unlock()
if config.Id == "" {
config.Id = fmt.Sprintf("net%d", len(s.networks)+1)
if config.ID == "" {
config.ID = fmt.Sprintf("net%d", len(s.networks)+1)
}
if _, exists := s.networks[config.Id]; exists {
return nil, fmt.Errorf("network exists: %s", config.Id)
if _, exists := s.networks[config.ID]; exists {
return nil, fmt.Errorf("network exists: %s", config.ID)
}
network := NewNetwork(s.NewAdapter(), config)
s.networks[config.Id] = network
s.networks[config.ID] = network
return network, nil
}()
if err != nil {
@ -393,7 +393,7 @@ func (s *Server) DeleteNetwork(w http.ResponseWriter, req *http.Request) {
}
s.mtx.Lock()
delete(s.networks, network.Id)
delete(s.networks, network.ID)
s.mtx.Unlock()
w.WriteHeader(http.StatusOK)
@ -421,7 +421,7 @@ func (s *Server) StartMocker(w http.ResponseWriter, req *http.Request) {
if mockerid == "default" {
//choose the default mocker
mockerid = s.DefaultMockerId
mockerid = s.DefaultMockerID
}
if mocker, ok := s.Mockers[mockerid]; ok {
@ -689,7 +689,7 @@ func (s *Server) wrapHandler(handler http.HandlerFunc) httprouter.Handle {
}
var node *Node
if nodeID, err := discover.HexID(id); err == nil {
node = network.GetNode(&adapters.NodeId{NodeID: nodeID})
node = network.GetNode(nodeID)
} else {
node = network.GetNodeByName(id)
}
@ -707,7 +707,7 @@ func (s *Server) wrapHandler(handler http.HandlerFunc) httprouter.Handle {
}
var peer *Node
if peerID, err := discover.HexID(id); err == nil {
peer = network.GetNode(&adapters.NodeId{NodeID: peerID})
peer = network.GetNode(peerID)
} else {
peer = network.GetNodeByName(id)
}

63
p2p/simulations/http_test.go
View file

@ -12,18 +12,19 @@ import (
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rpc"
)
type testService struct {
id *adapters.NodeId
id discover.NodeID
// state stores []byte used to test creating and loading snapshots
state atomic.Value
}
func newTestService(id *adapters.NodeId, snapshot []byte) node.Service {
func newTestService(id discover.NodeID, snapshot []byte) node.Service {
svc := &testService{id: id}
svc.state.Store(snapshot)
return svc
@ -151,15 +152,15 @@ func TestHTTPNetwork(t *testing.T) {
// subscribe to events so we can check them later
events := make(chan *Event, 100)
sub, err := client.SubscribeNetwork(network.Id, events)
sub, err := client.SubscribeNetwork(network.ID, events)
if err != nil {
t.Fatalf("error subscribing to network events: %s", err)
}
defer sub.Unsubscribe()
// check the network has an ID
if network.Id == "" {
t.Fatal("expected network.Id to be set")
if network.ID == "" {
t.Fatal("expected network.ID to be set")
}
// check the network exists
@ -170,22 +171,22 @@ func TestHTTPNetwork(t *testing.T) {
if len(networks) != 1 {
t.Fatalf("expected 1 network, got %d", len(networks))
}
if networks[0].Id != network.Id {
t.Fatalf("expected network to have ID %q, got %q", network.Id, networks[0].Id)
if networks[0].ID != network.ID {
t.Fatalf("expected network to have ID %q, got %q", network.ID, networks[0].ID)
}
gotNetwork, err := client.GetNetwork(network.Id)
gotNetwork, err := client.GetNetwork(network.ID)
if err != nil {
t.Fatalf("error getting network: %s", err)
}
if gotNetwork.Id != network.Id {
t.Fatalf("expected network to have ID %q, got %q", network.Id, gotNetwork.Id)
if gotNetwork.ID != network.ID {
t.Fatalf("expected network to have ID %q, got %q", network.ID, gotNetwork.ID)
}
// create 2 nodes
nodeIDs := make([]string, 2)
for i := 0; i < 2; i++ {
config := &adapters.NodeConfig{}
node, err := client.CreateNode(network.Id, config)
node, err := client.CreateNode(network.ID, config)
if err != nil {
t.Fatalf("error creating node: %s", err)
}
@ -193,7 +194,7 @@ func TestHTTPNetwork(t *testing.T) {
}
// check both nodes exist
nodes, err := client.GetNodes(network.Id)
nodes, err := client.GetNodes(network.ID)
if err != nil {
t.Fatalf("error getting nodes: %s", err)
}
@ -204,7 +205,7 @@ func TestHTTPNetwork(t *testing.T) {
if nodes[i].ID != nodeID {
t.Fatalf("expected node %d to have ID %q, got %q", i, nodeID, nodes[i].ID)
}
node, err := client.GetNode(network.Id, nodeID)
node, err := client.GetNode(network.ID, nodeID)
if err != nil {
t.Fatalf("error getting node %d: %s", i, err)
}
@ -215,13 +216,13 @@ func TestHTTPNetwork(t *testing.T) {
// start both nodes
for _, nodeID := range nodeIDs {
if err := client.StartNode(network.Id, nodeID); err != nil {
if err := client.StartNode(network.ID, nodeID); err != nil {
t.Fatalf("error starting node %q: %s", nodeID, err)
}
}
// connect the nodes
if err := client.ConnectNode(network.Id, nodeIDs[0], nodeIDs[1]); err != nil {
if err := client.ConnectNode(network.ID, nodeIDs[0], nodeIDs[1]); err != nil {
t.Fatalf("error connecting nodes: %s", err)
}
@ -249,7 +250,7 @@ func (t *expectEvents) nodeEvent(id string, up bool) *Event {
Type: EventTypeNode,
Node: &Node{
Config: &adapters.NodeConfig{
Id: adapters.NewNodeIdFromHex(id),
ID: discover.MustHexID(id),
},
Up: up,
},
@ -260,8 +261,8 @@ func (t *expectEvents) connEvent(one, other string, up bool) *Event {
return &Event{
Type: EventTypeConn,
Conn: &Conn{
One: adapters.NewNodeIdFromHex(one),
Other: adapters.NewNodeIdFromHex(other),
One: discover.MustHexID(one),
Other: discover.MustHexID(other),
Up: up,
},
}
@ -334,22 +335,22 @@ func TestHTTPNodeRPC(t *testing.T) {
if err != nil {
t.Fatalf("error creating network: %s", err)
}
node, err := client.CreateNode(network.Id, &adapters.NodeConfig{})
node, err := client.CreateNode(network.ID, &adapters.NodeConfig{})
if err != nil {
t.Fatalf("error creating node: %s", err)
}
if err := client.StartNode(network.Id, node.ID); err != nil {
if err := client.StartNode(network.ID, node.ID); err != nil {
t.Fatalf("error starting node: %s", err)
}
// create two RPC clients
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
rpcClient1, err := client.RPCClient(ctx, network.Id, node.ID)
rpcClient1, err := client.RPCClient(ctx, network.ID, node.ID)
if err != nil {
t.Fatalf("error getting node RPC client: %s", err)
}
rpcClient2, err := client.RPCClient(ctx, network.Id, node.ID)
rpcClient2, err := client.RPCClient(ctx, network.ID, node.ID)
if err != nil {
t.Fatalf("error getting node RPC client: %s", err)
}
@ -400,23 +401,23 @@ func TestHTTPSnapshot(t *testing.T) {
nodeCount := 2
nodes := make([]*p2p.NodeInfo, nodeCount)
for i := 0; i < nodeCount; i++ {
node, err := client.CreateNode(network.Id, &adapters.NodeConfig{})
node, err := client.CreateNode(network.ID, &adapters.NodeConfig{})
if err != nil {
t.Fatalf("error creating node: %s", err)
}
if err := client.StartNode(network.Id, node.ID); err != nil {
if err := client.StartNode(network.ID, node.ID); err != nil {
t.Fatalf("error starting node: %s", err)
}
nodes[i] = node
}
if err := client.ConnectNode(network.Id, nodes[0].ID, nodes[1].ID); err != nil {
if err := client.ConnectNode(network.ID, nodes[0].ID, nodes[1].ID); err != nil {
t.Fatalf("error connecting nodes: %s", err)
}
// store some state in the test services
states := make([]string, nodeCount)
for i, node := range nodes {
rpc, err := client.RPCClient(context.Background(), network.Id, node.ID)
rpc, err := client.RPCClient(context.Background(), network.ID, node.ID)
if err != nil {
t.Fatalf("error getting RPC client: %s", err)
}
@ -429,7 +430,7 @@ func TestHTTPSnapshot(t *testing.T) {
}
// create a snapshot
snap, err := client.CreateSnapshot(network.Id)
snap, err := client.CreateSnapshot(network.ID)
if err != nil {
t.Fatalf("error creating snapshot: %s", err)
}
@ -447,19 +448,19 @@ func TestHTTPSnapshot(t *testing.T) {
// subscribe to events so we can check them later
events := make(chan *Event, 100)
sub, err := client.SubscribeNetwork(network.Id, events)
sub, err := client.SubscribeNetwork(network.ID, events)
if err != nil {
t.Fatalf("error subscribing to network events: %s", err)
}
defer sub.Unsubscribe()
// load the snapshot
if err := client.LoadSnapshot(network.Id, snap); err != nil {
if err := client.LoadSnapshot(network.ID, snap); err != nil {
t.Fatalf("error loading snapshot: %s", err)
}
// check the nodes and connection exists
net, err := client.GetNetwork(network.Id)
net, err := client.GetNetwork(network.ID)
if err != nil {
t.Fatalf("error getting network: %s", err)
}
@ -485,7 +486,7 @@ func TestHTTPSnapshot(t *testing.T) {
// check the node states were restored
for i, node := range nodes {
rpc, err := client.RPCClient(context.Background(), network.Id, node.ID)
rpc, err := client.RPCClient(context.Background(), network.ID, node.ID)
if err != nil {
t.Fatalf("error getting RPC client: %s", err)
}

21
p2p/simulations/mocker.go
View file

@ -8,11 +8,12 @@ import (
"github.com/ethereum/go-ethereum/crypto"
"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/simulations/adapters"
)
type MockerConfig struct {
Id string
ID string
NodeCount int
UpdateInterval int
Mocker func(*Network)
@ -30,7 +31,7 @@ type MockerConfig struct {
func DefaultMockerConfig() *MockerConfig {
return &MockerConfig{
Id: "0",
ID: "0",
NodeCount: 100,
UpdateInterval: 1000,
SwitchonRate: 5,
@ -59,7 +60,7 @@ func DefaultMockerConfig() *MockerConfig {
// to the eventer
// The journal using the eventer can then be read to visualise or
// drive connections
func MockEvents(eventer *event.Feed, ids []*adapters.NodeId, conf *MockerConfig) {
func MockEvents(eventer *event.Feed, ids []discover.NodeID, conf *MockerConfig) {
var onNodes []*Node
offNodes := ids
@ -114,7 +115,7 @@ func MockEvents(eventer *event.Feed, ids []*adapters.NodeId, conf *MockerConfig)
var mustconnect []int
for i := 0; len(offNodes) > 0 && i < nodesUp; i++ {
c := rand.Intn(len(offNodes))
sn := &Node{Config: &adapters.NodeConfig{Id: offNodes[c]}}
sn := &Node{Config: &adapters.NodeConfig{ID: offNodes[c]}}
eventer.Send(ControlEvent(sn))
mustconnect = append(mustconnect, len(onNodes))
onNodes = append(onNodes, sn)
@ -186,19 +187,21 @@ func MockEvents(eventer *event.Feed, ids []*adapters.NodeId, conf *MockerConfig)
}
}
func RandomNodeId() *adapters.NodeId {
func RandomNodeID() discover.NodeID {
key, err := crypto.GenerateKey()
if err != nil {
panic("unable to generate key")
}
pubkey := crypto.FromECDSAPub(&key.PublicKey)
return adapters.NewNodeId(pubkey[1:])
var id discover.NodeID
copy(id[:], pubkey[1:])
return id
}
func RandomNodeIds(n int) []*adapters.NodeId {
var ids []*adapters.NodeId
func RandomNodeIDs(n int) []discover.NodeID {
ids := make([]discover.NodeID, n)
for i := 0; i < n; i++ {
ids = append(ids, RandomNodeId())
ids[i] = RandomNodeID()
}
return ids
}

128
p2p/simulations/network.go
View file

@ -41,7 +41,7 @@ import (
)
type NetworkConfig struct {
Id string `json:"id"`
ID string `json:"id"`
DefaultService string `json:"default_service,omitempty"`
}
@ -141,12 +141,12 @@ type Node struct {
controlFired bool
}
func (self *Node) ID() *adapters.NodeId {
return self.Config.Id
func (self *Node) ID() discover.NodeID {
return self.Config.ID
}
func (self *Node) String() string {
return fmt.Sprintf("Node %v", self.ID().Label())
return fmt.Sprintf("Node %v", self.ID().TerminalString())
}
func (self *Node) NodeInfo() *p2p.NodeInfo {
@ -159,8 +159,8 @@ func (self *Node) NodeInfo() *p2p.NodeInfo {
// you journal updates could filter if passive knowledge about peers is
// irrelevant
type Conn struct {
One *adapters.NodeId `json:"one"`
Other *adapters.NodeId `json:"other"`
One discover.NodeID `json:"one"`
Other discover.NodeID `json:"other"`
one, other *Node
// connection down by default
Up bool `json:"up"`
@ -172,19 +172,19 @@ type Conn struct {
}
func (self *Conn) String() string {
return fmt.Sprintf("Conn %v->%v", self.One.Label(), self.Other.Label())
return fmt.Sprintf("Conn %v->%v", self.One.TerminalString(), self.Other.TerminalString())
}
type Msg struct {
One *adapters.NodeId `json:"one"`
Other *adapters.NodeId `json:"other"`
One discover.NodeID `json:"one"`
Other discover.NodeID `json:"other"`
Code uint64 `json:"code"`
Received bool `json:"received"`
controlFired bool
}
func (self *Msg) String() string {
return fmt.Sprintf("Msg(%d) %v->%v", self.Code, self.One.Label(), self.Other.Label())
return fmt.Sprintf("Msg(%d) %v->%v", self.Code, self.One.TerminalString(), self.Other.TerminalString())
}
// NewNode adds a new node to the network with a random ID
@ -199,7 +199,7 @@ func (self *Network) NewNode() (*Node, error) {
func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error) {
self.lock.Lock()
defer self.lock.Unlock()
id := conf.Id
id := conf.ID
if conf.Name == "" {
conf.Name = fmt.Sprintf("node%02d", len(self.Nodes)+1)
}
@ -207,11 +207,11 @@ func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error)
conf.Services = []string{self.DefaultService}
}
_, found := self.nodeMap[id.NodeID]
_, found := self.nodeMap[id]
if found {
return nil, fmt.Errorf("node %v already added", id)
}
self.nodeMap[id.NodeID] = len(self.Nodes)
self.nodeMap[id] = len(self.Nodes)
adapterNode, err := self.nodeAdapter.NewNode(conf)
if err != nil {
@ -233,18 +233,18 @@ func (self *Network) Config() *NetworkConfig {
// newConn adds a new connection to the network
// it errors if the respective nodes do not exist
func (self *Network) newConn(oneId, otherId *adapters.NodeId) (*Conn, error) {
one := self.getNode(oneId)
func (self *Network) newConn(oneID, otherID discover.NodeID) (*Conn, error) {
one := self.getNode(oneID)
if one == nil {
return nil, fmt.Errorf("one %v does not exist", one)
}
other := self.getNode(otherId)
other := self.getNode(otherID)
if other == nil {
return nil, fmt.Errorf("other %v does not exist", other)
}
return &Conn{
One: oneId,
Other: otherId,
One: oneID,
Other: otherID,
one: one,
other: other,
}, nil
@ -285,11 +285,11 @@ func (self *Network) StopAll() error {
}
// Start(id) starts up the node (relevant only for instance with own p2p or remote)
func (self *Network) Start(id *adapters.NodeId) error {
func (self *Network) Start(id discover.NodeID) error {
return self.startWithSnapshots(id, nil)
}
func (self *Network) startWithSnapshots(id *adapters.NodeId, snapshots map[string][]byte) error {
func (self *Network) startWithSnapshots(id discover.NodeID, snapshots map[string][]byte) error {
node := self.GetNode(id)
if node == nil {
return fmt.Errorf("node %v does not exist", id)
@ -321,7 +321,7 @@ func (self *Network) startWithSnapshots(id *adapters.NodeId, snapshots map[strin
return nil
}
func (self *Network) watchPeerEvents(id *adapters.NodeId, events chan *p2p.PeerEvent, sub event.Subscription) {
func (self *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEvent, sub event.Subscription) {
defer sub.Unsubscribe()
for {
select {
@ -329,23 +329,23 @@ func (self *Network) watchPeerEvents(id *adapters.NodeId, events chan *p2p.PeerE
if !ok {
return
}
peer := &adapters.NodeId{NodeID: event.Peer}
peer := event.Peer
switch event.Type {
case p2p.PeerEventTypeAdd:
if err := self.DidConnect(id, peer); err != nil {
log.Error(fmt.Sprintf("error generating connection up event %s => %s", id.Label(), peer.Label()), "err", err)
log.Error(fmt.Sprintf("error generating connection up event %s => %s", id.TerminalString(), peer.TerminalString()), "err", err)
}
case p2p.PeerEventTypeDrop:
if err := self.DidDisconnect(id, peer); err != nil {
log.Error(fmt.Sprintf("error generating connection down event %s => %s", id.Label(), peer.Label()), "err", err)
log.Error(fmt.Sprintf("error generating connection down event %s => %s", id.TerminalString(), peer.TerminalString()), "err", err)
}
case p2p.PeerEventTypeMsgSend:
if err := self.DidSend(id, peer, *event.MsgCode); err != nil {
log.Error(fmt.Sprintf("error generating msg send event %s => %s", id.Label(), peer.Label()), "err", err)
log.Error(fmt.Sprintf("error generating msg send event %s => %s", id.TerminalString(), peer.TerminalString()), "err", err)
}
case p2p.PeerEventTypeMsgRecv:
if err := self.DidReceive(peer, id, *event.MsgCode); err != nil {
log.Error(fmt.Sprintf("error generating msg receive event %s => %s", peer.Label(), id.Label()), "err", err)
log.Error(fmt.Sprintf("error generating msg receive event %s => %s", peer.TerminalString(), id.TerminalString()), "err", err)
}
}
case err := <-sub.Err():
@ -358,7 +358,7 @@ func (self *Network) watchPeerEvents(id *adapters.NodeId, events chan *p2p.PeerE
}
// Stop(id) shuts down the node (relevant only for instance with own p2p or remote)
func (self *Network) Stop(id *adapters.NodeId) error {
func (self *Network) Stop(id discover.NodeID) error {
node := self.GetNode(id)
if node == nil {
return fmt.Errorf("node %v does not exist", id)
@ -376,24 +376,24 @@ func (self *Network) Stop(id *adapters.NodeId) error {
return nil
}
// Connect(i, j) attempts to connect nodes i and j (args given as nodeId)
// Connect(i, j) attempts to connect nodes i and j (args given as nodeID)
// calling the node's nodadapters Connect method
// connection is established (as if) the first node dials out to the other
func (self *Network) Connect(oneId, otherId *adapters.NodeId) error {
log.Debug(fmt.Sprintf("connecting %s to %s", oneId, otherId))
conn, err := self.GetOrCreateConn(oneId, otherId)
func (self *Network) Connect(oneID, otherID discover.NodeID) error {
log.Debug(fmt.Sprintf("connecting %s to %s", oneID, otherID))
conn, err := self.GetOrCreateConn(oneID, otherID)
if err != nil {
return err
}
if conn.Up {
return fmt.Errorf("%v and %v already connected", oneId, otherId)
return fmt.Errorf("%v and %v already connected", oneID, otherID)
}
err = conn.nodesUp()
if err != nil {
return err
}
var rev bool
if conn.One.NodeID != oneId.NodeID {
if conn.One != oneID {
rev = true
}
// if Connect is called because of external trigger, it needs to call
@ -417,21 +417,21 @@ func (self *Network) Connect(oneId, otherId *adapters.NodeId) error {
return client.Call(nil, "admin_addPeer", string(addr))
}
// Disconnect(i, j) attempts to disconnect nodes i and j (args given as nodeId)
// Disconnect(i, j) attempts to disconnect nodes i and j (args given as nodeID)
// calling the node's nodadapters Disconnect method
// sets the Conn model to Down
// the disconnect will be initiated (the connection is dropped by) the first node
// it errors if either of the nodes is down (or does not exist)
func (self *Network) Disconnect(oneId, otherId *adapters.NodeId) error {
conn := self.GetConn(oneId, otherId)
func (self *Network) Disconnect(oneID, otherID discover.NodeID) error {
conn := self.GetConn(oneID, otherID)
if conn == nil {
return fmt.Errorf("connection between %v and %v does not exist", oneId, otherId)
return fmt.Errorf("connection between %v and %v does not exist", oneID, otherID)
}
if !conn.Up {
return fmt.Errorf("%v and %v already disconnected", oneId, otherId)
return fmt.Errorf("%v and %v already disconnected", oneID, otherID)
}
var rev bool
if conn.One.NodeID != oneId.NodeID {
if conn.One != oneID {
rev = true
}
var addr []byte
@ -451,7 +451,7 @@ func (self *Network) Disconnect(oneId, otherId *adapters.NodeId) error {
return client.Call(nil, "admin_removePeer", string(addr))
}
func (self *Network) DidConnect(one, other *adapters.NodeId) error {
func (self *Network) DidConnect(one, other discover.NodeID) error {
conn, err := self.GetOrCreateConn(one, other)
if err != nil {
return fmt.Errorf("connection between %v and %v does not exist", one, other)
@ -459,14 +459,14 @@ func (self *Network) DidConnect(one, other *adapters.NodeId) error {
if conn.Up {
return fmt.Errorf("%v and %v already connected", one, other)
}
conn.Reverse = conn.One.NodeID != one.NodeID
conn.Reverse = conn.One != one
conn.Up = true
// connection event posted
self.events.Send(NewEvent(conn))
return nil
}
func (self *Network) DidDisconnect(one, other *adapters.NodeId) error {
func (self *Network) DidDisconnect(one, other discover.NodeID) error {
conn, err := self.GetOrCreateConn(one, other)
if err != nil {
return fmt.Errorf("connection between %v and %v does not exist", one, other)
@ -474,14 +474,14 @@ func (self *Network) DidDisconnect(one, other *adapters.NodeId) error {
if !conn.Up {
return fmt.Errorf("%v and %v already disconnected", one, other)
}
conn.Reverse = conn.One.NodeID != one.NodeID
conn.Reverse = conn.One != one
conn.Up = false
self.events.Send(NewEvent(conn))
return nil
}
// Send(senderid, receiverid) sends a message from one node to another
func (self *Network) Send(senderid, receiverid *adapters.NodeId, msgcode uint64, protomsg interface{}) {
func (self *Network) Send(senderid, receiverid discover.NodeID, msgcode uint64, protomsg interface{}) {
msg := &Msg{
One: senderid,
Other: receiverid,
@ -491,7 +491,7 @@ func (self *Network) Send(senderid, receiverid *adapters.NodeId, msgcode uint64,
self.events.Send(ControlEvent(msg))
}
func (self *Network) DidSend(sender, receiver *adapters.NodeId, msgcode uint64) error {
func (self *Network) DidSend(sender, receiver discover.NodeID, msgcode uint64) error {
msg := &Msg{
One: sender,
Other: receiver,
@ -502,7 +502,7 @@ func (self *Network) DidSend(sender, receiver *adapters.NodeId, msgcode uint64)
return nil
}
func (self *Network) DidReceive(sender, receiver *adapters.NodeId, msgcode uint64) error {
func (self *Network) DidReceive(sender, receiver discover.NodeID, msgcode uint64) error {
msg := &Msg{
One: sender,
Other: receiver,
@ -515,7 +515,7 @@ func (self *Network) DidReceive(sender, receiver *adapters.NodeId, msgcode uint6
// GetNode retrieves the node model for the id given as arg
// returns nil if the node does not exist
func (self *Network) GetNode(id *adapters.NodeId) *Node {
func (self *Network) GetNode(id discover.NodeID) *Node {
self.lock.Lock()
defer self.lock.Unlock()
return self.getNode(id)
@ -538,8 +538,8 @@ func (self *Network) GetNodes() []*Node {
return self.Nodes
}
func (self *Network) getNode(id *adapters.NodeId) *Node {
i, found := self.nodeMap[id.NodeID]
func (self *Network) getNode(id discover.NodeID) *Node {
i, found := self.nodeMap[id]
if !found {
return nil
}
@ -549,34 +549,34 @@ func (self *Network) getNode(id *adapters.NodeId) *Node {
// GetConn(i, j) retrieves the connectiton model for the connection between
// the order of nodes does not matter, i.e., GetConn(i,j) == GetConn(j, i)
// returns nil if the node does not exist
func (self *Network) GetConn(oneId, otherId *adapters.NodeId) *Conn {
func (self *Network) GetConn(oneID, otherID discover.NodeID) *Conn {
self.lock.Lock()
defer self.lock.Unlock()
return self.getConn(oneId, otherId)
return self.getConn(oneID, otherID)
}
// GetConn(i, j) retrieves the connectiton model for the connection between
// i and j, or creates a new one if it does not exist
// the order of nodes does not matter, i.e., GetConn(i,j) == GetConn(j, i)
func (self *Network) GetOrCreateConn(oneId, otherId *adapters.NodeId) (*Conn, error) {
func (self *Network) GetOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) {
self.lock.Lock()
defer self.lock.Unlock()
conn := self.getConn(oneId, otherId)
conn := self.getConn(oneID, otherID)
if conn != nil {
return conn, nil
}
conn, err := self.newConn(oneId, otherId)
conn, err := self.newConn(oneID, otherID)
if err != nil {
return nil, err
}
label := ConnLabel(oneId, otherId)
label := ConnLabel(oneID, otherID)
self.connMap[label] = len(self.Conns)
self.Conns = append(self.Conns, conn)
return conn, nil
}
func (self *Network) getConn(oneId, otherId *adapters.NodeId) *Conn {
label := ConnLabel(oneId, otherId)
func (self *Network) getConn(oneID, otherID discover.NodeID) *Conn {
label := ConnLabel(oneID, otherID)
i, found := self.connMap[label]
if !found {
return nil
@ -587,23 +587,23 @@ func (self *Network) getConn(oneId, otherId *adapters.NodeId) *Conn {
func (self *Network) Shutdown() {
// disconnect all nodes
for _, conn := range self.Conns {
log.Debug(fmt.Sprintf("disconnecting %s from %s", conn.One.Label(), conn.Other.Label()))
log.Debug(fmt.Sprintf("disconnecting %s from %s", conn.One.TerminalString(), conn.Other.TerminalString()))
if err := self.Disconnect(conn.One, conn.Other); err != nil {
log.Warn(fmt.Sprintf("error disconnecting %s from %s", conn.One.Label(), conn.Other.Label()), "err", err)
log.Warn(fmt.Sprintf("error disconnecting %s from %s", conn.One.TerminalString(), conn.Other.TerminalString()), "err", err)
}
}
// stop all nodes
for _, node := range self.Nodes {
log.Debug(fmt.Sprintf("stopping node %s", node.ID().Label()))
log.Debug(fmt.Sprintf("stopping node %s", node.ID().TerminalString()))
if err := node.Stop(); err != nil {
log.Warn(fmt.Sprintf("error stopping node %s", node.ID().Label()), "err", err)
log.Warn(fmt.Sprintf("error stopping node %s", node.ID().TerminalString()), "err", err)
}
}
}
func ConnLabel(source, target *adapters.NodeId) string {
var first, second *adapters.NodeId
func ConnLabel(source, target discover.NodeID) string {
var first, second discover.NodeID
if bytes.Compare(source.Bytes(), target.Bytes()) > 0 {
first = target
second = source
@ -662,7 +662,7 @@ func (self *Network) Load(snap *Snapshot) error {
if !node.Up {
continue
}
if err := self.startWithSnapshots(node.Config.Id, node.Snapshots); err != nil {
if err := self.startWithSnapshots(node.Config.ID, node.Snapshots); err != nil {
return err
}
}

12
p2p/simulations/simulation.go
View file

@ -4,7 +4,7 @@ import (
"context"
"time"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/p2p/discover"
)
// Simulation provides a framework for running actions in a simulated network
@ -94,7 +94,7 @@ type Step struct {
// Trigger is a channel which receives node ids and triggers an
// expectation check for that node
Trigger chan *adapters.NodeId
Trigger chan discover.NodeID
// Expect is the expectation to wait for when performing this step
Expect *Expectation
@ -102,15 +102,15 @@ type Step struct {
type Expectation struct {
// Nodes is a list of nodes to check
Nodes []*adapters.NodeId
Nodes []discover.NodeID
// Check checks whether a given node meets the expectation
Check func(context.Context, *adapters.NodeId) (bool, error)
Check func(context.Context, discover.NodeID) (bool, error)
}
func newStepResult() *StepResult {
return &StepResult{
Passes: make(map[*adapters.NodeId]time.Time),
Passes: make(map[discover.NodeID]time.Time),
}
}
@ -125,7 +125,7 @@ type StepResult struct {
FinishedAt time.Time
// Passes are the timestamps of the successful node expectations
Passes map[*adapters.NodeId]time.Time
Passes map[discover.NodeID]time.Time
// NetworkEvents are the network events which occurred during the step
NetworkEvents []*Event

9
p2p/testing/peerpool.go
View file

@ -6,7 +6,6 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
)
type TestPeer interface {
@ -38,15 +37,15 @@ func (self *TestPeerPool) Remove(p TestPeer) {
delete(self.peers, p.ID())
}
func (self *TestPeerPool) Has(n *adapters.NodeId) bool {
func (self *TestPeerPool) Has(id discover.NodeID) bool {
self.lock.Lock()
defer self.lock.Unlock()
_, ok := self.peers[n.NodeID]
_, ok := self.peers[id]
return ok
}
func (self *TestPeerPool) Get(n *adapters.NodeId) TestPeer {
func (self *TestPeerPool) Get(id discover.NodeID) TestPeer {
self.lock.Lock()
defer self.lock.Unlock()
return self.peers[n.NodeID]
return self.peers[id]
}

18
p2p/testing/protocolsession.go
View file

@ -14,7 +14,7 @@ import (
type ProtocolSession struct {
Server *p2p.Server
Ids []*adapters.NodeId
IDs []discover.NodeID
adapter *adapters.SimAdapter
events chan *p2p.PeerEvent
}
@ -35,27 +35,27 @@ type Exchange struct {
type Trigger struct {
Msg interface{} // type of message to be sent
Code uint64 // code of message is given
Peer *adapters.NodeId // the peer to send the message to
Peer discover.NodeID // the peer to send the message to
Timeout time.Duration // timeout duration for the sending
}
type Expect struct {
Msg interface{} // type of message to expect
Code uint64 // code of message is now given
Peer *adapters.NodeId // the peer that expects the message
Peer discover.NodeID // the peer that expects the message
Timeout time.Duration // timeout duration for receiving
}
type Disconnect struct {
Peer *adapters.NodeId // discconnected peer
Peer discover.NodeID // discconnected peer
Error error // disconnect reason
}
// trigger sends messages from peers
func (self *ProtocolSession) trigger(trig Trigger) error {
simNode, ok := self.adapter.GetNode(trig.Peer.NodeID)
simNode, ok := self.adapter.GetNode(trig.Peer)
if !ok {
return fmt.Errorf("trigger: peer %v does not exist (1- %v)", trig.Peer, len(self.Ids))
return fmt.Errorf("trigger: peer %v does not exist (1- %v)", trig.Peer, len(self.IDs))
}
mockNode, ok := simNode.Services()[0].(*mockNode)
if !ok {
@ -88,9 +88,9 @@ func (self *ProtocolSession) expect(exp Expect) error {
if exp.Msg == nil {
return errors.New("no message to expect")
}
simNode, ok := self.adapter.GetNode(exp.Peer.NodeID)
simNode, ok := self.adapter.GetNode(exp.Peer)
if !ok {
return fmt.Errorf("trigger: peer %v does not exist (1- %v)", exp.Peer, len(self.Ids))
return fmt.Errorf("trigger: peer %v does not exist (1- %v)", exp.Peer, len(self.IDs))
}
mockNode, ok := simNode.Services()[0].(*mockNode)
if !ok {
@ -178,7 +178,7 @@ func (self *ProtocolSession) TestExchanges(exchanges ...Exchange) error {
func (self *ProtocolSession) TestDisconnected(disconnects ...*Disconnect) error {
expects := make(map[discover.NodeID]error)
for _, disconnect := range disconnects {
expects[disconnect.Peer.NodeID] = disconnect.Error
expects[disconnect.Peer] = disconnect.Error
}
timeout := time.After(time.Second)

31
p2p/testing/protocoltester.go
View file

@ -8,6 +8,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rpc"
@ -18,18 +19,18 @@ type ProtocolTester struct {
network *simulations.Network
}
func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(*p2p.Peer, p2p.MsgReadWriter) error) *ProtocolTester {
func NewProtocolTester(t *testing.T, id discover.NodeID, n int, run func(*p2p.Peer, p2p.MsgReadWriter) error) *ProtocolTester {
services := adapters.Services{
"test": func(id *adapters.NodeId, _ []byte) node.Service {
"test": func(id discover.NodeID, _ []byte) node.Service {
return &testNode{run}
},
"mock": func(id *adapters.NodeId, _ []byte) node.Service {
"mock": func(id discover.NodeID, _ []byte) node.Service {
return newMockNode()
},
}
adapter := adapters.NewSimAdapter(services)
net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{})
if _, err := net.NewNodeWithConfig(&adapters.NodeConfig{Id: id, Services: []string{"test"}}); err != nil {
if _, err := net.NewNodeWithConfig(&adapters.NodeConfig{ID: id, Services: []string{"test"}}); err != nil {
panic(err.Error())
}
if err := net.Start(id); err != nil {
@ -38,17 +39,17 @@ func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(*p2p.P
node := net.GetNode(id).Node.(*adapters.SimNode)
peers := make([]*adapters.NodeConfig, n)
peerIDs := make([]*adapters.NodeId, n)
peerIDs := make([]discover.NodeID, n)
for i := 0; i < n; i++ {
peers[i] = adapters.RandomNodeConfig()
peers[i].Services = []string{"mock"}
peerIDs[i] = peers[i].Id
peerIDs[i] = peers[i].ID
}
events := make(chan *p2p.PeerEvent, 1000)
node.SubscribeEvents(events)
ps := &ProtocolSession{
Server: node.Server(),
Ids: peerIDs,
IDs: peerIDs,
adapter: adapter,
events: events,
}
@ -66,18 +67,18 @@ func (self *ProtocolTester) Stop() error {
return self.Server.Stop()
}
func (self *ProtocolTester) Connect(selfId *adapters.NodeId, peers ...*adapters.NodeConfig) {
func (self *ProtocolTester) Connect(selfID discover.NodeID, peers ...*adapters.NodeConfig) {
for _, peer := range peers {
log.Trace(fmt.Sprintf("start node %v", peer.Id))
log.Trace(fmt.Sprintf("start node %v", peer.ID))
if _, err := self.network.NewNodeWithConfig(peer); err != nil {
panic(fmt.Sprintf("error starting peer %v: %v", peer.Id, err))
panic(fmt.Sprintf("error starting peer %v: %v", peer.ID, err))
}
if err := self.network.Start(peer.Id); err != nil {
panic(fmt.Sprintf("error starting peer %v: %v", peer.Id, err))
if err := self.network.Start(peer.ID); err != nil {
panic(fmt.Sprintf("error starting peer %v: %v", peer.ID, err))
}
log.Trace(fmt.Sprintf("connect to %v", peer.Id))
if err := self.network.Connect(selfId, peer.Id); err != nil {
panic(fmt.Sprintf("error connecting to peer %v: %v", peer.Id, err))
log.Trace(fmt.Sprintf("connect to %v", peer.ID))
if err := self.network.Connect(selfID, peer.ID); err != nil {
panic(fmt.Sprintf("error connecting to peer %v: %v", peer.ID, err))
}
}

2
swarm/network/discovery_test.go
View file

@ -34,7 +34,7 @@ func TestDiscovery(t *testing.T) {
p2ptest.Expect{
Code: 3,
Msg: &subPeersMsg{Depth: 0},
Peer: s.ProtocolTester.Ids[0],
Peer: s.ProtocolTester.IDs[0],
},
},
})

9
swarm/network/hive.go
View file

@ -25,7 +25,6 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
)
/*
@ -188,7 +187,7 @@ func (self *Hive) NodeInfo() interface{} {
func (self *Hive) PeerInfo(id discover.NodeID) interface{} {
self.lock.Lock()
defer self.lock.Unlock()
addr := NewAddrFromNodeId(adapters.NewNodeId(id[:]))
addr := NewAddrFromNodeID(id)
return interface{}(addr)
}
@ -208,12 +207,6 @@ func (self *Hive) wake() {
}
}
// HexToBytes reads a hex string ontp
func HexToBytes(s string) []byte {
id := discover.MustHexID(s)
return id[:]
}
// ToAddr returns the serialisable version of u
func ToAddr(pa OverlayPeer) *bzzAddr {
if addr, ok := pa.(*bzzAddr); ok {

4
swarm/network/hive_test.go
View file

@ -21,8 +21,8 @@ func TestRegisterAndConnect(t *testing.T) {
s, pp := newHiveTester(t, params)
defer s.Stop()
id := s.Ids[0]
raddr := NewAddrFromNodeId(id)
id := s.IDs[0]
raddr := NewAddrFromNodeID(id)
ch := make(chan OverlayAddr)
go func() {

5
swarm/network/kademlia.go
View file

@ -24,7 +24,6 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/pot"
)
@ -506,7 +505,7 @@ func (self *Kademlia) Prune(c <-chan time.Time) {
}()
}
func NewPeerPot(kadMinProxSize int, ids ...*adapters.NodeId) map[discover.NodeID][][]byte {
func NewPeerPot(kadMinProxSize int, ids ...discover.NodeID) map[discover.NodeID][][]byte {
// create a table of all nodes for health check
np := pot.NewPot(nil, 0)
for _, id := range ids {
@ -526,7 +525,7 @@ func NewPeerPot(kadMinProxSize int, ids ...*adapters.NodeId) map[discover.NodeID
}
return pl == 0 || pl == po
})
nnmap[id.NodeID] = nns
nnmap[id] = nns
}
return nnmap
}

18
swarm/network/protocol.go
View file

@ -29,7 +29,6 @@ import (
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rpc"
)
@ -243,7 +242,7 @@ func (self *bzzPeer) LastActive() time.Time {
Handshake
* Version: 8 byte integer version of the protocol
* NetworkID: 8 byte integer network identifier
* NetworkId: 8 byte integer network identifier
* Addr: the address advertised by the node including underlay and overlay connecctions
*/
type bzzHandshake struct {
@ -340,17 +339,16 @@ func RandomAddr() *bzzAddr {
}
}
// NewNodeIdFromAddr transforms the underlay address to an adapters.NodeId
func NewNodeIdFromAddr(addr Addr) *adapters.NodeId {
return adapters.NewNodeId(addr.Under())
// NewNodeIDFromAddr transforms the underlay address to an adapters.NodeID
func NewNodeIDFromAddr(addr Addr) discover.NodeID {
return discover.MustBytesID(addr.Under())
}
// NewAddrFromNodeId constucts a bzzAddr from an adapters.NodeId
// the overlay address is derived as the hash of the nodeId
func NewAddrFromNodeId(n *adapters.NodeId) *bzzAddr {
id := n.NodeID
// NewAddrFromNodeID constucts a bzzAddr from a discover.NodeID
// the overlay address is derived as the hash of the nodeID
func NewAddrFromNodeID(id discover.NodeID) *bzzAddr {
return &bzzAddr{
OAddr: ToOverlayAddr(n.Bytes()),
OAddr: ToOverlayAddr(id.Bytes()),
UAddr: []byte(discover.NewNode(id, net.IP{127, 0, 0, 1}, 30303, 30303).String()),
}
}

38
swarm/network/protocol_test.go
View file

@ -6,8 +6,8 @@ import (
"testing"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
)
@ -38,7 +38,7 @@ func (t *testStore) Save(key string, v []byte) error {
return nil
}
func bzzHandshakeExchange(lhs, rhs *bzzHandshake, id *adapters.NodeId) []p2ptest.Exchange {
func bzzHandshakeExchange(lhs, rhs *bzzHandshake, id discover.NodeID) []p2ptest.Exchange {
return []p2ptest.Exchange{
p2ptest.Exchange{
@ -74,14 +74,14 @@ func newBzzBaseTester(t *testing.T, n int, addr *bzzAddr, spec *protocols.Spec,
return srv(&bzzPeer{
Peer: protocols.NewPeer(p, rw, spec),
localAddr: addr,
bzzAddr: NewAddrFromNodeId(&adapters.NodeId{NodeID: p.ID()}),
bzzAddr: NewAddrFromNodeID(p.ID()),
})
}
s := p2ptest.NewProtocolTester(t, NewNodeIdFromAddr(addr), n, protocall)
s := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), n, protocall)
for _, id := range s.Ids {
cs[id.NodeID.String()] = make(chan bool)
for _, id := range s.IDs {
cs[id.String()] = make(chan bool)
}
return &bzzTester{
@ -112,27 +112,27 @@ func newBzzTester(t *testing.T, n int, addr *bzzAddr, pp *p2ptest.TestPeerPool,
// should test handshakes in one exchange? parallelisation
func (s *bzzTester) testHandshake(lhs, rhs *bzzHandshake, disconnects ...*p2ptest.Disconnect) {
var peers []*adapters.NodeId
id := NewNodeIdFromAddr(rhs.Addr)
var peers []discover.NodeID
id := NewNodeIDFromAddr(rhs.Addr)
if len(disconnects) > 0 {
for _, d := range disconnects {
peers = append(peers, d.Peer)
}
} else {
peers = []*adapters.NodeId{id}
peers = []discover.NodeID{id}
}
s.TestExchanges(bzzHandshakeExchange(lhs, rhs, id)...)
s.TestDisconnected(disconnects...)
}
func (s *bzzTester) runHandshakes(ids ...*adapters.NodeId) {
func (s *bzzTester) runHandshakes(ids ...discover.NodeID) {
if len(ids) == 0 {
ids = s.Ids
ids = s.IDs
}
for _, id := range ids {
s.testHandshake(correctBzzHandshake(s.addr), correctBzzHandshake(NewAddrFromNodeId(id)))
<-s.cs[id.NodeID.String()]
s.testHandshake(correctBzzHandshake(s.addr), correctBzzHandshake(NewAddrFromNodeID(id)))
<-s.cs[id.String()]
}
}
@ -151,10 +151,10 @@ func TestBzzHandshakeNetworkIdMismatch(t *testing.T) {
s := newBzzTester(t, 1, addr, pp, nil, nil)
defer s.Stop()
id := s.Ids[0]
id := s.IDs[0]
s.testHandshake(
correctBzzHandshake(addr),
&bzzHandshake{Version: 0, NetworkId: 321, Addr: NewAddrFromNodeId(id)},
&bzzHandshake{Version: 0, NetworkId: 321, Addr: NewAddrFromNodeID(id)},
&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("network id mismatch 321 (!= 322)")},
)
}
@ -165,10 +165,10 @@ func TestBzzHandshakeVersionMismatch(t *testing.T) {
s := newBzzTester(t, 1, addr, pp, nil, nil)
defer s.Stop()
id := s.Ids[0]
id := s.IDs[0]
s.testHandshake(
correctBzzHandshake(addr),
&bzzHandshake{Version: 1, NetworkId: 322, Addr: NewAddrFromNodeId(id)},
&bzzHandshake{Version: 1, NetworkId: 322, Addr: NewAddrFromNodeID(id)},
&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("version mismatch 1 (!= 0)")},
)
}
@ -179,9 +179,9 @@ func TestBzzHandshakeSuccess(t *testing.T) {
s := newBzzTester(t, 1, addr, pp, nil, nil)
defer s.Stop()
id := s.Ids[0]
id := s.IDs[0]
s.testHandshake(
correctBzzHandshake(addr),
&bzzHandshake{Version: 0, NetworkId: 322, Addr: NewAddrFromNodeId(id)},
&bzzHandshake{Version: 0, NetworkId: 322, Addr: NewAddrFromNodeID(id)},
)
}

31
swarm/network/simulations/discovery/discovery_test.go
View file

@ -11,6 +11,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/network"
@ -59,22 +60,22 @@ func testDiscoverySimulation(t *testing.T, adapter adapters.NodeAdapter) {
// create 10 node network
nodeCount := 10
net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
Id: "0",
ID: "0",
DefaultService: serviceName,
})
defer net.Shutdown()
trigger := make(chan *adapters.NodeId)
ids := make([]*adapters.NodeId, nodeCount)
trigger := make(chan discover.NodeID)
ids := make([]discover.NodeID, nodeCount)
for i := 0; i < nodeCount; i++ {
node, err := net.NewNode()
if err != nil {
t.Fatalf("error starting node: %s", err)
}
if err := net.Start(node.ID()); err != nil {
t.Fatalf("error starting node %s: %s", node.ID().Label(), err)
t.Fatalf("error starting node %s: %s", node.ID().TerminalString(), err)
}
if err := triggerChecks(trigger, net, node.ID()); err != nil {
t.Fatal("error triggering checks for node %s: %s", node.ID().Label(), err)
t.Fatal("error triggering checks for node %s: %s", node.ID().TerminalString(), err)
}
ids[i] = node.ID()
}
@ -83,20 +84,20 @@ func testDiscoverySimulation(t *testing.T, adapter adapters.NodeAdapter) {
// for full peer discovery
action := func(ctx context.Context) error {
for i, id := range ids {
var peerId *adapters.NodeId
var peerID discover.NodeID
if i == 0 {
peerId = ids[len(ids)-1]
peerID = ids[len(ids)-1]
} else {
peerId = ids[i-1]
peerID = ids[i-1]
}
if err := net.Connect(id, peerId); err != nil {
if err := net.Connect(id, peerID); err != nil {
return err
}
}
return nil
}
nnmap := network.NewPeerPot(testMinProxBinSize, ids...)
check := func(ctx context.Context, id *adapters.NodeId) (bool, error) {
check := func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
return false, ctx.Err()
@ -112,7 +113,7 @@ func testDiscoverySimulation(t *testing.T, adapter adapters.NodeAdapter) {
return false, fmt.Errorf("error getting node client: %s", err)
}
var healthy bool
if err := client.Call(&healthy, "hive_healthy", nnmap[id.NodeID]); err != nil {
if err := client.Call(&healthy, "hive_healthy", nnmap[id]); err != nil {
return false, fmt.Errorf("error getting node health: %s", err)
}
return healthy, nil
@ -137,7 +138,7 @@ func testDiscoverySimulation(t *testing.T, adapter adapters.NodeAdapter) {
t.Log("Simulation Passed:")
t.Logf("Duration: %s", result.FinishedAt.Sub(result.StartedAt))
for _, id := range ids {
t.Logf("Node %s passed in %s", id.Label(), result.Passes[id].Sub(result.StartedAt))
t.Logf("Node %s passed in %s", id.TerminalString(), result.Passes[id].Sub(result.StartedAt))
}
t.Logf("Events:")
for _, event := range result.NetworkEvents {
@ -147,7 +148,7 @@ func testDiscoverySimulation(t *testing.T, adapter adapters.NodeAdapter) {
// triggerChecks triggers a simulation step check whenever a peer is added or
// removed from the given node
func triggerChecks(trigger chan *adapters.NodeId, net *simulations.Network, id *adapters.NodeId) error {
func triggerChecks(trigger chan discover.NodeID, net *simulations.Network, id discover.NodeID) error {
node := net.GetNode(id)
if node == nil {
return fmt.Errorf("unknown node: %s", id)
@ -178,8 +179,8 @@ func triggerChecks(trigger chan *adapters.NodeId, net *simulations.Network, id *
return nil
}
func newService(id *adapters.NodeId, snapshot []byte) node.Service {
addr := network.NewAddrFromNodeId(id)
func newService(id discover.NodeID, snapshot []byte) node.Service {
addr := network.NewAddrFromNodeID(id)
kp := network.NewKadParams()
kp.MinProxBinSize = testMinProxBinSize

36
swarm/network/simulations/overlay.go
View file

@ -33,16 +33,16 @@ func NewSimulation() *Simulation {
}
}
func (s *Simulation) NewService(id *adapters.NodeId, snapshot []byte) node.Service {
func (s *Simulation) NewService(id discover.NodeID, snapshot []byte) node.Service {
s.mtx.Lock()
store, ok := s.stores[id.NodeID]
store, ok := s.stores[id]
if !ok {
store = adapters.NewSimStateStore()
s.stores[id.NodeID] = store
s.stores[id] = store
}
s.mtx.Unlock()
addr := network.NewAddrFromNodeId(id)
addr := network.NewAddrFromNodeID(id)
kp := network.NewKadParams()
kp.MinProxBinSize = 2
@ -71,33 +71,33 @@ func createMockers() map[string]*simulations.MockerConfig {
configs := make(map[string]*simulations.MockerConfig)
defaultCfg := simulations.DefaultMockerConfig()
defaultCfg.Id = "start-stop"
defaultCfg.ID = "start-stop"
defaultCfg.Description = "Starts and Stops nodes in go routines"
defaultCfg.Mocker = startStopMocker
bootNetworkCfg := simulations.DefaultMockerConfig()
bootNetworkCfg.Id = "bootNet"
bootNetworkCfg.ID = "bootNet"
bootNetworkCfg.Description = "Only boots up all nodes in the config"
bootNetworkCfg.Mocker = bootMocker
randomNodesCfg := simulations.DefaultMockerConfig()
randomNodesCfg.Id = "randomNodes"
randomNodesCfg.ID = "randomNodes"
randomNodesCfg.Description = "Boots nodes and then starts and stops some picking randomly"
randomNodesCfg.Mocker = randomMocker
configs[defaultCfg.Id] = defaultCfg
configs[bootNetworkCfg.Id] = bootNetworkCfg
configs[randomNodesCfg.Id] = randomNodesCfg
configs[defaultCfg.ID] = defaultCfg
configs[bootNetworkCfg.ID] = bootNetworkCfg
configs[randomNodesCfg.ID] = randomNodesCfg
return configs
}
func setupMocker(net *simulations.Network) []*adapters.NodeId {
func setupMocker(net *simulations.Network) []discover.NodeID {
conf := net.Config()
conf.DefaultService = "overlay"
nodeCount := 60
ids := make([]*adapters.NodeId, nodeCount)
ids := make([]discover.NodeID, nodeCount)
for i := 0; i < nodeCount; i++ {
node, err := net.NewNode()
if err != nil {
@ -113,16 +113,16 @@ func setupMocker(net *simulations.Network) []*adapters.NodeId {
log.Debug(fmt.Sprintf("node %v starting up", id))
}
for i, id := range ids {
var peerId *adapters.NodeId
var peerID discover.NodeID
if i == 0 {
peerId = ids[len(ids)-1]
peerID = ids[len(ids)-1]
} else {
peerId = ids[i-1]
peerID = ids[i-1]
}
ch := make(chan network.OverlayAddr)
go func() {
defer close(ch)
ch <- network.NewAddrFromNodeId(peerId)
ch <- network.NewAddrFromNodeID(peerID)
}()
if err := net.GetNode(id).Node.(*adapters.SimNode).Services()[0].(*network.Bzz).Hive.Register(ch); err != nil {
panic(err.Error())
@ -160,7 +160,7 @@ func randomMocker(net *simulations.Network) {
for i := lowid; i < highid; i++ {
log.Debug(fmt.Sprintf("node %v shutting down", ids[i]))
net.Stop(ids[i])
go func(id *adapters.NodeId) {
go func(id discover.NodeID) {
time.Sleep(time.Duration(randWait) * time.Millisecond)
net.Start(id)
}(ids[i])
@ -208,7 +208,7 @@ func main() {
config := &simulations.ServerConfig{
NewAdapter: func() adapters.NodeAdapter { return adapters.NewSimAdapter(services) },
DefaultMockerId: "bootNet",
DefaultMockerID: "bootNet",
Mockers: mockers,
}

180
swarm/pss/pss_test.go
View file

@ -14,6 +14,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
@ -177,7 +178,7 @@ func TestPssRegisterHandler(t *testing.T) {
func TestPssSimpleLinear(t *testing.T) {
var err error
nodeconfig := adapters.RandomNodeConfig()
addr := network.NewAddrFromNodeId(nodeconfig.Id)
addr := network.NewAddrFromNodeID(nodeconfig.ID)
_ = p2ptest.NewTestPeerPool()
ps := newTestPss(addr.Over())
@ -205,23 +206,23 @@ func TestPssSimpleLinear(t *testing.T) {
return bp.Run(ps.handlePssMsg)
}
pt := p2ptest.NewProtocolTester(t, nodeconfig.Id, 2, run)
pt := p2ptest.NewProtocolTester(t, nodeconfig.ID, 2, run)
msg := newPssPingMsg(ps, network.ToOverlayAddr(pt.Ids[0].Bytes()), pssPingProtocol, pssPingTopic, []byte{1, 2, 3})
msg := newPssPingMsg(ps, network.ToOverlayAddr(pt.IDs[0].Bytes()), pssPingProtocol, pssPingTopic, []byte{1, 2, 3})
exchange := p2ptest.Exchange{
Expects: []p2ptest.Expect{
p2ptest.Expect{
Code: 0,
Msg: msg,
Peer: pt.Ids[0],
Peer: pt.IDs[0],
},
},
Triggers: []p2ptest.Trigger{
p2ptest.Trigger{
Code: 0,
Msg: msg,
Peer: pt.Ids[1],
Peer: pt.IDs[1],
},
},
}
@ -238,16 +239,16 @@ func TestPssFullRandom10_5_5(t *testing.T) {
}
func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int, fullnodecount int, msgcount int) {
var lastid *adapters.NodeId = nil
var lastid discover.NodeID
nodeCount := 5
net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
Id: "0",
ID: "0",
})
defer net.Shutdown()
trigger := make(chan *adapters.NodeId)
ids := make([]*adapters.NodeId, nodeCount)
trigger := make(chan discover.NodeID)
ids := make([]discover.NodeID, nodeCount)
fullids := ids[0:fullnodecount]
fullpeers := [][]byte{}
@ -260,11 +261,11 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int
}
if err := net.Start(node.ID()); err != nil {
t.Fatalf("error starting node %s: %s", node.ID().Label(), err)
t.Fatalf("error starting node %s: %s", node.ID().TerminalString(), err)
}
if err := triggerChecks(trigger, net, node.ID()); err != nil {
t.Fatal("error triggering checks for node %s: %s", node.ID().Label(), err)
t.Fatal("error triggering checks for node %s: %s", node.ID().TerminalString(), err)
}
ids[i] = node.ID()
if i < fullnodecount {
@ -276,19 +277,19 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int
// for full peer discovery
action := func(ctx context.Context) error {
for i, id := range ids {
var peerId *adapters.NodeId
var peerID discover.NodeID
if i == 0 {
peerId = ids[len(ids)-1]
peerID = ids[len(ids)-1]
} else {
peerId = ids[i-1]
peerID = ids[i-1]
}
if err := net.Connect(id, peerId); err != nil {
if err := net.Connect(id, peerID); err != nil {
return err
}
}
return nil
}
check := func(ctx context.Context, id *adapters.NodeId) (bool, error) {
check := func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
return false, ctx.Err()
@ -341,12 +342,12 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int
t.Fatalf("simulation failed: %s", result.Error)
}
trigger = make(chan *adapters.NodeId)
trigger = make(chan discover.NodeID)
action = func(ctx context.Context) error {
return nil
}
check = func(ctx context.Context, id *adapters.NodeId) (bool, error) {
check = func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
return false, ctx.Err()
@ -379,9 +380,9 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int
// triggerChecks triggers a simulation step check whenever a peer is added or
// removed from the given node
func triggerChecks(trigger chan *adapters.NodeId, net *simulations.Network, id *adapters.NodeId) error {
func triggerChecks(trigger chan discover.NodeID, net *simulations.Network, id discover.NodeID) error {
gotpeer := make(map[*adapters.NodeId]bool)
gotpeer := make(map[discover.NodeID]bool)
node := net.GetNode(id)
if node == nil {
@ -410,10 +411,9 @@ func triggerChecks(trigger chan *adapters.NodeId, net *simulations.Network, id *
for {
select {
case event := <-peerevents:
nid := adapters.NewNodeId(event.Peer[:])
if event.Type == "add" && !gotpeer[nid] {
if event.Type == "add" && !gotpeer[event.Peer] {
trigger <- id
gotpeer[nid] = true
gotpeer[event.Peer] = true
}
case <-msgevents:
trigger <- id
@ -436,12 +436,12 @@ func triggerChecks(trigger chan *adapters.NodeId, net *simulations.Network, id *
func newServices() adapters.Services {
stateStore := adapters.NewSimStateStore()
kademlias := make(map[*adapters.NodeId]*network.Kademlia)
kademlia := func(id *adapters.NodeId) *network.Kademlia {
kademlias := make(map[discover.NodeID]*network.Kademlia)
kademlia := func(id discover.NodeID) *network.Kademlia {
if k, ok := kademlias[id]; ok {
return k
}
addr := network.NewAddrFromNodeId(id)
addr := network.NewAddrFromNodeID(id)
params := network.NewKadParams()
params.MinProxBinSize = 2
params.MaxBinSize = 3
@ -453,7 +453,7 @@ func newServices() adapters.Services {
return kademlias[id]
}
return adapters.Services{
"pss": func(id *adapters.NodeId, snapshot []byte) node.Service {
"pss": func(id discover.NodeID, snapshot []byte) node.Service {
cachedir, err := ioutil.TempDir("", "pss-cache")
if err != nil {
log.Error("create pss cache tmpdir failed", "error", err)
@ -479,8 +479,8 @@ func newServices() adapters.Services {
return ps
},
"bzz": func(id *adapters.NodeId, snapshot []byte) node.Service {
addr := network.NewAddrFromNodeId(id)
"bzz": func(id discover.NodeID, snapshot []byte) node.Service {
addr := network.NewAddrFromNodeID(id)
config := &network.BzzConfig{
OverlayAddr: addr.Over(),
UnderlayAddr: addr.Under(),
@ -509,9 +509,9 @@ type pssTestNode struct {
*Hive
*Pss
id *adapters.NodeId
id discover.NodeID
network *simulations.Network
trigger chan *adapters.NodeId
trigger chan discover.NodeID
run adapters.RunProtocol
ct *protocols.CodeMap
expectC chan []int
@ -558,7 +558,7 @@ func newPssTestService(t *testing.T, handlefunc func(interface{}) error, testnod
bzz := NewBzz(testnode.OverlayAddr(), testnode.UnderlayAddr(), newTestStore())
testnode.Hive = NewHive(hp, testnode.Pss.Overlay, bzz)
return &pssTestService{
//nid := adapters.NewNodeId(addr.UnderlayAddr())
//nid := adapters.NewNodeID(addr.UnderlayAddr())
msgFunc: handlefunc,
node: testnode,
}
@ -608,28 +608,28 @@ func (self *pssTestService) Run(peer *bzzPeer) error {
func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes int) {
var action func(ctx context.Context) error
var i int
var check func(ctx context.Context, id *adapters.NodeId) (bool, error)
var check func(ctx context.Context, id discover.NodeID) (bool, error)
var ctx context.Context
var result *simulations.StepResult
var timeout time.Duration
var cancel context.CancelFunc
fullnodes := []*adapters.NodeId{}
fullnodes := []discover.NodeID{}
sends := []int{} // sender/receiver ids array indices pairs
expectnodes := make(map[*adapters.NodeId]int) // how many messages we're expecting on each respective node
expectnodesids := []*adapters.NodeId{} // the nodes to expect on (needed by checker)
expectnodesresults := make(map[*adapters.NodeId][]int) // which messages expect actually got
expectnodes := make(map[discover.NodeID]int) // how many messages we're expecting on each respective node
expectnodesids := []discover.NodeID{} // the nodes to expect on (needed by checker)
expectnodesresults := make(map[discover.NodeID][]int) // which messages expect actually got
vct := protocols.NewCodeMap(map[uint64]interface{}{
0: pssTestPayload{},
})
topic, _ := MakeTopic(protocolName, protocolVersion)
trigger := make(chan *adapters.NodeId)
testpeers := make(map[*adapters.NodeId]*pssTestPeer)
trigger := make(chan discover.NodeID)
testpeers := make(map[discover.NodeID]*pssTestPeer)
net, nodes := newPssSimulationTester(t, numnodes, numfullnodes, trigger, vct, protocolName, protocolVersion, testpeers)
ids := []*adapters.NodeId{}
ids := []discover.NodeID{}
// connect the peers
action = func(ctx context.Context) error {
@ -641,17 +641,17 @@ func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes in
}
}
for i, id := range ids {
var peerId *adapters.NodeId
var peerID discover.NodeID
if i != 0 {
peerId = ids[i-1]
if err := net.Connect(id, peerId); err != nil {
peerID = ids[i-1]
if err := net.Connect(id, peerID); err != nil {
return err
}
}
}
return nil
}
check = func(ctx context.Context, id *adapters.NodeId) (bool, error) {
check = func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
return false, ctx.Err()
@ -690,7 +690,7 @@ func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes in
}
// ensure that the channel is clean
trigger = make(chan *adapters.NodeId)
trigger = make(chan discover.NodeID)
// randomly decide which nodes to send to and from
rand.Seed(time.Now().Unix())
@ -725,7 +725,7 @@ func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes in
msgbytes, _ := makeMsg(code, &pssTestPayload{
Data: fmt.Sprintf("%v", i+1),
})
go func(i int, expectnodesresults map[*adapters.NodeId][]int) {
go func(i int, expectnodesresults map[discover.NodeID][]int) {
expectnode := fullnodes[sends[i+1]] // the receiving node
sendnode := fullnodes[sends[i]] // the sending node
oaddr := nodes[expectnode].OverlayAddr()
@ -756,7 +756,7 @@ func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes in
}
// results
check = func(ctx context.Context, id *adapters.NodeId) (bool, error) {
check = func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
return false, ctx.Err()
@ -819,27 +819,27 @@ func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes in
func TestPssFullLinearEcho(t *testing.T) {
var action func(ctx context.Context) error
var check func(ctx context.Context, id *adapters.NodeId) (bool, error)
var check func(ctx context.Context, id discover.NodeID) (bool, error)
var ctx context.Context
var result *simulations.StepResult
var timeout time.Duration
var cancel context.CancelFunc
var firstpssnode *adapters.NodeId
var secondpssnode *adapters.NodeId
var firstpssnode discover.NodeID
var secondpssnode discover.NodeID
vct := protocols.NewCodeMap(protocolName, protocolVersion, ProtocolMaxMsgSize)
vct.Register(0, &pssTestPayload{})
topic, _ := MakeTopic(protocolName, protocolVersion)
fullnodes := []*adapters.NodeId{}
trigger := make(chan *adapters.NodeId)
testpeers := make(map[*adapters.NodeId]*pssTestPeer)
fullnodes := []discover.NodeID{}
trigger := make(chan discover.NodeID)
testpeers := make(map[discover.NodeID]*pssTestPeer)
net, nodes := newPssSimulationTester(t, 3, 2, trigger, vct, protocolName, protocolVersion, testpeers)
ids := []*adapters.NodeId{} // ohh risky! but the action for a specific id should come before the expect anyway
ids := []discover.NodeID{} // ohh risky! but the action for a specific id should come before the expect anyway
action = func(ctx context.Context) error {
var thinnodeid *adapters.NodeId
var thinnodeid discover.NodeID
for id, _ := range nodes {
ids = append(ids, id)
if _, ok := testpeers[id]; ok {
@ -857,17 +857,17 @@ func TestPssFullLinearEcho(t *testing.T) {
}
// for i, id := range ids {
// var peerId *adapters.NodeId
// var peerID discover.NodeID
// if i != 0 {
// peerId = ids[i-1]
// if err := net.Connect(id, peerId); err != nil {
// peerID = ids[i-1]
// if err := net.Connect(id, peerID); err != nil {
// return err
// }
// }
// }
return nil
}
check = func(ctx context.Context, id *adapters.NodeId) (bool, error) {
check = func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
return false, ctx.Err()
@ -899,7 +899,7 @@ func TestPssFullLinearEcho(t *testing.T) {
}
cancel()
nonode := &adapters.NodeId{}
nonode := &adapters.NodeID{}
firstpssnode = nonode
secondpssnode = nonode
@ -960,7 +960,7 @@ func TestPssFullLinearEcho(t *testing.T) {
return nil
}
check = func(ctx context.Context, id *adapters.NodeId) (bool, error) {
check = func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
return false, ctx.Err()
@ -981,7 +981,7 @@ func TestPssFullLinearEcho(t *testing.T) {
Action: action,
Trigger: trigger,
Expect: &simulations.Expectation{
Nodes: []*adapters.NodeId{ids[0]},
Nodes: []discover.NodeID{ids[0]},
Check: check,
},
})
@ -1001,25 +1001,25 @@ func TestPssFullWS(t *testing.T) {
var clientrecv, clientsend *rpc.Client
var action func(ctx context.Context) error
var check func(ctx context.Context, id *adapters.NodeId) (bool, error)
var check func(ctx context.Context, id discover.NodeID) (bool, error)
var ctx context.Context
var result *simulations.StepResult
var timeout time.Duration
var cancel context.CancelFunc
var firstpssnode, secondpssnode *adapters.NodeId
fullnodes := []*adapters.NodeId{}
var firstpssnode, secondpssnode discover.NodeID
fullnodes := []discover.NodeID{}
vct := protocols.NewCodeMap(protocolName, protocolVersion, ProtocolMaxMsgSize)
vct.Register(0, &pssTestPayload{})
topic, _ := MakeTopic(pingTopicName, pingTopicVersion)
trigger := make(chan *adapters.NodeId)
testpeers := make(map[*adapters.NodeId]*pssTestPeer)
trigger := make(chan discover.NodeID)
testpeers := make(map[discover.NodeID]*pssTestPeer)
simnet, nodes := newPssSimulationTester(t, 3, 2, trigger, vct, protocolName, protocolVersion, testpeers)
ids := []*adapters.NodeId{} // ohh risky! but the action for a specific id should come before the expect anyway
ids := []discover.NodeID{} // ohh risky! but the action for a specific id should come before the expect anyway
action = func(ctx context.Context) error {
var thinnodeid *adapters.NodeId
var thinnodeid discover.NodeID
for id, node := range nodes {
ids = append(ids, id)
if _, ok := testpeers[id]; ok {
@ -1046,7 +1046,7 @@ func TestPssFullWS(t *testing.T) {
return nil
}
check = func(ctx context.Context, id *adapters.NodeId) (bool, error) {
check = func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
return false, ctx.Err()
@ -1079,7 +1079,7 @@ func TestPssFullWS(t *testing.T) {
}
cancel()
nonode := &adapters.NodeId{}
nonode := &adapters.NodeID{}
firstpssnode = nonode
secondpssnode = nonode
@ -1167,7 +1167,7 @@ func TestPssFullWS(t *testing.T) {
}
}
trigger = make(chan *adapters.NodeId)
trigger = make(chan discover.NodeID)
ch := make(chan string)
action = func(ctx context.Context) error {
@ -1178,7 +1178,7 @@ func TestPssFullWS(t *testing.T) {
}()
return nil
}
check = func(ctx context.Context, id *adapters.NodeId) (bool, error) {
check = func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
return false, ctx.Err()
@ -1204,7 +1204,7 @@ func TestPssFullWS(t *testing.T) {
Action: action,
Trigger: trigger,
Expect: &simulations.Expectation{
Nodes: []*adapters.NodeId{secondpssnode},
Nodes: []discover.NodeID{secondpssnode},
Check: check,
},
})
@ -1228,12 +1228,12 @@ func TestPssFullWS(t *testing.T) {
// the simulation tester constructor is currently a hack to fit previous code with later stack using node.Services to start SimNodes
func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, trigger chan *adapters.NodeId, vct *protocols.CodeMap, name string, version int, testpeers map[*adapters.NodeId]*pssTestPeer) (*simulations.Network, map[*adapters.NodeId]*pssTestNode) {
func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, trigger chan discover.NodeID, vct *protocols.CodeMap, name string, version int, testpeers map[discover.NodeID]*pssTestPeer) (*simulations.Network, map[discover.NodeID]*pssTestNode) {
topic, _ := MakeTopic(name, version)
nodes := make(map[*adapters.NodeId]*pssTestNode, numnodes)
psss := make(map[*adapters.NodeId]*Pss)
nodes := make(map[discover.NodeID]*pssTestNode, numnodes)
psss := make(map[discover.NodeID]*Pss)
var simnet *simulations.Network
serviceFunc := func(id *adapters.NodeId) node.Service {
serviceFunc := func(id discover.NodeID) node.Service {
node := &pssTestNode{
Pss: psss[id],
Hive: nil,
@ -1249,7 +1249,7 @@ func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, trigge
var handlefunc func(interface{}) error
addr := NewPeerAddrFromNodeId(id)
addr := NewPeerAddrFromNodeID(id)
if testpeers[id] != nil {
handlefunc = makePssHandleProtocol(psss[id])
@ -1273,7 +1273,7 @@ func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, trigge
}
adapter := adapters.NewSimAdapter(map[string]adapters.ServiceFunc{"pss": serviceFunc})
simnet = simulations.NewNetwork(adapter, &simulations.NetworkConfig{
Id: "0",
ID: "0",
Backend: true,
})
configs := make([]*adapters.NodeConfig, numnodes)
@ -1282,8 +1282,8 @@ func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, trigge
configs[i].Service = "pss"
}
for i, conf := range configs {
addr := NewPeerAddrFromNodeId(conf.Id)
psss[conf.Id] = makePss(addr.Over())
addr := NewPeerAddrFromNodeID(conf.ID)
psss[conf.ID] = makePss(addr.Over())
if i < numfullnodes {
tp := &pssTestPeer{
Peer: &protocols.Peer{
@ -1292,17 +1292,17 @@ func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, trigge
successC: make(chan bool),
resultC: make(chan int),
}
testpeers[conf.Id] = tp
targetprotocol := makeCustomProtocol(name, version, vct, testpeers[conf.Id])
pssprotocol := NewPssProtocol(psss[conf.Id], &topic, vct, targetprotocol)
psss[conf.Id].Register(topic, pssprotocol.GetHandler())
testpeers[conf.ID] = tp
targetprotocol := makeCustomProtocol(name, version, vct, testpeers[conf.ID])
pssprotocol := NewPssProtocol(psss[conf.ID], &topic, vct, targetprotocol)
psss[conf.ID].Register(topic, pssprotocol.GetHandler())
}
if err := simnet.NewNodeWithConfig(conf); err != nil {
t.Fatalf("error creating node %s: %s", conf.Id.Label(), err)
t.Fatalf("error creating node %s: %s", conf.ID.Label(), err)
}
if err := simnet.Start(conf.Id); err != nil {
t.Fatalf("error starting node %s: %s", conf.Id.Label(), err)
if err := simnet.Start(conf.ID); err != nil {
t.Fatalf("error starting node %s: %s", conf.ID.Label(), err)
}
}
@ -1422,7 +1422,7 @@ func makePssHandleProtocol(ps *Pss) func(msg interface{}) error {
if f == nil {
return fmt.Errorf("No registered handler for topic '%s'", env.Topic)
}
nid := adapters.NewNodeId(env.SenderUAddr)
nid := adapters.NewNodeID(env.SenderUAddr)
p := p2p.NewPeer(nid.NodeID, fmt.Sprintf("%x", common.ByteLabel(nid.Bytes())), []p2p.Cap{})
return f(umsg, p, env.SenderOAddr)
} else {

3
swarm/swarm.go
View file

@ -38,7 +38,6 @@ import (
"github.com/ethereum/go-ethereum/swarm/fuse"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/pss"
)
// the swarm stack
@ -257,7 +256,7 @@ func (self *Swarm) Protocols() []p2p.Protocol {
// if f == nil {
// return fmt.Errorf("No registered handler for topic '%s'", env.Topic)
// }
// nid := adapters.NewNodeId(env.SenderUAddr)
// nid := discover.MustBytesID(env.SenderUAddr)
// p := p2p.NewPeer(nid.NodeID, fmt.Sprintf("%x", common.ByteLabel(nid.Bytes())), []p2p.Cap{})
// return f(umsg, p, env.SenderOAddr)
// } else {