mirror
/
go-ethereum
mirror of https://github.com/ethereum/go-ethereum
2
0
Fork 1
Code Issues Releases Wiki Activity

p2p: fix some golint warnings (#16577)

Browse Source
pull/16709/head
kiel barry 7 years ago committed by Felix Lange
parent a42be3b78d
commit 864e80a48f
13 changed files with 292 additions and 291 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 20
      p2p/discover/table_test.go
  2. 20
      p2p/discv5/net_test.go
  3. 8
      p2p/discv5/node.go
  4. 4
      p2p/discv5/ticket.go
  5. 16
      p2p/discv5/topic.go
  6. 24
      p2p/message.go
  7. 4
      p2p/peer_error.go
  8. 98
      p2p/simulations/adapters/inproc.go
  9. 8
      p2p/simulations/adapters/state.go
  10. 309
      p2p/simulations/network.go
  11. 34
      p2p/testing/peerpool.go
  12. 26
      p2p/testing/protocolsession.go
  13. 12
      p2p/testing/protocoltester.go

20
p2p/discover/table_test.go
Unescape View File

@ -582,26 +582,26 @@ func (*preminedTestnet) ping(toid NodeID, toaddr *net.UDPAddr) error { return ni
// mine generates a testnet struct literal with nodes at // mine generates a testnet struct literal with nodes at
// various distances to the given target. // various distances to the given target.
func (n *preminedTestnet) mine(target NodeID) { func (tn *preminedTestnet) mine(target NodeID) {
n.target = target tn.target = target
n.targetSha = crypto.Keccak256Hash(n.target[:]) tn.targetSha = crypto.Keccak256Hash(tn.target[:])
found := 0 found := 0
for found < bucketSize*10 { for found < bucketSize*10 {
k := newkey() k := newkey()
id := PubkeyID(&k.PublicKey) id := PubkeyID(&k.PublicKey)
sha := crypto.Keccak256Hash(id[:]) sha := crypto.Keccak256Hash(id[:])
ld := logdist(n.targetSha, sha) ld := logdist(tn.targetSha, sha)
if len(n.dists[ld]) < bucketSize { if len(tn.dists[ld]) < bucketSize {
n.dists[ld] = append(n.dists[ld], id) tn.dists[ld] = append(tn.dists[ld], id)
fmt.Println("found ID with ld", ld) fmt.Println("found ID with ld", ld)
found++ found++
} }
} }
fmt.Println("&preminedTestnet{") fmt.Println("&preminedTestnet{")
fmt.Printf(" target: %#v,\n", n.target) fmt.Printf(" target: %#v,\n", tn.target)
fmt.Printf(" targetSha: %#v,\n", n.targetSha) fmt.Printf(" targetSha: %#v,\n", tn.targetSha)
fmt.Printf(" dists: [%d][]NodeID{\n", len(n.dists)) fmt.Printf(" dists: [%d][]NodeID{\n", len(tn.dists))
for ld, ns := range n.dists { for ld, ns := range tn.dists {
if len(ns) == 0 { if len(ns) == 0 {
continue continue
} }

20
p2p/discv5/net_test.go
Unescape View File

@ -336,26 +336,26 @@ func (*preminedTestnet) localAddr() *net.UDPAddr {
// mine generates a testnet struct literal with nodes at // mine generates a testnet struct literal with nodes at
// various distances to the given target. // various distances to the given target.
func (n *preminedTestnet) mine(target NodeID) { func (tn *preminedTestnet) mine(target NodeID) {
n.target = target tn.target = target
n.targetSha = crypto.Keccak256Hash(n.target[:]) tn.targetSha = crypto.Keccak256Hash(tn.target[:])
found := 0 found := 0
for found < bucketSize*10 { for found < bucketSize*10 {
k := newkey() k := newkey()
id := PubkeyID(&k.PublicKey) id := PubkeyID(&k.PublicKey)
sha := crypto.Keccak256Hash(id[:]) sha := crypto.Keccak256Hash(id[:])
ld := logdist(n.targetSha, sha) ld := logdist(tn.targetSha, sha)
if len(n.dists[ld]) < bucketSize { if len(tn.dists[ld]) < bucketSize {
n.dists[ld] = append(n.dists[ld], id) tn.dists[ld] = append(tn.dists[ld], id)
fmt.Println("found ID with ld", ld) fmt.Println("found ID with ld", ld)
found++ found++
} }
} }
fmt.Println("&preminedTestnet{") fmt.Println("&preminedTestnet{")
fmt.Printf(" target: %#v,\n", n.target) fmt.Printf(" target: %#v,\n", tn.target)
fmt.Printf(" targetSha: %#v,\n", n.targetSha) fmt.Printf(" targetSha: %#v,\n", tn.targetSha)
fmt.Printf(" dists: [%d][]NodeID{\n", len(n.dists)) fmt.Printf(" dists: [%d][]NodeID{\n", len(tn.dists))
for ld, ns := range n.dists { for ld, ns := range tn.dists {
if len(ns) == 0 { if len(ns) == 0 {
continue continue
} }

8
p2p/discv5/node.go
Unescape View File

@ -315,11 +315,11 @@ func PubkeyID(pub *ecdsa.PublicKey) NodeID {
// Pubkey returns the public key represented by the node ID. // Pubkey returns the public key represented by the node ID.
// It returns an error if the ID is not a point on the curve. // It returns an error if the ID is not a point on the curve.
func (id NodeID) Pubkey() (*ecdsa.PublicKey, error) { func (n NodeID) Pubkey() (*ecdsa.PublicKey, error) {
p := &ecdsa.PublicKey{Curve: crypto.S256(), X: new(big.Int), Y: new(big.Int)} p := &ecdsa.PublicKey{Curve: crypto.S256(), X: new(big.Int), Y: new(big.Int)}
half := len(id) / 2 half := len(n) / 2
p.X.SetBytes(id[:half]) p.X.SetBytes(n[:half])
p.Y.SetBytes(id[half:]) p.Y.SetBytes(n[half:])
if !p.Curve.IsOnCurve(p.X, p.Y) { if !p.Curve.IsOnCurve(p.X, p.Y) {
return nil, errors.New("id is invalid secp256k1 curve point") return nil, errors.New("id is invalid secp256k1 curve point")
} }

4
p2p/discv5/ticket.go
Unescape View File

@ -304,8 +304,8 @@ func (s ticketRefByWaitTime) Len() int {
return len(s) return len(s)
} }
func (r ticketRef) waitTime() mclock.AbsTime { func (ref ticketRef) waitTime() mclock.AbsTime {
return r.t.regTime[r.idx] - r.t.issueTime return ref.t.regTime[ref.idx] - ref.t.issueTime
} }
// Less reports whether the element with // Less reports whether the element with

16
p2p/discv5/topic.go
Unescape View File

@ -271,15 +271,15 @@ func (t *topicTable) useTicket(node *Node, serialNo uint32, topics []Topic, idx
return false return false
} }
func (topictab *topicTable) getTicket(node *Node, topics []Topic) *ticket { func (t *topicTable) getTicket(node *Node, topics []Topic) *ticket {
topictab.collectGarbage() t.collectGarbage()
now := mclock.Now() now := mclock.Now()
n := topictab.getOrNewNode(node) n := t.getOrNewNode(node)
n.lastIssuedTicket++ n.lastIssuedTicket++
topictab.storeTicketCounters(node) t.storeTicketCounters(node)
t := &ticket{ tic := &ticket{
issueTime: now, issueTime: now,
topics: topics, topics: topics,
serial: n.lastIssuedTicket, serial: n.lastIssuedTicket,
@ -287,15 +287,15 @@ func (topictab *topicTable) getTicket(node *Node, topics []Topic) *ticket {
} }
for i, topic := range topics { for i, topic := range topics {
var waitPeriod time.Duration var waitPeriod time.Duration
if topic := topictab.topics[topic]; topic != nil { if topic := t.topics[topic]; topic != nil {
waitPeriod = topic.wcl.waitPeriod waitPeriod = topic.wcl.waitPeriod
} else { } else {
waitPeriod = minWaitPeriod waitPeriod = minWaitPeriod
} }
t.regTime[i] = now + mclock.AbsTime(waitPeriod) tic.regTime[i] = now + mclock.AbsTime(waitPeriod)
} }
return t return tic
} }
const gcInterval = time.Minute const gcInterval = time.Minute

24
p2p/message.go
Unescape View File

@ -270,15 +270,15 @@ func newMsgEventer(rw MsgReadWriter, feed *event.Feed, peerID discover.NodeID, p
// ReadMsg reads a message from the underlying MsgReadWriter and emits a // ReadMsg reads a message from the underlying MsgReadWriter and emits a
// "message received" event // "message received" event
func (self *msgEventer) ReadMsg() (Msg, error) { func (ev *msgEventer) ReadMsg() (Msg, error) {
msg, err := self.MsgReadWriter.ReadMsg() msg, err := ev.MsgReadWriter.ReadMsg()
if err != nil { if err != nil {
return msg, err return msg, err
} }
self.feed.Send(&PeerEvent{ ev.feed.Send(&PeerEvent{
Type: PeerEventTypeMsgRecv, Type: PeerEventTypeMsgRecv,
Peer: self.peerID, Peer: ev.peerID,
Protocol: self.Protocol, Protocol: ev.Protocol,
MsgCode: &msg.Code, MsgCode: &msg.Code,
MsgSize: &msg.Size, MsgSize: &msg.Size,
}) })
@ -287,15 +287,15 @@ func (self *msgEventer) ReadMsg() (Msg, error) {
// WriteMsg writes a message to the underlying MsgReadWriter and emits a // WriteMsg writes a message to the underlying MsgReadWriter and emits a
// "message sent" event // "message sent" event
func (self *msgEventer) WriteMsg(msg Msg) error { func (ev *msgEventer) WriteMsg(msg Msg) error {
err := self.MsgReadWriter.WriteMsg(msg) err := ev.MsgReadWriter.WriteMsg(msg)
if err != nil { if err != nil {
return err return err
} }
self.feed.Send(&PeerEvent{ ev.feed.Send(&PeerEvent{
Type: PeerEventTypeMsgSend, Type: PeerEventTypeMsgSend,
Peer: self.peerID, Peer: ev.peerID,
Protocol: self.Protocol, Protocol: ev.Protocol,
MsgCode: &msg.Code, MsgCode: &msg.Code,
MsgSize: &msg.Size, MsgSize: &msg.Size,
}) })
@ -304,8 +304,8 @@ func (self *msgEventer) WriteMsg(msg Msg) error {
// Close closes the underlying MsgReadWriter if it implements the io.Closer // Close closes the underlying MsgReadWriter if it implements the io.Closer
// interface // interface
func (self *msgEventer) Close() error { func (ev *msgEventer) Close() error {
if v, ok := self.MsgReadWriter.(io.Closer); ok { if v, ok := ev.MsgReadWriter.(io.Closer); ok {
return v.Close() return v.Close()
} }
return nil return nil

4
p2p/peer_error.go
Unescape View File

@ -48,8 +48,8 @@ func newPeerError(code int, format string, v ...interface{}) *peerError {
return err return err
} }
func (self *peerError) Error() string { func (pe *peerError) Error() string {
return self.message return pe.message
} }
var errProtocolReturned = errors.New("protocol returned") var errProtocolReturned = errors.New("protocol returned")

98
p2p/simulations/adapters/inproc.go
Unescape View File

@ -154,30 +154,30 @@ type SimNode struct {
} }
// Addr returns the node's discovery address // Addr returns the node's discovery address
func (self *SimNode) Addr() []byte { func (sn *SimNode) Addr() []byte {
return []byte(self.Node().String()) return []byte(sn.Node().String())
} }
// Node returns a discover.Node representing the SimNode // Node returns a discover.Node representing the SimNode
func (self *SimNode) Node() *discover.Node { func (sn *SimNode) Node() *discover.Node {
return discover.NewNode(self.ID, net.IP{127, 0, 0, 1}, 30303, 30303) return discover.NewNode(sn.ID, net.IP{127, 0, 0, 1}, 30303, 30303)
} }
// Client returns an rpc.Client which can be used to communicate with the // Client returns an rpc.Client which can be used to communicate with the
// underlying services (it is set once the node has started) // underlying services (it is set once the node has started)
func (self *SimNode) Client() (*rpc.Client, error) { func (sn *SimNode) Client() (*rpc.Client, error) {
self.lock.RLock() sn.lock.RLock()
defer self.lock.RUnlock() defer sn.lock.RUnlock()
if self.client == nil { if sn.client == nil {
return nil, errors.New("node not started") return nil, errors.New("node not started")
} }
return self.client, nil return sn.client, nil
} }
// ServeRPC serves RPC requests over the given connection by creating an // ServeRPC serves RPC requests over the given connection by creating an
// in-memory client to the node's RPC server // in-memory client to the node's RPC server
func (self *SimNode) ServeRPC(conn net.Conn) error { func (sn *SimNode) ServeRPC(conn net.Conn) error {
handler, err := self.node.RPCHandler() handler, err := sn.node.RPCHandler()
if err != nil { if err != nil {
return err return err
} }
@ -187,13 +187,13 @@ func (self *SimNode) ServeRPC(conn net.Conn) error {
// Snapshots creates snapshots of the services by calling the // Snapshots creates snapshots of the services by calling the
// simulation_snapshot RPC method // simulation_snapshot RPC method
func (self *SimNode) Snapshots() (map[string][]byte, error) { func (sn *SimNode) Snapshots() (map[string][]byte, error) {
self.lock.RLock() sn.lock.RLock()
services := make(map[string]node.Service, len(self.running)) services := make(map[string]node.Service, len(sn.running))
for name, service := range self.running { for name, service := range sn.running {
services[name] = service services[name] = service
} }
self.lock.RUnlock() sn.lock.RUnlock()
if len(services) == 0 { if len(services) == 0 {
return nil, errors.New("no running services") return nil, errors.New("no running services")
} }
@ -213,23 +213,23 @@ func (self *SimNode) Snapshots() (map[string][]byte, error) {
} }
// Start registers the services and starts the underlying devp2p node // Start registers the services and starts the underlying devp2p node
func (self *SimNode) Start(snapshots map[string][]byte) error { func (sn *SimNode) Start(snapshots map[string][]byte) error {
newService := func(name string) func(ctx *node.ServiceContext) (node.Service, error) { newService := func(name string) func(ctx *node.ServiceContext) (node.Service, error) {
return func(nodeCtx *node.ServiceContext) (node.Service, error) { return func(nodeCtx *node.ServiceContext) (node.Service, error) {
ctx := &ServiceContext{ ctx := &ServiceContext{
RPCDialer: self.adapter, RPCDialer: sn.adapter,
NodeContext: nodeCtx, NodeContext: nodeCtx,
Config: self.config, Config: sn.config,
} }
if snapshots != nil { if snapshots != nil {
ctx.Snapshot = snapshots[name] ctx.Snapshot = snapshots[name]
} }
serviceFunc := self.adapter.services[name] serviceFunc := sn.adapter.services[name]
service, err := serviceFunc(ctx) service, err := serviceFunc(ctx)
if err != nil { if err != nil {
return nil, err return nil, err
} }
self.running[name] = service sn.running[name] = service
return service, nil return service, nil
} }
} }
@ -237,9 +237,9 @@ func (self *SimNode) Start(snapshots map[string][]byte) error {
// ensure we only register the services once in the case of the node // ensure we only register the services once in the case of the node
// being stopped and then started again // being stopped and then started again
var regErr error var regErr error
self.registerOnce.Do(func() { sn.registerOnce.Do(func() {
for _, name := range self.config.Services { for _, name := range sn.config.Services {
if err := self.node.Register(newService(name)); err != nil { if err := sn.node.Register(newService(name)); err != nil {
regErr = err regErr = err
return return
} }
@ -249,54 +249,54 @@ func (self *SimNode) Start(snapshots map[string][]byte) error {
return regErr return regErr
} }
if err := self.node.Start(); err != nil { if err := sn.node.Start(); err != nil {
return err return err
} }
// create an in-process RPC client // create an in-process RPC client
handler, err := self.node.RPCHandler() handler, err := sn.node.RPCHandler()
if err != nil { if err != nil {
return err return err
} }
self.lock.Lock() sn.lock.Lock()
self.client = rpc.DialInProc(handler) sn.client = rpc.DialInProc(handler)
self.lock.Unlock() sn.lock.Unlock()
return nil return nil
} }
// Stop closes the RPC client and stops the underlying devp2p node // Stop closes the RPC client and stops the underlying devp2p node
func (self *SimNode) Stop() error { func (sn *SimNode) Stop() error {
self.lock.Lock() sn.lock.Lock()
if self.client != nil { if sn.client != nil {
self.client.Close() sn.client.Close()
self.client = nil sn.client = nil
} }
self.lock.Unlock() sn.lock.Unlock()
return self.node.Stop() return sn.node.Stop()
} }
// Services returns a copy of the underlying services // Services returns a copy of the underlying services
func (self *SimNode) Services() []node.Service { func (sn *SimNode) Services() []node.Service {
self.lock.RLock() sn.lock.RLock()
defer self.lock.RUnlock() defer sn.lock.RUnlock()
services := make([]node.Service, 0, len(self.running)) services := make([]node.Service, 0, len(sn.running))
for _, service := range self.running { for _, service := range sn.running {
services = append(services, service) services = append(services, service)
} }
return services return services
} }
// Server returns the underlying p2p.Server // Server returns the underlying p2p.Server
func (self *SimNode) Server() *p2p.Server { func (sn *SimNode) Server() *p2p.Server {
return self.node.Server() return sn.node.Server()
} }
// SubscribeEvents subscribes the given channel to peer events from the // SubscribeEvents subscribes the given channel to peer events from the
// underlying p2p.Server // underlying p2p.Server
func (self *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription { func (sn *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription {
srv := self.Server() srv := sn.Server()
if srv == nil { if srv == nil {
panic("node not running") panic("node not running")
} }
@ -304,12 +304,12 @@ func (self *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription
} }
// NodeInfo returns information about the node // NodeInfo returns information about the node
func (self *SimNode) NodeInfo() *p2p.NodeInfo { func (sn *SimNode) NodeInfo() *p2p.NodeInfo {
server := self.Server() server := sn.Server()
if server == nil { if server == nil {
return &p2p.NodeInfo{ return &p2p.NodeInfo{
ID: self.ID.String(), ID: sn.ID.String(),
Enode: self.Node().String(), Enode: sn.Node().String(),
} }
} }
return server.NodeInfo() return server.NodeInfo()

8
p2p/simulations/adapters/state.go
Unescape View File

@ -20,12 +20,12 @@ type SimStateStore struct {
m map[string][]byte m map[string][]byte
} }
func (self *SimStateStore) Load(s string) ([]byte, error) { func (st *SimStateStore) Load(s string) ([]byte, error) {
return self.m[s], nil return st.m[s], nil
} }
func (self *SimStateStore) Save(s string, data []byte) error { func (st *SimStateStore) Save(s string, data []byte) error {
self.m[s] = data st.m[s] = data
return nil return nil
} }

309
p2p/simulations/network.go
Unescape View File

@ -74,22 +74,22 @@ func NewNetwork(nodeAdapter adapters.NodeAdapter, conf *NetworkConfig) *Network
} }
// Events returns the output event feed of the Network. // Events returns the output event feed of the Network.
func (self *Network) Events() *event.Feed { func (net *Network) Events() *event.Feed {
return &self.events return &net.events
} }
// NewNode adds a new node to the network with a random ID // NewNode adds a new node to the network with a random ID
func (self *Network) NewNode() (*Node, error) { func (net *Network) NewNode() (*Node, error) {
conf := adapters.RandomNodeConfig() conf := adapters.RandomNodeConfig()
conf.Services = []string{self.DefaultService} conf.Services = []string{net.DefaultService}
return self.NewNodeWithConfig(conf) return net.NewNodeWithConfig(conf)
} }
// NewNodeWithConfig adds a new node to the network with the given config, // NewNodeWithConfig adds a new node to the network with the given config,
// returning an error if a node with the same ID or name already exists // returning an error if a node with the same ID or name already exists
func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error) { func (net *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error) {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
// create a random ID and PrivateKey if not set // create a random ID and PrivateKey if not set
if conf.ID == (discover.NodeID{}) { if conf.ID == (discover.NodeID{}) {
@ -100,31 +100,31 @@ func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error)
id := conf.ID id := conf.ID
if conf.Reachable == nil { if conf.Reachable == nil {
conf.Reachable = func(otherID discover.NodeID) bool { conf.Reachable = func(otherID discover.NodeID) bool {
_, err := self.InitConn(conf.ID, otherID) _, err := net.InitConn(conf.ID, otherID)
return err == nil return err == nil
} }
} }
// assign a name to the node if not set // assign a name to the node if not set
if conf.Name == "" { if conf.Name == "" {
conf.Name = fmt.Sprintf("node%02d", len(self.Nodes)+1) conf.Name = fmt.Sprintf("node%02d", len(net.Nodes)+1)
} }
// check the node doesn't already exist // check the node doesn't already exist
if node := self.getNode(id); node != nil { if node := net.getNode(id); node != nil {
return nil, fmt.Errorf("node with ID %q already exists", id) return nil, fmt.Errorf("node with ID %q already exists", id)
} }
if node := self.getNodeByName(conf.Name); node != nil { if node := net.getNodeByName(conf.Name); node != nil {
return nil, fmt.Errorf("node with name %q already exists", conf.Name) return nil, fmt.Errorf("node with name %q already exists", conf.Name)
} }
// if no services are configured, use the default service // if no services are configured, use the default service
if len(conf.Services) == 0 { if len(conf.Services) == 0 {
conf.Services = []string{self.DefaultService} conf.Services = []string{net.DefaultService}
} }
// use the NodeAdapter to create the node // use the NodeAdapter to create the node
adapterNode, err := self.nodeAdapter.NewNode(conf) adapterNode, err := net.nodeAdapter.NewNode(conf)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -133,27 +133,27 @@ func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error)
Config: conf, Config: conf,
} }
log.Trace(fmt.Sprintf("node %v created", id)) log.Trace(fmt.Sprintf("node %v created", id))
self.nodeMap[id] = len(self.Nodes) net.nodeMap[id] = len(net.Nodes)
self.Nodes = append(self.Nodes, node) net.Nodes = append(net.Nodes, node)
// emit a "control" event // emit a "control" event
self.events.Send(ControlEvent(node)) net.events.Send(ControlEvent(node))
return node, nil return node, nil
} }
// Config returns the network configuration // Config returns the network configuration
func (self *Network) Config() *NetworkConfig { func (net *Network) Config() *NetworkConfig {
return &self.NetworkConfig return &net.NetworkConfig
} }
// StartAll starts all nodes in the network // StartAll starts all nodes in the network
func (self *Network) StartAll() error { func (net *Network) StartAll() error {
for _, node := range self.Nodes { for _, node := range net.Nodes {
if node.Up { if node.Up {
continue continue
} }
if err := self.Start(node.ID()); err != nil { if err := net.Start(node.ID()); err != nil {
return err return err
} }
} }
@ -161,12 +161,12 @@ func (self *Network) StartAll() error {
} }
// StopAll stops all nodes in the network // StopAll stops all nodes in the network
func (self *Network) StopAll() error { func (net *Network) StopAll() error {
for _, node := range self.Nodes { for _, node := range net.Nodes {
if !node.Up { if !node.Up {
continue continue
} }
if err := self.Stop(node.ID()); err != nil { if err := net.Stop(node.ID()); err != nil {
return err return err
} }
} }
@ -174,21 +174,21 @@ func (self *Network) StopAll() error {
} }
// Start starts the node with the given ID // Start starts the node with the given ID
func (self *Network) Start(id discover.NodeID) error { func (net *Network) Start(id discover.NodeID) error {
return self.startWithSnapshots(id, nil) return net.startWithSnapshots(id, nil)
} }
// startWithSnapshots starts the node with the given ID using the give // startWithSnapshots starts the node with the given ID using the give
// snapshots // snapshots
func (self *Network) startWithSnapshots(id discover.NodeID, snapshots map[string][]byte) error { func (net *Network) startWithSnapshots(id discover.NodeID, snapshots map[string][]byte) error {
node := self.GetNode(id) node := net.GetNode(id)
if node == nil { if node == nil {
return fmt.Errorf("node %v does not exist", id) return fmt.Errorf("node %v does not exist", id)
} }
if node.Up { if node.Up {
return fmt.Errorf("node %v already up", id) return fmt.Errorf("node %v already up", id)
} }
log.Trace(fmt.Sprintf("starting node %v: %v using %v", id, node.Up, self.nodeAdapter.Name())) log.Trace(fmt.Sprintf("starting node %v: %v using %v", id, node.Up, net.nodeAdapter.Name()))
if err := node.Start(snapshots); err != nil { if err := node.Start(snapshots); err != nil {
log.Warn(fmt.Sprintf("start up failed: %v", err)) log.Warn(fmt.Sprintf("start up failed: %v", err))
return err return err
@ -196,7 +196,7 @@ func (self *Network) startWithSnapshots(id discover.NodeID, snapshots map[string
node.Up = true node.Up = true
log.Info(fmt.Sprintf("started node %v: %v", id, node.Up)) log.Info(fmt.Sprintf("started node %v: %v", id, node.Up))
self.events.Send(NewEvent(node)) net.events.Send(NewEvent(node))
// subscribe to peer events // subscribe to peer events
client, err := node.Client() client, err := node.Client()
@ -208,22 +208,22 @@ func (self *Network) startWithSnapshots(id discover.NodeID, snapshots map[string
if err != nil { if err != nil {
return fmt.Errorf("error getting peer events for node %v: %s", id, err) return fmt.Errorf("error getting peer events for node %v: %s", id, err)
} }
go self.watchPeerEvents(id, events, sub) go net.watchPeerEvents(id, events, sub)
return nil return nil
} }
// watchPeerEvents reads peer events from the given channel and emits // watchPeerEvents reads peer events from the given channel and emits
// corresponding network events // corresponding network events
func (self *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEvent, sub event.Subscription) { func (net *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEvent, sub event.Subscription) {
defer func() { defer func() {
sub.Unsubscribe() sub.Unsubscribe()
// assume the node is now down // assume the node is now down
self.lock.Lock() net.lock.Lock()
node := self.getNode(id) node := net.getNode(id)
node.Up = false node.Up = false
self.lock.Unlock() net.lock.Unlock()
self.events.Send(NewEvent(node)) net.events.Send(NewEvent(node))
}() }()
for { for {
select { select {
@ -235,16 +235,16 @@ func (self *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEv
switch event.Type { switch event.Type {
case p2p.PeerEventTypeAdd: case p2p.PeerEventTypeAdd:
self.DidConnect(id, peer) net.DidConnect(id, peer)
case p2p.PeerEventTypeDrop: case p2p.PeerEventTypeDrop:
self.DidDisconnect(id, peer) net.DidDisconnect(id, peer)
case p2p.PeerEventTypeMsgSend: case p2p.PeerEventTypeMsgSend:
self.DidSend(id, peer, event.Protocol, *event.MsgCode) net.DidSend(id, peer, event.Protocol, *event.MsgCode)
case p2p.PeerEventTypeMsgRecv: case p2p.PeerEventTypeMsgRecv:
self.DidReceive(peer, id, event.Protocol, *event.MsgCode) net.DidReceive(peer, id, event.Protocol, *event.MsgCode)
} }
@ -258,8 +258,8 @@ func (self *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEv
} }
// Stop stops the node with the given ID // Stop stops the node with the given ID
func (self *Network) Stop(id discover.NodeID) error { func (net *Network) Stop(id discover.NodeID) error {
node := self.GetNode(id) node := net.GetNode(id)
if node == nil { if node == nil {
return fmt.Errorf("node %v does not exist", id) return fmt.Errorf("node %v does not exist", id)
} }
@ -272,15 +272,15 @@ func (self *Network) Stop(id discover.NodeID) error {
node.Up = false node.Up = false
log.Info(fmt.Sprintf("stop node %v: %v", id, node.Up)) log.Info(fmt.Sprintf("stop node %v: %v", id, node.Up))
self.events.Send(ControlEvent(node)) net.events.Send(ControlEvent(node))
return nil return nil
} }
// Connect connects two nodes together by calling the "admin_addPeer" RPC // Connect connects two nodes together by calling the "admin_addPeer" RPC
// method on the "one" node so that it connects to the "other" node // method on the "one" node so that it connects to the "other" node
func (self *Network) Connect(oneID, otherID discover.NodeID) error { func (net *Network) Connect(oneID, otherID discover.NodeID) error {
log.Debug(fmt.Sprintf("connecting %s to %s", oneID, otherID)) log.Debug(fmt.Sprintf("connecting %s to %s", oneID, otherID))
conn, err := self.InitConn(oneID, otherID) conn, err := net.InitConn(oneID, otherID)
if err != nil { if err != nil {
return err return err
} }
@ -288,14 +288,14 @@ func (self *Network) Connect(oneID, otherID discover.NodeID) error {
if err != nil { if err != nil {
return err return err
} }
self.events.Send(ControlEvent(conn)) net.events.Send(ControlEvent(conn))
return client.Call(nil, "admin_addPeer", string(conn.other.Addr())) return client.Call(nil, "admin_addPeer", string(conn.other.Addr()))
} }
// Disconnect disconnects two nodes by calling the "admin_removePeer" RPC // Disconnect disconnects two nodes by calling the "admin_removePeer" RPC
// method on the "one" node so that it disconnects from the "other" node // method on the "one" node so that it disconnects from the "other" node
func (self *Network) Disconnect(oneID, otherID discover.NodeID) error { func (net *Network) Disconnect(oneID, otherID discover.NodeID) error {
conn := self.GetConn(oneID, otherID) conn := net.GetConn(oneID, otherID)
if conn == nil { 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)
} }
@ -306,13 +306,13 @@ func (self *Network) Disconnect(oneID, otherID discover.NodeID) error {
if err != nil { if err != nil {
return err return err
} }
self.events.Send(ControlEvent(conn)) net.events.Send(ControlEvent(conn))
return client.Call(nil, "admin_removePeer", string(conn.other.Addr())) return client.Call(nil, "admin_removePeer", string(conn.other.Addr()))
} }
// DidConnect tracks the fact that the "one" node connected to the "other" node // DidConnect tracks the fact that the "one" node connected to the "other" node
func (self *Network) DidConnect(one, other discover.NodeID) error { func (net *Network) DidConnect(one, other discover.NodeID) error {
conn, err := self.GetOrCreateConn(one, other) conn, err := net.GetOrCreateConn(one, other)
if err != nil { if err != nil {
return fmt.Errorf("connection between %v and %v does not exist", one, other) return fmt.Errorf("connection between %v and %v does not exist", one, other)
} }
@ -320,14 +320,14 @@ func (self *Network) DidConnect(one, other discover.NodeID) error {
return fmt.Errorf("%v and %v already connected", one, other) return fmt.Errorf("%v and %v already connected", one, other)
} }
conn.Up = true conn.Up = true
self.events.Send(NewEvent(conn)) net.events.Send(NewEvent(conn))
return nil return nil
} }
// DidDisconnect tracks the fact that the "one" node disconnected from the // DidDisconnect tracks the fact that the "one" node disconnected from the
// "other" node // "other" node
func (self *Network) DidDisconnect(one, other discover.NodeID) error { func (net *Network) DidDisconnect(one, other discover.NodeID) error {
conn := self.GetConn(one, other) conn := net.GetConn(one, other)
if conn == nil { if conn == nil {
return fmt.Errorf("connection between %v and %v does not exist", one, other) return fmt.Errorf("connection between %v and %v does not exist", one, other)
} }
@ -336,12 +336,12 @@ func (self *Network) DidDisconnect(one, other discover.NodeID) error {
} }
conn.Up = false conn.Up = false
conn.initiated = time.Now().Add(-dialBanTimeout) conn.initiated = time.Now().Add(-dialBanTimeout)
self.events.Send(NewEvent(conn)) net.events.Send(NewEvent(conn))
return nil return nil
} }
// DidSend tracks the fact that "sender" sent a message to "receiver" // DidSend tracks the fact that "sender" sent a message to "receiver"
func (self *Network) DidSend(sender, receiver discover.NodeID, proto string, code uint64) error { func (net *Network) DidSend(sender, receiver discover.NodeID, proto string, code uint64) error {
msg := &Msg{ msg := &Msg{
One: sender, One: sender,
Other: receiver, Other: receiver,
@ -349,12 +349,12 @@ func (self *Network) DidSend(sender, receiver discover.NodeID, proto string, cod
Code: code, Code: code,
Received: false, Received: false,
} }
self.events.Send(NewEvent(msg)) net.events.Send(NewEvent(msg))
return nil return nil
} }
// DidReceive tracks the fact that "receiver" received a message from "sender" // DidReceive tracks the fact that "receiver" received a message from "sender"
func (self *Network) DidReceive(sender, receiver discover.NodeID, proto string, code uint64) error { func (net *Network) DidReceive(sender, receiver discover.NodeID, proto string, code uint64) error {
msg := &Msg{ msg := &Msg{
One: sender, One: sender,
Other: receiver, Other: receiver,
@ -362,36 +362,36 @@ func (self *Network) DidReceive(sender, receiver discover.NodeID, proto string,
Code: code, Code: code,
Received: true, Received: true,
} }
self.events.Send(NewEvent(msg)) net.events.Send(NewEvent(msg))
return nil return nil
} }
// GetNode gets the node with the given ID, returning nil if the node does not // GetNode gets the node with the given ID, returning nil if the node does not
// exist // exist
func (self *Network) GetNode(id discover.NodeID) *Node { func (net *Network) GetNode(id discover.NodeID) *Node {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
return self.getNode(id) return net.getNode(id)
} }
// GetNode gets the node with the given name, returning nil if the node does // GetNode gets the node with the given name, returning nil if the node does
// not exist // not exist
func (self *Network) GetNodeByName(name string) *Node { func (net *Network) GetNodeByName(name string) *Node {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
return self.getNodeByName(name) return net.getNodeByName(name)
} }
func (self *Network) getNode(id discover.NodeID) *Node { func (net *Network) getNode(id discover.NodeID) *Node {
i, found := self.nodeMap[id] i, found := net.nodeMap[id]
if !found { if !found {
return nil return nil
} }
return self.Nodes[i] return net.Nodes[i]
} }
func (self *Network) getNodeByName(name string) *Node { func (net *Network) getNodeByName(name string) *Node {
for _, node := range self.Nodes { for _, node := range net.Nodes {
if node.Config.Name == name { if node.Config.Name == name {
return node return node
} }
@ -400,40 +400,40 @@ func (self *Network) getNodeByName(name string) *Node {
} }
// GetNodes returns the existing nodes // GetNodes returns the existing nodes
func (self *Network) GetNodes() (nodes []*Node) { func (net *Network) GetNodes() (nodes []*Node) {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
nodes = append(nodes, self.Nodes...) nodes = append(nodes, net.Nodes...)
return nodes return nodes
} }
// GetConn returns the connection which exists between "one" and "other" // GetConn returns the connection which exists between "one" and "other"
// regardless of which node initiated the connection // regardless of which node initiated the connection
func (self *Network) GetConn(oneID, otherID discover.NodeID) *Conn { func (net *Network) GetConn(oneID, otherID discover.NodeID) *Conn {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
return self.getConn(oneID, otherID) return net.getConn(oneID, otherID)
} }
// GetOrCreateConn is like GetConn but creates the connection if it doesn't // GetOrCreateConn is like GetConn but creates the connection if it doesn't
// already exist // already exist
func (self *Network) GetOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) { func (net *Network) GetOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
return self.getOrCreateConn(oneID, otherID) return net.getOrCreateConn(oneID, otherID)
} }
func (self *Network) getOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) { func (net *Network) getOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) {
if conn := self.getConn(oneID, otherID); conn != nil { if conn := net.getConn(oneID, otherID); conn != nil {
return conn, nil return conn, nil
} }
one := self.getNode(oneID) one := net.getNode(oneID)
if one == nil { if one == nil {
return nil, fmt.Errorf("node %v does not exist", oneID) return nil, fmt.Errorf("node %v does not exist", oneID)
} }
other := self.getNode(otherID) other := net.getNode(otherID)
if other == nil { if other == nil {
return nil, fmt.Errorf("node %v does not exist", otherID) return nil, fmt.Errorf("node %v does not exist", otherID)
} }
@ -444,18 +444,18 @@ func (self *Network) getOrCreateConn(oneID, otherID discover.NodeID) (*Conn, err
other: other, other: other,
} }
label := ConnLabel(oneID, otherID) label := ConnLabel(oneID, otherID)
self.connMap[label] = len(self.Conns) net.connMap[label] = len(net.Conns)
self.Conns = append(self.Conns, conn) net.Conns = append(net.Conns, conn)
return conn, nil return conn, nil
} }
func (self *Network) getConn(oneID, otherID discover.NodeID) *Conn { func (net *Network) getConn(oneID, otherID discover.NodeID) *Conn {
label := ConnLabel(oneID, otherID) label := ConnLabel(oneID, otherID)
i, found := self.connMap[label] i, found := net.connMap[label]
if !found { if !found {
return nil return nil
} }
return self.Conns[i] return net.Conns[i]
} }
// InitConn(one, other) retrieves the connectiton model for the connection between // InitConn(one, other) retrieves the connectiton model for the connection between
@ -466,13 +466,13 @@ func (self *Network) getConn(oneID, otherID discover.NodeID) *Conn {
// it also checks whether there has been recent attempt to connect the peers // it also checks whether there has been recent attempt to connect the peers
// this is cheating as the simulation is used as an oracle and know about // this is cheating as the simulation is used as an oracle and know about
// remote peers attempt to connect to a node which will then not initiate the connection // remote peers attempt to connect to a node which will then not initiate the connection
func (self *Network) InitConn(oneID, otherID discover.NodeID) (*Conn, error) { func (net *Network) InitConn(oneID, otherID discover.NodeID) (*Conn, error) {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
if oneID == otherID { if oneID == otherID {
return nil, fmt.Errorf("refusing to connect to self %v", oneID) return nil, fmt.Errorf("refusing to connect to self %v", oneID)
} }
conn, err := self.getOrCreateConn(oneID, otherID) conn, err := net.getOrCreateConn(oneID, otherID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -491,28 +491,28 @@ func (self *Network) InitConn(oneID, otherID discover.NodeID) (*Conn, error) {
} }
// Shutdown stops all nodes in the network and closes the quit channel // Shutdown stops all nodes in the network and closes the quit channel
func (self *Network) Shutdown() { func (net *Network) Shutdown() {
for _, node := range self.Nodes { for _, node := range net.Nodes {
log.Debug(fmt.Sprintf("stopping node %s", node.ID().TerminalString())) log.Debug(fmt.Sprintf("stopping node %s", node.ID().TerminalString()))
if err := node.Stop(); err != nil { if err := node.Stop(); err != nil {
log.Warn(fmt.Sprintf("error stopping node %s", node.ID().TerminalString()), "err", err) log.Warn(fmt.Sprintf("error stopping node %s", node.ID().TerminalString()), "err", err)
} }
} }
close(self.quitc) close(net.quitc)
} }
//Reset resets all network properties: //Reset resets all network properties:
//emtpies the nodes and the connection list //emtpies the nodes and the connection list
func (self *Network) Reset() { func (net *Network) Reset() {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
//re-initialize the maps //re-initialize the maps
self.connMap = make(map[string]int) net.connMap = make(map[string]int)
self.nodeMap = make(map[discover.NodeID]int) net.nodeMap = make(map[discover.NodeID]int)
self.Nodes = nil net.Nodes = nil
self.Conns = nil net.Conns = nil
} }
// Node is a wrapper around adapters.Node which is used to track the status // Node is a wrapper around adapters.Node which is used to track the status
@ -528,37 +528,37 @@ type Node struct {
} }
// ID returns the ID of the node // ID returns the ID of the node
func (self *Node) ID() discover.NodeID { func (n *Node) ID() discover.NodeID {
return self.Config.ID return n.Config.ID
} }
// String returns a log-friendly string // String returns a log-friendly string
func (self *Node) String() string { func (n *Node) String() string {
return fmt.Sprintf("Node %v", self.ID().TerminalString()) return fmt.Sprintf("Node %v", n.ID().TerminalString())
} }
// NodeInfo returns information about the node // NodeInfo returns information about the node
func (self *Node) NodeInfo() *p2p.NodeInfo { func (n *Node) NodeInfo() *p2p.NodeInfo {
// avoid a panic if the node is not started yet // avoid a panic if the node is not started yet
if self.Node == nil { if n.Node == nil {
return nil return nil
} }
info := self.Node.NodeInfo() info := n.Node.NodeInfo()
info.Name = self.Config.Name info.Name = n.Config.Name
return info return info
} }
// MarshalJSON implements the json.Marshaler interface so that the encoded // MarshalJSON implements the json.Marshaler interface so that the encoded
// JSON includes the NodeInfo // JSON includes the NodeInfo
func (self *Node) MarshalJSON() ([]byte, error) { func (n *Node) MarshalJSON() ([]byte, error) {
return json.Marshal(struct { return json.Marshal(struct {
Info *p2p.NodeInfo `json:"info,omitempty"` Info *p2p.NodeInfo `json:"info,omitempty"`
Config *adapters.NodeConfig `json:"config,omitempty"` Config *adapters.NodeConfig `json:"config,omitempty"`
Up bool `json:"up"` Up bool `json:"up"`
}{ }{
Info: self.NodeInfo(), Info: n.NodeInfo(),
Config: self.Config, Config: n.Config,
Up: self.Up, Up: n.Up,
}) })
} }
@ -580,19 +580,19 @@ type Conn struct {
} }
// nodesUp returns whether both nodes are currently up // nodesUp returns whether both nodes are currently up
func (self *Conn) nodesUp() error { func (c *Conn) nodesUp() error {
if !self.one.Up { if !c.one.Up {
return fmt.Errorf("one %v is not up", self.One) return fmt.Errorf("one %v is not up", c.One)
} }
if !self.other.Up { if !c.other.Up {
return fmt.Errorf("other %v is not up", self.Other) return fmt.Errorf("other %v is not up", c.Other)
} }
return nil return nil
} }
// String returns a log-friendly string // String returns a log-friendly string
func (self *Conn) String() string { func (c *Conn) String() string {
return fmt.Sprintf("Conn %v->%v", self.One.TerminalString(), self.Other.TerminalString()) return fmt.Sprintf("Conn %v->%v", c.One.TerminalString(), c.Other.TerminalString())
} }
// Msg represents a p2p message sent between two nodes in the network // Msg represents a p2p message sent between two nodes in the network
@ -605,8 +605,8 @@ type Msg struct {
} }
// String returns a log-friendly string // String returns a log-friendly string
func (self *Msg) String() string { func (m *Msg) String() string {
return fmt.Sprintf("Msg(%d) %v->%v", self.Code, self.One.TerminalString(), self.Other.TerminalString()) return fmt.Sprintf("Msg(%d) %v->%v", m.Code, m.One.TerminalString(), m.Other.TerminalString())
} }
// ConnLabel generates a deterministic string which represents a connection // ConnLabel generates a deterministic string which represents a connection
@ -640,14 +640,14 @@ type NodeSnapshot struct {
} }
// Snapshot creates a network snapshot // Snapshot creates a network snapshot
func (self *Network) Snapshot() (*Snapshot, error) { func (net *Network) Snapshot() (*Snapshot, error) {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
snap := &Snapshot{ snap := &Snapshot{
Nodes: make([]NodeSnapshot, len(self.Nodes)), Nodes: make([]NodeSnapshot, len(net.Nodes)),
Conns: make([]Conn, len(self.Conns)), Conns: make([]Conn, len(net.Conns)),
} }
for i, node := range self.Nodes { for i, node := range net.Nodes {
snap.Nodes[i] = NodeSnapshot{Node: *node} snap.Nodes[i] = NodeSnapshot{Node: *node}
if !node.Up { if !node.Up {
continue continue
@ -658,33 +658,33 @@ func (self *Network) Snapshot() (*Snapshot, error) {
} }
snap.Nodes[i].Snapshots = snapshots snap.Nodes[i].Snapshots = snapshots
} }
for i, conn := range self.Conns { for i, conn := range net.Conns {
snap.Conns[i] = *conn snap.Conns[i] = *conn
} }
return snap, nil return snap, nil
} }
// Load loads a network snapshot // Load loads a network snapshot
func (self *Network) Load(snap *Snapshot) error { func (net *Network) Load(snap *Snapshot) error {
for _, n := range snap.Nodes { for _, n := range snap.Nodes {
if _, err := self.NewNodeWithConfig(n.Node.Config); err != nil { if _, err := net.NewNodeWithConfig(n.Node.Config); err != nil {
return err return err
} }
if !n.Node.Up { if !n.Node.Up {
continue continue
} }
if err := self.startWithSnapshots(n.Node.Config.ID, n.Snapshots); err != nil { if err := net.startWithSnapshots(n.Node.Config.ID, n.Snapshots); err != nil {
return err return err
} }
} }
for _, conn := range snap.Conns { for _, conn := range snap.Conns {
if !self.GetNode(conn.One).Up || !self.GetNode(conn.Other).Up { if !net.GetNode(conn.One).Up || !net.GetNode(conn.Other).Up {
//in this case, at least one of the nodes of a connection is not up, //in this case, at least one of the nodes of a connection is not up,
//so it would result in the snapshot `Load` to fail //so it would result in the snapshot `Load` to fail
continue continue
} }
if err := self.Connect(conn.One, conn.Other); err != nil { if err := net.Connect(conn.One, conn.Other); err != nil {
return err return err
} }
} }
@ -692,7 +692,7 @@ func (self *Network) Load(snap *Snapshot) error {
} }
// Subscribe reads control events from a channel and executes them // Subscribe reads control events from a channel and executes them
func (self *Network) Subscribe(events chan *Event) { func (net *Network) Subscribe(events chan *Event) {
for { for {
select { select {
case event, ok := <-events: case event, ok := <-events:
@ -700,23 +700,23 @@ func (self *Network) Subscribe(events chan *Event) {
return return
} }
if event.Control { if event.Control {
self.executeControlEvent(event) net.executeControlEvent(event)
} }
case <-self.quitc: case <-net.quitc:
return return
} }
} }
} }
func (self *Network) executeControlEvent(event *Event) { func (net *Network) executeControlEvent(event *Event) {
log.Trace("execute control event", "type", event.Type, "event", event) log.Trace("execute control event", "type", event.Type, "event", event)
switch event.Type { switch event.Type {
case EventTypeNode: case EventTypeNode:
if err := self.executeNodeEvent(event); err != nil { if err := net.executeNodeEvent(event); err != nil {
log.Error("error executing node event", "event", event, "err", err) log.Error("error executing node event", "event", event, "err", err)
} }
case EventTypeConn: case EventTypeConn:
if err := self.executeConnEvent(event); err != nil { if err := net.executeConnEvent(event); err != nil {
log.Error("error executing conn event", "event", event, "err", err) log.Error("error executing conn event", "event", event, "err", err)
} }
case EventTypeMsg: case EventTypeMsg:
@ -724,20 +724,21 @@ func (self *Network) executeControlEvent(event *Event) {
} }
} }
func (self *Network) executeNodeEvent(e *Event) error { func (net *Network) executeNodeEvent(e *Event) error {
if !e.Node.Up { if !e.Node.Up {
return self.Stop(e.Node.ID()) return net.Stop(e.Node.ID())
} }
if _, err := self.NewNodeWithConfig(e.Node.Config); err != nil { if _, err := net.NewNodeWithConfig(e.Node.Config); err != nil {
return err return err
} }
return self.Start(e.Node.ID()) return net.Start(e.Node.ID())
} }
func (self *Network) executeConnEvent(e *Event) error { func (net *Network) executeConnEvent(e *Event) error {
if e.Conn.Up { if e.Conn.Up {
return self.Connect(e.Conn.One, e.Conn.Other) return net.Connect(e.Conn.One, e.Conn.Other)
} else {
return net.Disconnect(e.Conn.One, e.Conn.Other)
} }
return self.Disconnect(e.Conn.One, e.Conn.Other)
} }

34
p2p/testing/peerpool.go
Unescape View File

@ -39,29 +39,29 @@ func NewTestPeerPool() *TestPeerPool {
return &TestPeerPool{peers: make(map[discover.NodeID]TestPeer)} return &TestPeerPool{peers: make(map[discover.NodeID]TestPeer)}
} }
func (self *TestPeerPool) Add(p TestPeer) { func (p *TestPeerPool) Add(peer TestPeer) {
self.lock.Lock() p.lock.Lock()
defer self.lock.Unlock() defer p.lock.Unlock()
log.Trace(fmt.Sprintf("pp add peer %v", p.ID())) log.Trace(fmt.Sprintf("pp add peer %v", peer.ID()))
self.peers[p.ID()] = p p.peers[peer.ID()] = peer
} }
func (self *TestPeerPool) Remove(p TestPeer) { func (p *TestPeerPool) Remove(peer TestPeer) {
self.lock.Lock() p.lock.Lock()
defer self.lock.Unlock() defer p.lock.Unlock()
delete(self.peers, p.ID()) delete(p.peers, peer.ID())
} }
func (self *TestPeerPool) Has(id discover.NodeID) bool { func (p *TestPeerPool) Has(id discover.NodeID) bool {
self.lock.Lock() p.lock.Lock()
defer self.lock.Unlock() defer p.lock.Unlock()
_, ok := self.peers[id] _, ok := p.peers[id]
return ok return ok
} }
func (self *TestPeerPool) Get(id discover.NodeID) TestPeer { func (p *TestPeerPool) Get(id discover.NodeID) TestPeer {
self.lock.Lock() p.lock.Lock()
defer self.lock.Unlock() defer p.lock.Unlock()
return self.peers[id] return p.peers[id]
} }

26
p2p/testing/protocolsession.go
Unescape View File

@ -78,10 +78,10 @@ type Disconnect struct {
} }
// trigger sends messages from peers // trigger sends messages from peers
func (self *ProtocolSession) trigger(trig Trigger) error { func (s *ProtocolSession) trigger(trig Trigger) error {
simNode, ok := self.adapter.GetNode(trig.Peer) simNode, ok := s.adapter.GetNode(trig.Peer)
if !ok { 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(s.IDs))
} }
mockNode, ok := simNode.Services()[0].(*mockNode) mockNode, ok := simNode.Services()[0].(*mockNode)
if !ok { if !ok {
@ -107,7 +107,7 @@ func (self *ProtocolSession) trigger(trig Trigger) error {
} }
// expect checks an expectation of a message sent out by the pivot node // expect checks an expectation of a message sent out by the pivot node
func (self *ProtocolSession) expect(exps []Expect) error { func (s *ProtocolSession) expect(exps []Expect) error {
// construct a map of expectations for each node // construct a map of expectations for each node
peerExpects := make(map[discover.NodeID][]Expect) peerExpects := make(map[discover.NodeID][]Expect)
for _, exp := range exps { for _, exp := range exps {
@ -120,9 +120,9 @@ func (self *ProtocolSession) expect(exps []Expect) error {
// construct a map of mockNodes for each node // construct a map of mockNodes for each node
mockNodes := make(map[discover.NodeID]*mockNode) mockNodes := make(map[discover.NodeID]*mockNode)
for nodeID := range peerExpects { for nodeID := range peerExpects {
simNode, ok := self.adapter.GetNode(nodeID) simNode, ok := s.adapter.GetNode(nodeID)
if !ok { if !ok {
return fmt.Errorf("trigger: peer %v does not exist (1- %v)", nodeID, len(self.IDs)) return fmt.Errorf("trigger: peer %v does not exist (1- %v)", nodeID, len(s.IDs))
} }
mockNode, ok := simNode.Services()[0].(*mockNode) mockNode, ok := simNode.Services()[0].(*mockNode)
if !ok { if !ok {
@ -202,9 +202,9 @@ func (self *ProtocolSession) expect(exps []Expect) error {
} }
// TestExchanges tests a series of exchanges against the session // TestExchanges tests a series of exchanges against the session
func (self *ProtocolSession) TestExchanges(exchanges ...Exchange) error { func (s *ProtocolSession) TestExchanges(exchanges ...Exchange) error {
for i, e := range exchanges { for i, e := range exchanges {
if err := self.testExchange(e); err != nil { if err := s.testExchange(e); err != nil {
return fmt.Errorf("exchange #%d %q: %v", i, e.Label, err) return fmt.Errorf("exchange #%d %q: %v", i, e.Label, err)
} }
log.Trace(fmt.Sprintf("exchange #%d %q: run successfully", i, e.Label)) log.Trace(fmt.Sprintf("exchange #%d %q: run successfully", i, e.Label))
@ -214,14 +214,14 @@ func (self *ProtocolSession) TestExchanges(exchanges ...Exchange) error {
// testExchange tests a single Exchange. // testExchange tests a single Exchange.
// Default timeout value is 2 seconds. // Default timeout value is 2 seconds.
func (self *ProtocolSession) testExchange(e Exchange) error { func (s *ProtocolSession) testExchange(e Exchange) error {
errc := make(chan error) errc := make(chan error)
done := make(chan struct{}) done := make(chan struct{})
defer close(done) defer close(done)
go func() { go func() {
for _, trig := range e.Triggers { for _, trig := range e.Triggers {
err := self.trigger(trig) err := s.trigger(trig)
if err != nil { if err != nil {
errc <- err errc <- err
return return
@ -229,7 +229,7 @@ func (self *ProtocolSession) testExchange(e Exchange) error {
} }
select { select {
case errc <- self.expect(e.Expects): case errc <- s.expect(e.Expects):
case <-done: case <-done:
} }
}() }()
@ -250,7 +250,7 @@ func (self *ProtocolSession) testExchange(e Exchange) error {
// TestDisconnected tests the disconnections given as arguments // TestDisconnected tests the disconnections given as arguments
// the disconnect structs describe what disconnect error is expected on which peer // the disconnect structs describe what disconnect error is expected on which peer
func (self *ProtocolSession) TestDisconnected(disconnects ...*Disconnect) error { func (s *ProtocolSession) TestDisconnected(disconnects ...*Disconnect) error {
expects := make(map[discover.NodeID]error) expects := make(map[discover.NodeID]error)
for _, disconnect := range disconnects { for _, disconnect := range disconnects {
expects[disconnect.Peer] = disconnect.Error expects[disconnect.Peer] = disconnect.Error
@ -259,7 +259,7 @@ func (self *ProtocolSession) TestDisconnected(disconnects ...*Disconnect) error
timeout := time.After(time.Second) timeout := time.After(time.Second)
for len(expects) > 0 { for len(expects) > 0 {
select { select {
case event := <-self.events: case event := <-s.events:
if event.Type != p2p.PeerEventTypeDrop { if event.Type != p2p.PeerEventTypeDrop {
continue continue
} }

12
p2p/testing/protocoltester.go
Unescape View File

@ -101,24 +101,24 @@ func NewProtocolTester(t *testing.T, id discover.NodeID, n int, run func(*p2p.Pe
} }
// Stop stops the p2p server // Stop stops the p2p server
func (self *ProtocolTester) Stop() error { func (t *ProtocolTester) Stop() error {
self.Server.Stop() t.Server.Stop()
return nil return nil
} }
// Connect brings up the remote peer node and connects it using the // Connect brings up the remote peer node and connects it using the
// p2p/simulations network connection with the in memory network adapter // p2p/simulations network connection with the in memory network adapter
func (self *ProtocolTester) Connect(selfID discover.NodeID, peers ...*adapters.NodeConfig) { func (t *ProtocolTester) Connect(selfID discover.NodeID, peers ...*adapters.NodeConfig) {
for _, peer := range peers { 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 { if _, err := t.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 { if err := t.network.Start(peer.ID); err != nil {
panic(fmt.Sprintf("error starting peer %v: %v", peer.ID, err)) panic(fmt.Sprintf("error starting peer %v: %v", peer.ID, err))
} }
log.Trace(fmt.Sprintf("connect to %v", peer.ID)) log.Trace(fmt.Sprintf("connect to %v", peer.ID))
if err := self.network.Connect(selfID, peer.ID); err != nil { if err := t.network.Connect(selfID, peer.ID); err != nil {
panic(fmt.Sprintf("error connecting to peer %v: %v", peer.ID, err)) panic(fmt.Sprintf("error connecting to peer %v: %v", peer.ID, err))
} }
} }

Write Preview
Loading…
Cancel
Save