swarm/network: WIP protocol testfixes

p2p/adapters/inproc.go

@@ -130,7 +130,8 @@ func (self *SimNode) Disconnect(rid []byte) error {
 	// peer = na.(*SimNode).GetPeer(self.Id)
 	// peer.RW = nil
 	glog.V(6).Infof("dropped peer %v", id)
-	return self.network.DidDisconnect(self.Id, id)
+	
+	return nil
 }
 
 func (self *SimNode) Connect(rid []byte) error {
@@ -155,7 +156,7 @@ func (self *SimNode) Connect(rid []byte) error {
 	if err != nil {
 		return fmt.Errorf("cannot run protocol (%v -> %v): %v", id, self.Id, err)
 	}
-	self.network.DidConnect(self.Id, id)
+	
 	return nil
 }
 
@@ -172,11 +173,13 @@ func (self *SimNode) RunProtocol(id *NodeId, rw, rrw p2p.MsgReadWriter, runc cha
 	peer = self.setPeer(id, self.Messenger(rrw))
 	p := p2p.NewPeer(id.NodeID, Name(id.Bytes()), []p2p.Cap{})
 	go func() {
+		self.network.DidConnect(self.Id, id)
 		err := self.Run(p, rw)
-		glog.V(6).Infof("protocol quit on peer %v (connection with %v broken: %v)", self.Id, id, err)
 		<-runc
-		// self.Disconnect(id.Bytes())
+		self.Disconnect(id.Bytes())
 		peer.Errc <- err
+		glog.V(6).Infof("protocol quit on peer %v (connection with %v broken: %v)", self.Id, id, err)
+		self.network.DidDisconnect(self.Id, id)
 	}()
 	return nil
 }

p2p/protocols/protocol.go

@@ -168,12 +168,7 @@ func NewProtocol(protocolname string, protocolversion uint, run func(*Peer) erro
 
 		m := na.Messenger(rw)
 
-		disc := func() {
+		peer := NewPeer(p, ct, m)
-			id := p.ID()
-			na.Disconnect(id[:])
-		}
-
-		peer := NewPeer(p, ct, m, disc)
 
 		return run(peer)
 
@@ -201,7 +196,6 @@ type Peer struct {
 	*p2p.Peer                                             // the p2p.Peer object representing the remote
 	rw         p2p.MsgReadWriter                          // p2p.MsgReadWriter to send messages to and read messages from
 	handlers   map[reflect.Type][]func(interface{}) error //  message type -> message handler callback(s) map
-	disconnect func()                                     // Disconnect function set differently for testing
 	Err        error
 }
 
@@ -209,13 +203,12 @@ type Peer struct {
 // this constructor is called by the p2p.Protocol#Run function
 // the first two arguments are comming the arguments passed to p2p.Protocol.Run function
 // the third argument is the CodeMap describing the protocol messages and options
-func NewPeer(p *p2p.Peer, ct *CodeMap, m adapters.Messenger, disconn func()) *Peer {
+func NewPeer(p *p2p.Peer, ct *CodeMap, m adapters.Messenger) *Peer {
 	return &Peer{
 		ct:         ct,
 		m:          m,
 		Peer:       p,
 		handlers:   make(map[reflect.Type][]func(interface{}) error),
-		disconnect: disconn,
 	}
 }
 
@@ -244,14 +237,18 @@ func (self *Peer) Run() error {
 	var err error
 	for {
 		_, err = self.handleIncoming()
-		if self.Err == nil {
+		if self.Err != nil {
-			err = self.Err
+			self.Err = err
+		}
+		if self.Err != nil {
+			break
+		}
+		
 	}
 	for _, f := range self.handlers[reflect.TypeOf(&Disconnect{})] {
 		f(err)
 	}
 	return err
-	}
 }
 
 // Drop disconnects a peer.
@@ -283,6 +280,12 @@ func (self *Peer) Send(msg interface{}) error {
 	return nil
 }
 
+func (self *Peer) DisconnectHook(f func(e interface{}) error) {
+	typ := reflect.TypeOf(&Disconnect{})
+	self.handlers[typ] = append(self.handlers[typ], f)
+}
+
+
 // handleIncoming(code)
 // is called each cycle of the main forever loop that handles and dispatches incoming messages
 // if this returns an error the loop returns and the peer is disconnected with the error

p2p/protocols/protocol_test.go

@@ -158,7 +158,7 @@ func protoHandshakeExchange(id *adapters.NodeId, proto *protoHandshake) []p2ptes
 }
 
 func runProtoHandshake(t *testing.T, proto *protoHandshake, errs ...error) {
-	pp := NewTestPeerPool()
+	pp := p2ptest.NewTestPeerPool()
 	s := protocolTester(t, pp, nil)
 	// TODO: make this more than one handshake
 	id := s.Ids[0]
@@ -207,7 +207,7 @@ func moduleHandshakeExchange(id *adapters.NodeId, resp uint) []p2ptest.Exchange
 }
 
 func runModuleHandshake(t *testing.T, resp uint, errs ...error) {
-	pp := NewTestPeerPool()
+	pp := p2ptest.NewTestPeerPool()
 	s := protocolTester(t, pp, nil)
 	id := s.Ids[0]
 	s.TestExchanges(protoHandshakeExchange(id, &protoHandshake{42, "420"})...)
@@ -282,7 +282,7 @@ func testMultiPeerSetup(a, b *adapters.NodeId) []p2ptest.Exchange {
 
 func runMultiplePeers(t *testing.T, peer int, errs ...error) {
 	wg := &sync.WaitGroup{}
-	pp := NewTestPeerPool()
+	pp := p2ptest.NewTestPeerPool()
 	s := protocolTester(t, pp, wg)
 
 	s.TestExchanges(testMultiPeerSetup(s.Ids[0], s.Ids[1])...)

p2p/testing/peerpool.go

@@ -0,0 +1,53 @@
+package testing
+
+import (
+	"sync"
+
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
+
+	"github.com/ethereum/go-ethereum/p2p/adapters"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+)
+
+type TestPeer interface {
+	ID() discover.NodeID
+	Drop(error)
+}
+
+// TestPeerPool is an example peerPool to demonstrate registration of peer connections
+type TestPeerPool struct {
+	lock  sync.Mutex
+	peers map[discover.NodeID]TestPeer
+}
+
+func NewTestPeerPool() *TestPeerPool {
+	return &TestPeerPool{peers: make(map[discover.NodeID]TestPeer)}
+}
+
+func (self *TestPeerPool) Add(p TestPeer) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	glog.V(logger.Detail).Infof("pp add peer  %v", p.ID())
+	self.peers[p.ID()] = p
+
+}
+
+func (self *TestPeerPool) Remove(p TestPeer) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	delete(self.peers, p.ID())
+}
+
+func (self *TestPeerPool) Has(n *adapters.NodeId) bool {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	_, ok := self.peers[n.NodeID]
+	return ok
+}
+
+func (self *TestPeerPool) Get(n *adapters.NodeId) TestPeer {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	return self.peers[n.NodeID]
+}

swarm/network/discovery.go

@@ -9,12 +9,12 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/discover"
 )
 
-// discovery bzz hive extension doing peer relaying
+// discovery bzz overlay extension doing peer relaying
 // can be switched off
 
 type discPeer struct {
 	Peer
-	hive      *Hive
+	overlay		Overlay
 	proxLimit uint8
 	peers     map[discover.NodeID]bool
 }
@@ -40,9 +40,9 @@ func (self *discPeer) NotifyProx(po uint8) error {
 }
 
 // new discovery contructor
-func NewDiscovery(p Peer, h *Hive) *discPeer {
+func NewDiscovery(p Peer, o Overlay) *discPeer {
 	self := &discPeer{
-		hive:  h,
+		overlay: o,
 		Peer:  p,
 		peers: make(map[discover.NodeID]bool),
 	}
@@ -118,7 +118,7 @@ func (p *discPeer) handlePeersMsg(msg interface{}) error {
 		nas = append(nas, addr)
 		p.peers[NodeId(addr).NodeID] = true
 	}
-	return p.hive.Register(nas...)
+	return p.overlay.Register(nas...)
 }
 
 // handleGetPeersMsg is called by the protocol when receiving a
@@ -131,7 +131,7 @@ func (p *discPeer) handleGetPeersMsg(msg interface{}) error {
 	var peers []*peerAddr
 	alreadySent := p.peers
 	i := 0
-	p.hive.EachLivePeer(p.OverlayAddr(), int(req.Order), func(n Peer, po int) bool {
+	p.overlay.EachLivePeer(p.OverlayAddr(), int(req.Order), func(n Peer, po int) bool {
 		i++
 		if bytes.Compare(n.OverlayAddr(), p.OverlayAddr()) != 0 &&
 			// only send peers we have not sent before in this session

swarm/network/discovery_test.go

@@ -6,12 +6,39 @@ import (
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
 )
 
+/***
+ * 
+ * - after connect, that outgoing subpeersmsg is sent
+ * 
+ */
 func TestDiscovery(t *testing.T) {
 	addr := RandomAddr()
 	to := NewKademlia(addr.OAddr, NewKadParams())
-	pp := NewHive(NewHiveParams(), to)
+	pp := p2ptest.NewTestPeerPool()
+	//pp := NewHive(NewHiveParams(), to)
 	ct := BzzCodeMap(HiveMsgs...)
-	s := newBzzTester(t, addr, pp, ct, nil)
+	
+	services := func(p Peer) error {
+		dp := NewDiscovery(p, to)
+		//pp.Add(p)
+		to.On(dp)
+		p.DisconnectHook(func(e interface{}) error {
+			dp := e.(Peer)
+			to.Off(dp)
+			return nil
+		})
+		return nil
+	}
+	/*
+	protocall := func (na adapters.NodeAdapter) adapters.ProtoCall {
+		protocol := Bzz(addr.OverlayAddr(), na, ct, services, nil, nil)	
+		return protocol.Run
+	}
+	
+	s := p2ptest.NewProtocolTester(t, NodeId(addr), 1, protocall)
+*/
+
+	s := newBzzTester(t, addr, pp, ct, services)
 
 	s.runHandshakes()
 	s.TestExchanges(p2ptest.Exchange{
@@ -19,7 +46,7 @@ func TestDiscovery(t *testing.T) {
 			p2ptest.Expect{
 				Code: 3,
 				Msg:  &SubPeersMsg{ProxLimit: 0, MinProxBinSize: 8},
-				Peer: s.ExchangeSession.Ids[1],
+				Peer: s.ExchangeSession.Ids[0],
 			},
 		},
 	})

swarm/network/hive.go

@@ -216,11 +216,11 @@ func (self *Hive) wake() {
 // to register a connected (live) peer
 func (self *Hive) Add(p Peer) error {
 	defer self.wake()
-	dp := NewDiscovery(p, self)
+	dp := NewDiscovery(p, self.Overlay)
 	glog.V(logger.Debug).Infof("to add new bee %v", p)
 	self.On(dp)
 	glog.V(logger.Warn).Infof("%v", self)
-	dp.NotifyProx(0)
+	//dp.NotifyProx(0)
 	return nil
 }
 

swarm/network/hive_test.go

@@ -43,22 +43,40 @@ func TestOverlayRegistration(t *testing.T) {
 	to := NewTestOverlay(addr.OverlayAddr()) // overlay topology driver
 	pp := NewHive(NewHiveParams(), to)       // hive
 	ct := BzzCodeMap(HiveMsgs...)            // bzz protocol code map
+	services := func(p Peer) error {
+		pp.Add(p)
+		p.DisconnectHook(func(e interface{}) error {
+			pp.Remove(p)
+			return nil
+		})
+		return nil
+	}
 	
-	s := newBzzTester(t, addr, pp, ct, nil)
+	protocall := func (na adapters.NodeAdapter) adapters.ProtoCall {
+		protocol := Bzz(addr.OverlayAddr(), na, ct, services, nil, nil)	
+		return protocol.Run
+	}
 	
-	// connect to the other peer
+	es := p2ptest.NewProtocolTester(t, NodeId(addr), 1, protocall)
+	
+	s := &bzzTester{
+		addr: addr,
+		ExchangeSession: es,
+	}
 	id := s.Ids[0]
 	raddr := NewPeerAddrFromNodeId(id)
+	
 	s.runHandshakes()
 
 	// hive should have called the overlay
 	if to.posMap[string(raddr.OverlayAddr())] == nil {
 		t.Fatalf("Overlay#On not called on new peer")
 	}
+	
 }
 
 func TestRegisterAndConnect(t *testing.T) {
-	addr := RandomAddr()
+	/*addr := RandomAddr()
 	to := NewTestOverlay(addr.OverlayAddr())
 	pp := NewHive(NewHiveParams(), to)
 	ct := BzzCodeMap(HiveMsgs...)
@@ -86,7 +104,39 @@ func TestRegisterAndConnect(t *testing.T) {
 	tc.ticker <- time.Now()
 
 	// run bzz handshake
+	s.runHandshakes()*/
+	
+	// setup
+	addr := RandomAddr()                     // tested peers peer address
+	to := NewTestOverlay(addr.OverlayAddr()) // overlay topology driver
+	pp := NewHive(NewHiveParams(), to)       // hive
+	ct := BzzCodeMap(HiveMsgs...)            // bzz protocol code map
+	services := func(p Peer) error {
+		pp.Add(p)
+		p.DisconnectHook(func(e interface{}) error {
+			pp.Remove(p)
+			return nil
+		})
+		return nil
+	}
+	
+	protocall := func (na adapters.NodeAdapter) adapters.ProtoCall {
+		protocol := Bzz(addr.OverlayAddr(), na, ct, services, nil, nil)	
+		return protocol.Run
+	}
+	
+	es := p2ptest.NewProtocolTester(t, NodeId(addr), 1, protocall)
+	
+	s := &bzzTester{
+		addr: addr,
+		ExchangeSession: es,
+	}
+	id := s.Ids[0]
+	raddr := NewPeerAddrFromNodeId(id)
+	
 	s.runHandshakes()
+	
+	
 	if to.posMap[string(raddr.OverlayAddr())] == nil {
 		t.Fatalf("Overlay#On not called on new peer")
 	}

swarm/network/protocol.go

@@ -65,6 +65,7 @@ type Peer interface {
 	Send(interface{}) error                               // can send messages
 	Drop(error)                                           // disconnect this peer
 	Register(interface{}, func(interface{}) error) uint64 // register message-handler callbacks
+	DisconnectHook(func(interface{}) error)
 }
 
 func BzzCodeMap(msgs ...interface{}) *protocols.CodeMap {

swarm/network/protocol_test.go

@@ -34,13 +34,17 @@ func bzzHandshakeExchange(lhs, rhs *bzzHandshake, id *adapters.NodeId) []p2ptest
 	}
 }
 
-func newBzzTester(t *testing.T, addr *peerAddr, pp *Hive, ct *protocols.CodeMap, services func(Peer) error) *bzzTester {
+
+func newBzzTester(t *testing.T, addr *peerAddr, pp *p2ptest.TestPeerPool, ct *protocols.CodeMap, services func(Peer) error) *bzzTester {
 	if ct == nil {
 		ct = BzzCodeMap()
 	}
 	extraservices := func(p Peer) error {
 		pp.Add(p)
-		p.Register(&protocols.Disconnect{}, func(e interface{}) error { pp.Remove(p) })
+		p.DisconnectHook(func(e interface{}) error {
+			pp.Remove(p)
+			return nil
+		})
 		
 		if services != nil {
 			err := services(p)
@@ -51,8 +55,8 @@ func newBzzTester(t *testing.T, addr *peerAddr, pp *Hive, ct *protocols.CodeMap,
 		return nil
 	}
 	
-	protocall := func(na adapters.NodeAdapter) adapters.ProtoCall {
+	protocall := func (na adapters.NodeAdapter) adapters.ProtoCall {
-		protocol := Bzz(addr.OverlayAddr(), pp, na, ct, extraservices, nil, nil)
+		protocol := Bzz(addr.OverlayAddr(), na, ct, extraservices, nil, nil)	
 		return protocol.Run
 	}
 	
@@ -161,7 +165,7 @@ func TestBzzPeerPoolRemove(t *testing.T) {
 	s.runHandshakes()
 
 	id := s.Ids[0]
-	pp.Get(id).Drop()
+	pp.Get(id).Drop(nil)
 	s.TestDisconnected(&p2ptest.Disconnect{id, fmt.Errorf("p2p: read or write on closed message pipe")})
 	if pp.Has(id) {
 		t.Fatalf("peer '%v' not removed: %v", id, pp)
@@ -179,7 +183,7 @@ func TestBzzPeerPoolBothAddRemove(t *testing.T) {
 		t.Fatalf("peer '%v' not added: %v", id, pp)
 	}
 
-	pp.Get(id).Drop()
+	pp.Get(id).Drop(nil)
 	s.TestDisconnected(&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("p2p: read or write on closed message pipe")})
 	if pp.Has(id) {
 		t.Fatalf("peer '%v' not removed: %v", id, pp)