swarm/pss, p2p/simulations/adapters: added pss routing peerpool

p2p/simulations/adapters/inproc.go

@@ -35,8 +35,8 @@ import (
 // SimAdapter is a NodeAdapter which creates in-memory nodes and connects them
 // using an in-memory p2p.MsgReadWriter pipe
 type SimAdapter struct {
-	mtx      sync.RWMutex
+	mtx   sync.RWMutex
-	nodes    map[discover.NodeID]*SimNode
+	nodes map[discover.NodeID]*SimNode
 }
 
 // NewSimAdapter creates a SimAdapter which is capable of running in-memory
@@ -44,7 +44,7 @@ type SimAdapter struct {
 // node is passed to the NewNode function in the NodeConfig)
 func NewSimAdapter(services map[string]ServiceFunc) *SimAdapter {
 	return &SimAdapter{
-		nodes:    make(map[discover.NodeID]*SimNode),
+		nodes: make(map[discover.NodeID]*SimNode),
 	}
 }
 
@@ -67,18 +67,18 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
 	}
 
 	// check the service is valid and initialize it
-/*
+	/*
-	serviceFunc, exists := s.services[config.Service]
+		serviceFunc, exists := s.services[config.Service]
-	if !exists {
+		if !exists {
-		return nil, fmt.Errorf("unknown node service %q", config.Service)
+			return nil, fmt.Errorf("unknown node service %q", config.Service)
-	}
+		}
 
-	node := &SimNode{
+		node := &SimNode{
-		Id:          id,
+			Id:          id,
-		config:      config,
+			config:      config,
-		adapter:     s,
+			adapter:     s,
-		serviceFunc: serviceFunc,
+			serviceFunc: serviceFunc,
-*/
+	*/
 	//serviceFunc, exists := s.services[config.Service]
 
 	//if !exists {
@@ -86,6 +86,12 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
 	//}
 	//service := serviceFunc(id)
 
+	for _, service := range serviceFuncs[config.Service](id, nil) {
+		for _, proto := range service.Protocols() {
+			nodeprotos = append(nodeprotos, proto)
+		}
+	}
+
 	_, err := node.New(&node.Config{
 		P2P: p2p.Config{
 			PrivateKey:      config.PrivateKey,
@@ -100,18 +106,12 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
 		return nil, err
 	}
 
-	for _, service := range serviceFuncs[config.Service](id, nil) {
-		for _, proto := range service.Protocols() {
-			nodeprotos = append(nodeprotos, proto)
-		}
-	}
-	
 	simnode := &SimNode{
-		Id:      id,
+		Id:          id,
 		serviceFunc: serviceFuncs[config.Service],
 		adapter:     s,
 		config:      config,
-		running:	[]node.Service{},
+		running:     []node.Service{},
 	}
 	s.nodes[id.NodeID] = simnode
 	return simnode, nil
@@ -150,11 +150,11 @@ type SimNode struct {
 	Id          *NodeId
 	config      *NodeConfig
 	adapter     *SimAdapter
-	serviceFunc 	ServiceFunc
+	serviceFunc ServiceFunc
 	node        *node.Node
 	client      *rpc.Client
 	rpcMux      *rpcMux
-	running		[]node.Service
+	running     []node.Service
 }
 
 // Addr returns the node's discovery address

swarm/pss/client.go

@@ -18,49 +18,47 @@ import (
 )
 
 const (
-	inboxCapacity = 3000
+	inboxCapacity  = 3000
 	outboxCapacity = 100
-	addrLen = common.HashLength
+	addrLen        = common.HashLength
 )
 
-// implements p2p.Server
-// implements net.Conn
 type PssClient struct {
-	localuri string
+	localuri     string
-	remoteuri  string
+	remoteuri    string
-	ctx context.Context
+	ctx          context.Context
-	cancel func()
+	cancel       func()
 	subscription *rpc.ClientSubscription
-	topicsC chan []byte
+	topicsC      chan []byte
-	msgC chan PssAPIMsg
+	msgC         chan PssAPIMsg
-	quitC chan struct{}
+	quitC        chan struct{}
-	quitting uint32
+	quitting     uint32
-	ws *rpc.Client
+	ws           *rpc.Client
-	lock sync.Mutex
+	lock         sync.Mutex
-	peerPool map[PssTopic]map[pot.Address]*pssRPCRW
+	peerPool     map[PssTopic]map[pot.Address]*pssRPCRW
-	protos []*p2p.Protocol
+	protos       []*p2p.Protocol
 }
 
 type pssRPCRW struct {
 	*PssClient
 	topic *PssTopic
-	spec *protocols.Spec
+	spec  *protocols.Spec
-	msgC chan []byte
+	msgC  chan []byte
-	addr pot.Address
+	addr  pot.Address
 }
 
 func (self *PssClient) newpssRPCRW(addr pot.Address, spec *protocols.Spec, topic *PssTopic) *pssRPCRW {
-	return &pssRPCRW {
+	return &pssRPCRW{
 		PssClient: self,
-		topic: topic,
+		topic:     topic,
-		spec: spec,
+		spec:      spec,
-		msgC: make(chan []byte),
+		msgC:      make(chan []byte),
-		addr: addr,
+		addr:      addr,
 	}
 }
 
 func (rw *pssRPCRW) ReadMsg() (p2p.Msg, error) {
-	msg := <- rw.msgC
+	msg := <-rw.msgC
 	log.Warn("pssrpcrw read", "msg", msg)
 	pmsg, err := ToP2pMsg(msg)
 	if err != nil {
@@ -84,7 +82,7 @@ func (rw *pssRPCRW) WriteMsg(msg p2p.Msg) error {
 
 	return rw.PssClient.ws.CallContext(rw.PssClient.ctx, nil, "pss_sendRaw", rw.topic, PssAPIMsg{
 		Addr: rw.addr.Bytes(),
-		Msg: pmsg,
+		Msg:  pmsg,
 	})
 
 }
@@ -99,14 +97,14 @@ func NewPssClient(ctx context.Context, cancel func(), remotehost string, remotep
 
 	if ctx == nil {
 		ctx = context.Background()
-		cancel = func() {return}
+		cancel = func() { return }
 	}
 	pssc := &PssClient{
-		msgC: make(chan PssAPIMsg),
+		msgC:     make(chan PssAPIMsg),
-		quitC: make(chan struct{}),
+		quitC:    make(chan struct{}),
 		peerPool: make(map[PssTopic]map[pot.Address]*pssRPCRW),
-		ctx: ctx,
+		ctx:      ctx,
-		cancel: cancel,
+		cancel:   cancel,
 	}
 
 	if remotehost == "" {
@@ -163,21 +161,21 @@ func (self *PssClient) RunProtocol(proto *p2p.Protocol, spec *protocols.Spec) er
 	go func() {
 		for {
 			select {
-				case msg := <- msgC:
+			case msg := <-msgC:
-					var addr pot.Address
+				var addr pot.Address
-					copy(addr[:], msg.Addr)
+				copy(addr[:], msg.Addr)
-					if self.peerPool[topic][addr] == nil {
+				if self.peerPool[topic][addr] == nil {
-						self.peerPool[topic][addr] = self.newpssRPCRW(addr, spec, &topic)
+					self.peerPool[topic][addr] = self.newpssRPCRW(addr, spec, &topic)
-						nid, _ := discover.HexID("0x00")
+					nid, _ := discover.HexID("0x00")
-						p := p2p.NewPeer(nid, fmt.Sprintf("%v", addr), []p2p.Cap{})
+					p := p2p.NewPeer(nid, fmt.Sprintf("%v", addr), []p2p.Cap{})
-						go proto.Run(p, self.peerPool[topic][addr])
+					go proto.Run(p, self.peerPool[topic][addr])
-					}
+				}
-					go func() {
+				go func() {
-						self.peerPool[topic][addr].msgC <- msg.Msg
+					self.peerPool[topic][addr].msgC <- msg.Msg
-					}()
+				}()
-				case <-self.quitC:
+			case <-self.quitC:
-					self.shutdown()
+				self.shutdown()
-					return
+				return
 			}
 		}
 	}()

swarm/pss/pss.go

@@ -13,8 +13,8 @@ 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"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
+	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
 	"github.com/ethereum/go-ethereum/pot"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/rpc"
@@ -76,23 +76,22 @@ func (self *PssTopic) String() string {
 
 // Pre-Whisper placeholder, payload of PssMsg
 type PssEnvelope struct {
-	Topic       PssTopic
+	Topic   PssTopic
-	TTL         uint16
+	TTL     uint16
-	Payload     []byte
+	Payload []byte
-	From		[]byte
+	From    []byte
 }
 
 // creates Pss envelope from sender address, topic and raw payload
 func NewPssEnvelope(addr []byte, topic PssTopic, payload []byte) *PssEnvelope {
 	return &PssEnvelope{
-		From: 		 addr,
+		From:    addr,
-		Topic:       topic,
+		Topic:   topic,
-		TTL:         DefaultTTL,
+		TTL:     DefaultTTL,
-		Payload:     payload,
+		Payload: payload,
 	}
 }
 
-
 func (msg *PssMsg) serialize() []byte {
 	rlpdata, _ := rlp.EncodeToBytes(msg)
 	/*buf := bytes.NewBuffer(nil)
@@ -103,7 +102,6 @@ func (msg *PssMsg) serialize() []byte {
 	return rlpdata
 }
 
-
 var pssSpec = &protocols.Spec{
 	Name:       "pss",
 	Version:    1,
@@ -144,13 +142,15 @@ type pssHandler func(msg []byte, p *p2p.Peer, from []byte) error
 // - a dispatcher lookup, mapping protocols to topics
 // - a message cache to spot messages that previously have been forwarded
 type Pss struct {
-	network.Overlay                                                 // we can get the overlayaddress from this
+	network.Overlay // we can get the overlayaddress from this
+	//peerPool map[pot.Address]map[PssTopic]p2p.MsgReadWriter // keep track of all virtual p2p.Peers we are currently speaking to
 	peerPool map[pot.Address]map[PssTopic]p2p.MsgReadWriter // keep track of all virtual p2p.Peers we are currently speaking to
+	fwdPool  map[pot.Address]*protocols.Peer                // keep track of all peers sitting on the pssmsg routing layer
 	handlers map[PssTopic]map[*pssHandler]bool              // topic and version based pss payload handlers
 	fwdcache map[pssDigest]pssCacheEntry                    // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg
 	cachettl time.Duration                                  // how long to keep messages in fwdcache
 	lock     sync.Mutex
-	dpa		*storage.DPA
+	dpa      *storage.DPA
 }
 
 func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
@@ -173,12 +173,13 @@ func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
 // TODO: error check overlay integrity
 func NewPss(k network.Overlay, dpa *storage.DPA, params *PssParams) *Pss {
 	return &Pss{
-		Overlay: k,
+		Overlay:  k,
 		peerPool: make(map[pot.Address]map[PssTopic]p2p.MsgReadWriter, PssPeerCapacity),
+		fwdPool:  make(map[pot.Address]*protocols.Peer),
 		handlers: make(map[PssTopic]map[*pssHandler]bool),
 		fwdcache: make(map[pssDigest]pssCacheEntry),
 		cachettl: params.Cachettl,
-		dpa: dpa,
+		dpa:      dpa,
 	}
 }
 
@@ -203,12 +204,15 @@ func (self *Pss) Protocols() []p2p.Protocol {
 
 func (self *Pss) Run(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 	pp := protocols.NewPeer(p, rw, pssSpec)
+	id := p.ID()
+	h := pot.NewHashAddressFromBytes(network.ToOverlayAddr(id[:]))
+	self.fwdPool[h.Address] = pp
 	return pp.Run(self.handlePssMsg)
 }
 
 func (self *Pss) APIs() []rpc.API {
 	return []rpc.API{
-		rpc.API {
+		rpc.API{
 			Namespace: "pss",
 			Version:   "0.1",
 			Service:   NewPssAPI(self),
@@ -251,7 +255,6 @@ func (self *Pss) deregister(topic *PssTopic, h *pssHandler) {
 // currently not in use as forwarder address is not known in the handler function hooked to the pss dispatcher.
 // it is included as a courtesy to custom transport layers that may want to implement this
 func (self *Pss) AddToCache(addr []byte, msg *PssMsg) error {
-	//digest := self.hashMsg(msg)
 	digest, err := self.storeMsg(msg)
 	if err != nil {
 		return err
@@ -376,17 +379,16 @@ 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.(senderPeer)
+		//p, ok := op.(senderPeer)
-		if !ok {
+		h := pot.NewHashAddressFromBytes(op.Address())
-			return true
+		pp := self.fwdPool[h.Address]
-		}
 		addr := self.Overlay.BaseAddr()
-		sendMsg := fmt.Sprintf("%x: msg to %x via %x", common.ByteLabel(addr), common.ByteLabel(msg.To), common.ByteLabel(p.Address()))
+		sendMsg := fmt.Sprintf("%x: msg to %x via %x", common.ByteLabel(addr), common.ByteLabel(msg.To), common.ByteLabel(op.Address()))
-		if self.checkFwdCache(p.Address(), digest) {
+		if self.checkFwdCache(op.Address(), digest) {
 			log.Info(fmt.Sprintf("%v: peer already forwarded to", sendMsg))
 			return true
 		}
-		err := p.Send(msg)
+		err := pp.Send(msg)
 		if err != nil {
 			log.Warn(fmt.Sprintf("%v: failed forwarding: %v", sendMsg, err))
 			return true
@@ -394,10 +396,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.Address()) || !isproxbin {
+		if bytes.Equal(msg.To, op.Address()) || !isproxbin {
 			return false
 		}
-		log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ByteLabel(p.Address())))
+		log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ByteLabel(op.Address())))
 		return true
 	})
 
@@ -459,7 +461,7 @@ type PssReadWriter struct {
 	To         pot.Address
 	LastActive time.Time
 	rw         chan p2p.Msg
-	spec		*protocols.Spec
+	spec       *protocols.Spec
 	topic      *PssTopic
 }
 
@@ -497,21 +499,20 @@ func (prw PssReadWriter) injectMsg(msg p2p.Msg) error {
 type PssProtocol struct {
 	*Pss
 	proto *p2p.Protocol
-	topic           *PssTopic
+	topic *PssTopic
-	spec			*protocols.Spec
+	spec  *protocols.Spec
 }
 
 // Constructor
 //func RegisterPssProtocol(pss *Pss, topic *PssTopic, spec *protocols.Spec, targetprotocol *p2p.Protocol) *PssProtocol {
 func RegisterPssProtocol(pss *Pss, topic *PssTopic, spec *protocols.Spec, targetprotocol *p2p.Protocol) error {
 	pp := &PssProtocol{
-		Pss:             pss,
+		Pss:   pss,
 		proto: targetprotocol,
-		topic:           topic,
+		topic: topic,
-		spec:			 spec,
+		spec:  spec,
 	}
 	pss.Register(topic, pp.handle)
-	//return pp
 	return nil
 }
 
@@ -522,7 +523,7 @@ func (self *PssProtocol) handle(msg []byte, p *p2p.Peer, senderAddr []byte) erro
 			Pss:   self.Pss,
 			To:    hashoaddr,
 			rw:    make(chan p2p.Msg),
-			spec:	self.spec,
+			spec:  self.spec,
 			topic: self.topic,
 		}
 		self.Pss.AddPeer(p, hashoaddr, self.proto.Run, *self.topic, rw)
@@ -554,8 +555,8 @@ func newProtocolMsg(code uint64, msg interface{}) ([]byte, error) {
 	// therefore we use two separate []byte fields instead of peerAddr
 	// TODO verify that nested structs cannot be used in rlp
 	smsg := &PssProtocolMsg{
-		Code: code,
+		Code:    code,
-		Size: uint32(len(rlpdata)),
+		Size:    uint32(len(rlpdata)),
 		Payload: rlpdata,
 	}
 
@@ -568,14 +569,13 @@ func newProtocolMsg(code uint64, msg interface{}) ([]byte, error) {
 func NewTopic(s string, v int) (topic PssTopic) {
 	h := sha3.NewKeccak256()
 	h.Write([]byte(s))
-	buf := make([]byte, TopicLength / 8)
+	buf := make([]byte, TopicLength/8)
 	binary.PutUvarint(buf, uint64(v))
 	h.Write(buf)
 	copy(topic[:], h.Sum(buf)[:])
 	return topic
 }
 
-
 func ToP2pMsg(msg []byte) (p2p.Msg, error) {
 	payload := &PssProtocolMsg{}
 	if err := rlp.DecodeBytes(msg, payload); err != nil {

swarm/pss/pss_test.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"fmt"
 	"io/ioutil"
+	"math/rand"
 	"os"
 	"testing"
 	"time"
@@ -17,6 +18,7 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/simulations"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
+	"github.com/ethereum/go-ethereum/pot"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
@@ -190,10 +192,13 @@ func TestPssSimpleLinear(t *testing.T) {
 	}
 	run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 		id := p.ID()
+		pp := protocols.NewPeer(p, rw, pssSpec)
 		bp := &testPssPeer{
-			Peer: protocols.NewPeer(p, rw, pssSpec),
+			Peer: pp,
 			addr: network.ToOverlayAddr(id[:]),
 		}
+		h := pot.NewHashAddressFromBytes(bp.addr)
+		ps.fwdPool[h.Address] = pp
 		ps.Overlay.On(bp)
 		defer ps.Overlay.Off(bp)
 		log.Debug(fmt.Sprintf("%v", ps.Overlay))
@@ -244,6 +249,8 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int
 
 	trigger := make(chan *adapters.NodeId)
 	ids := make([]*adapters.NodeId, nodeCount)
+	fullids := ids[0:fullnodecount]
+	fullpeers := [][]byte{}
 
 	for i := 0; i < nodeCount; i++ {
 		nodeconfig := adapters.RandomNodeConfig()
@@ -261,6 +268,9 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int
 			t.Fatal("error triggering checks for node %s: %s", node.ID().Label(), err)
 		}
 		ids[i] = node.ID()
+		if i < fullnodecount {
+			fullpeers = append(fullpeers, network.ToOverlayAddr(node.ID().Bytes()))
+		}
 	}
 
 	// run a simulation which connects the 10 nodes in a ring and waits
@@ -295,14 +305,21 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int
 			return false, fmt.Errorf("error getting node client: %s", err)
 		}
 
-		log.Debug("in check", "node", id)
+		for _, fid := range fullids {
-
+			if fid == id {
-		if lastid != nil {
+				fpeeridx := rand.Int() % (fullnodecount - 1)
-			//msg := pssPingMsg{Created: time.Now(),}
+				log.Debug(fmt.Sprintf("fpeeridx %d, fpeer len %d", fpeeridx, len(fullpeers)))
-			client.CallContext(context.Background(), nil, "pss_sendRaw", pssPingTopic, PssAPIMsg{
+				if bytes.Equal(fullpeers[fpeeridx], network.ToOverlayAddr(fid.Bytes())) {
-				Addr: lastid.Bytes(),
+					fpeeridx++
-				Msg:  []byte{1, 2, 3},
+				}
-			})
+				msg := pssPingMsg{Created: time.Now()}
+				code, _ := pssPingProtocol.GetCode(&pssPingMsg{})
+				pmsg, _ := newProtocolMsg(code, msg)
+				client.CallContext(context.Background(), nil, "pss_sendRaw", pssPingTopic, PssAPIMsg{
+					Addr: fullpeers[fpeeridx],
+					Msg:  pmsg,
+				})
+			}
 		}
 		lastid = id
 
@@ -325,6 +342,36 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int
 		t.Fatalf("simulation failed: %s", result.Error)
 	}
 
+	trigger = make(chan *adapters.NodeId)
+
+	action = func(ctx context.Context) error {
+		return nil
+	}
+	check = func(ctx context.Context, id *adapters.NodeId) (bool, error) {
+		select {
+		case <-ctx.Done():
+			return false, ctx.Err()
+		default:
+		}
+
+		return true, nil
+	}
+	timeout = 5 * time.Second
+	ctx, cancel = context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	result = simulations.NewSimulation(net).Run(ctx, &simulations.Step{
+		Action:  action,
+		Trigger: trigger,
+		Expect: &simulations.Expectation{
+			Nodes: fullids,
+			Check: check,
+		},
+	})
+	if result.Error != nil {
+		t.Fatalf("simulation failed: %s", result.Error)
+	}
+
 	t.Log("Simulation Passed:")
 	t.Logf("Duration: %s", result.FinishedAt.Sub(result.StartedAt))
 
@@ -334,40 +381,57 @@ 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 {
+
+	gotpeer := make(map[*adapters.NodeId]bool)
+
 	node := net.GetNode(id)
 	if node == nil {
 		return fmt.Errorf("unknown node: %s", id)
 	}
+	client, err := node.Client()
+	if err != nil {
+		return err
+	}
+
+	peerevents := make(chan *p2p.PeerEvent)
+	peersub, err := client.Subscribe(context.Background(), "admin", peerevents, "peerEvents")
+	if err != nil {
+		return fmt.Errorf("error getting peer events for node %v: %s", id, err)
+	}
+
+	msgevents := make(chan PssAPIMsg)
+	msgsub, err := client.Subscribe(context.Background(), "pss", msgevents, "newMsg", pssPingTopic)
+	if err != nil {
+		return fmt.Errorf("error getting peer events for node %v: %s", id, err)
+	}
+
 	go func() {
-		time.Sleep(time.Second)
+		defer msgsub.Unsubscribe()
-		trigger <- id
+		defer peersub.Unsubscribe()
-	}()
+		for {
-	/*
+			select {
-		client, err := node.Client()
+			case event := <-peerevents:
-		if err != nil {
+				nid := adapters.NewNodeId(event.Peer[:])
-			return err
+				if event.Type == "add" && !gotpeer[nid] {
-		}
-		events := make(chan PssAPIMsg)
-		sub, err := client.Subscribe(context.Background(), "pss", events, "newMsg", topic)
-		if err != nil {
-			return fmt.Errorf("error getting peer events for node %v: %s", id, err)
-		}
-		go func() {
-			defer sub.Unsubscribe()
-			for {
-				select {
-				case msg := <-events:
-					log.Warn("pss rpc got msg", "msg", msg)
 					trigger <- id
-				case err := <-sub.Err():
+					gotpeer[nid] = true
-					if err != nil {
-						log.Error(fmt.Sprintf("error getting peer events for node %v", id), "err", err)
-					}
-					return
 				}
+			case <-msgevents:
+				trigger <- id
+			case err := <-peersub.Err():
+				if err != nil {
+					log.Error(fmt.Sprintf("error getting peer events for node %v", id), "err", err)
+				}
+				return
+
+			case err := <-msgsub.Err():
+				if err != nil {
+					log.Error(fmt.Sprintf("error getting msg for node %v", id), "err", err)
+				}
+				return
 			}
-		}()
+		}
-	*/
+	}()
 	return nil
 }
 
@@ -404,9 +468,20 @@ func newServices() adapters.Services {
 				log.Error("local dpa creation failed", "error", err)
 				return nil
 			}
-			pssp := NewPssParams()
 
-			return []node.Service{network.NewBzz(config, kademlia, adapters.NewSimStateStore()), NewPss(kademlia, dpa, pssp)}
+			pssp := NewPssParams()
+			ps := NewPss(kademlia, dpa, pssp)
+
+			ping := &pssPing{
+				quitC: make(chan struct{}),
+			}
+			err = RegisterPssProtocol(ps, &pssPingTopic, pssPingProtocol, newPssPingProtocol(ping.pssPingHandler))
+			if err != nil {
+				log.Error("Couldnt register pss protocol", "err", err)
+				os.Exit(1)
+			}
+
+			return []node.Service{network.NewBzz(config, kademlia, adapters.NewSimStateStore()), ps}
 		},
 	}
 }

swarm/pss/pssapi.go

@@ -36,11 +36,11 @@ func (pssapi *PssAPI) NewMsg(ctx context.Context, topic PssTopic) (*rpc.Subscrip
 	psssub := notifier.CreateSubscription()
 	handler := func(msg []byte, p *p2p.Peer, from []byte) error {
 		apimsg := &PssAPIMsg{
-			Msg: msg,
+			Msg:  msg,
 			Addr: from,
 		}
 		if err := notifier.Notify(psssub.ID, apimsg); err != nil {
-				log.Warn(fmt.Sprintf("notification on pss sub topic %v rpc (sub %v) msg %v failed!", topic, psssub.ID, msg))
+			log.Warn(fmt.Sprintf("notification on pss sub topic %v rpc (sub %v) msg %v failed!", topic, psssub.ID, msg))
 		}
 		return nil
 	}
@@ -60,7 +60,7 @@ func (pssapi *PssAPI) NewMsg(ctx context.Context, topic PssTopic) (*rpc.Subscrip
 	return psssub, nil
 }
 
-// SendRaw sends the message (serialised into byte slice) to a peer with topic
+// SendRaw sends the message (serialized into byte slice) to a peer with topic
 func (pssapi *PssAPI) SendRaw(topic PssTopic, msg PssAPIMsg) error {
 	err := pssapi.Pss.Send(msg.Addr, topic, msg.Msg)
 	if err != nil {