swarm/pss, swarm/network, p2p/simulations: pss prototest fix

p2p/simulations/adapters/state.go

@@ -0,0 +1,20 @@
+package adapters
+
+type SimStateStore struct {
+	m map[string][]byte
+}
+
+func (self *SimStateStore) Load(s string) ([]byte, error) {
+	return self.m[s], nil
+}
+
+func (self *SimStateStore) Save(s string, data []byte) error {
+	self.m[s] = data
+	return nil
+}
+
+func NewSimStateStore() *SimStateStore {
+	return &SimStateStore{
+		make(map[string][]byte),
+	}
+}

swarm/network/protocol.go

@@ -347,15 +347,15 @@ func NewNodeIdFromAddr(addr Addr) *adapters.NodeId {
 
 // 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 {
+func NewAddrFromNodeId(n *adapters.NodeId) Addr {
 	id := n.NodeID
 	return &bzzAddr{
-		OAddr: ToOverlayAddr(n),
+		OAddr: ToOverlayAddr(n.Bytes()),
 		UAddr: []byte(discover.NewNode(id, net.IP{127, 0, 0, 1}, 30303, 30303).String()),
 	}
 }
 
 // ToOverlayAddr creates an overlayaddress from NodeID
-func ToOverlayAddr(id *adapters.NodeId) []byte {
-	return crypto.Keccak256(id.Bytes())
+func ToOverlayAddr(id []byte) []byte {
+	return crypto.Keccak256(id)
 }

swarm/pss/common.go

@@ -68,7 +68,7 @@ func newTestPss(addr []byte) *Pss {
 	return ps
 }
 
-func newPssPingMsg(ps *Pss, spec *protocols.Spec, topic PssTopic, senderaddr []byte) PssMsg {
+func newPssPingMsg(ps *Pss, to []byte, spec *protocols.Spec, topic PssTopic, senderaddr []byte) PssMsg {
 	data := pssPingMsg{
 		Created: time.Now(),
 	}
@@ -83,7 +83,7 @@ func newPssPingMsg(ps *Pss, spec *protocols.Spec, topic PssTopic, senderaddr []b
 	}
 
 	pssmsg := PssMsg{
-		To: ps.Overlay.BaseAddr(),
+		To: to,
 		Payload: NewPssEnvelope(senderaddr, topic, rlpbundle),
 	}
 
@@ -103,3 +103,23 @@ func newPssPingProtocol(handler func (interface{}) error) *p2p.Protocol {
 		},
 	}
 }
+
+type testPssPeer struct {
+	*protocols.Peer
+	addr []byte
+}
+
+func (self *testPssPeer) Address() []byte {
+	return self.addr
+}
+
+func (self *testPssPeer) Off() network.OverlayAddr {
+	return self
+}
+
+func (self *testPssPeer) Drop(err error) {
+}
+
+func (self *testPssPeer) Update(o network.OverlayAddr) network.OverlayAddr {
+	return self
+}

swarm/pss/pss.go

@@ -37,6 +37,11 @@ var (
 	errorForwardToSelf = errors.New("forward to self")
 )
 
+type senderPeer interface {
+	Address() []byte
+	Send(interface{}) error
+}
+
 // Defines params for Pss
 type PssParams struct {
 	Cachettl time.Duration
@@ -99,7 +104,7 @@ func (msg *PssMsg) serialize() []byte {
 }
 
 
-var pssTransportProtocol = &protocols.Spec{
+var pssSpec = &protocols.Spec{
 	Name:       "pss",
 	Version:    1,
 	MaxMsgSize: 10 * 1024 * 1024,
@@ -188,16 +193,16 @@ func (self *Pss) Stop() error {
 func (self *Pss) Protocols() []p2p.Protocol {
 	return []p2p.Protocol{
 		p2p.Protocol{
-			Name:    pssTransportProtocol.Name,
-			Version: pssTransportProtocol.Version,
-			Length:  pssTransportProtocol.Length(),
+			Name:    pssSpec.Name,
+			Version: pssSpec.Version,
+			Length:  pssSpec.Length(),
 			Run:     self.Run,
 		},
 	}
 }
 
 func (self *Pss) Run(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-	pp := protocols.NewPeer(p, rw, pssTransportProtocol)
+	pp := protocols.NewPeer(p, rw, pssSpec)
 	return pp.Run(self.handlePssMsg)
 }
 
@@ -371,13 +376,13 @@ func (self *Pss) Forward(msg *PssMsg) error {
 	// send with kademlia
 	// find the closest peer to the recipient and attempt to send
 	self.Overlay.EachConn(msg.To, 256, func(op network.OverlayConn, po int, isproxbin bool) bool {
-		p, ok := op.(network.Peer)
+		p, ok := op.(senderPeer)
 		if !ok {
 			return true
 		}
 		addr := self.Overlay.BaseAddr()
-		sendMsg := fmt.Sprintf("%x: msg to %x via %x", common.ByteLabel(addr), common.ByteLabel(msg.To), common.ByteLabel(p.Over()))
-		if self.checkFwdCache(p.Over(), digest) {
+		sendMsg := fmt.Sprintf("%x: msg to %x via %x", common.ByteLabel(addr), common.ByteLabel(msg.To), common.ByteLabel(p.Address()))
+		if self.checkFwdCache(p.Address(), digest) {
 			log.Info(fmt.Sprintf("%v: peer already forwarded to", sendMsg))
 			return true
 		}
@@ -389,10 +394,10 @@ func (self *Pss) Forward(msg *PssMsg) error {
 		log.Trace(fmt.Sprintf("%v: successfully forwarded", sendMsg))
 		sent++
 		// if equality holds, p is always the first peer given in the iterator
-		if bytes.Equal(msg.To, p.Over()) || !isproxbin {
+		if bytes.Equal(msg.To, p.Address()) || !isproxbin {
 			return false
 		}
-		log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ByteLabel(p.Over())))
+		log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ByteLabel(p.Address())))
 		return true
 	})
 

swarm/pss/pss_test.go

@@ -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/protocols"
 	"github.com/ethereum/go-ethereum/p2p/simulations"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
@@ -131,7 +132,7 @@ func TestPssRegisterHandler(t *testing.T) {
 	ps := newTestPss(addr.OAddr)
 	from := network.RandomAddr()
 	payload := []byte("payload")
-	topic := NewTopic(pssTransportProtocol.Name, int(pssTransportProtocol.Version))
+	topic := NewTopic(pssSpec.Name, int(pssSpec.Version))
 	wrongtopic := NewTopic("foo", 42)
 	checkMsg := func(msg []byte, p *p2p.Peer, sender []byte) error {
 		if !bytes.Equal(from.OAddr, sender) {
@@ -172,23 +173,36 @@ func TestPssRegisterHandler(t *testing.T) {
 }
 
 func TestPssSimpleLinear(t *testing.T) {
+	var err error
 	nodeconfig := adapters.RandomNodeConfig()
 	addr := network.NewAddrFromNodeId(nodeconfig.Id)
 	_ = p2ptest.NewTestPeerPool()
-	ps := newTestPss(addr.OAddr)
+	ps := newTestPss(addr.Over())
 
 	ping := &pssPing{
 		quitC: make(chan struct{}),
 	}
 
-	err := RegisterPssProtocol(ps, &pssPingTopic, pssPingProtocol, newPssPingProtocol(ping.pssPingHandler))
+	err = RegisterPssProtocol(ps, &pssPingTopic, pssPingProtocol, newPssPingProtocol(ping.pssPingHandler))
 
 	if err != nil {
 		t.Fatalf("Failed to register virtual protocol in pss: %v", err)
 	}
-	pt := p2ptest.NewProtocolTester(t, nodeconfig.Id, 2, ps.Run)
+	run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
+		id := p.ID()
+		bp := &testPssPeer{
+			Peer: protocols.NewPeer(p, rw, pssSpec),
+			addr: network.ToOverlayAddr(id[:]),
+		}
+		ps.Overlay.On(bp)
+		defer ps.Overlay.Off(bp)
+		log.Debug(fmt.Sprintf("%v", ps.Overlay))
+		return bp.Run(ps.handlePssMsg)
+	}
 
-	msg := newPssPingMsg(ps, pssPingProtocol, pssPingTopic, []byte{1, 2, 3})
+	pt := p2ptest.NewProtocolTester(t, nodeconfig.Id, 2, run)
+
+	msg := newPssPingMsg(ps, network.ToOverlayAddr(pt.Ids[0].Bytes()), pssPingProtocol, pssPingTopic, []byte{1, 2, 3})
 
 	exchange := p2ptest.Exchange{
 		Expects: []p2ptest.Expect{
@@ -207,7 +221,10 @@ func TestPssSimpleLinear(t *testing.T) {
 		},
 	}
 
-	pt.TestExchanges(exchange)
+	err = pt.TestExchanges(exchange)
+	if err != nil {
+		t.Fatalf("exchange failed %v", err)
+	}
 }
 
 func TestPssFullRandom10_5_5(t *testing.T) {
@@ -367,7 +384,7 @@ func newServices() adapters.Services {
 			kadparams.MaxRetries = 1000
 			kadparams.RetryExponent = 2
 			kadparams.RetryInterval = 1000000
-			kademlia := network.NewKademlia(addr.OAddr, kadparams)
+			kademlia := network.NewKademlia(addr.Over(), kadparams)
 
 			config := &network.BzzConfig{
 				OverlayAddr:  addr.Over(),