swarm, swarm/network, swarm/pss, les: fixed swarm bin + minor adj

removed all Pss in symbol names whereever possible (and sensible)

discovery switch in hive now prevents add of suggested peers
this is probably not a permanent solution

bzz and pss now started from swarm.go,
Outputs from bzz/pss Protocols() and APIs()
are appended to the Swarm methods

les/handler.go

@@ -259,7 +259,7 @@ func (pm *ProtocolManager) removePeer(id string) {
 func (pm *ProtocolManager) Start(srvr *p2p.Server) {
 	var topicDisc *discv5.Network
 	if srvr != nil {
-		topicDisc = srvr.DiscV5
+		topicDisc = srvr.DiscV5()
 	}
 	lesTopic := discv5.Topic("LES@" + common.Bytes2Hex(pm.blockchain.Genesis().Hash().Bytes()[0:8]))
 	if pm.lightSync {

les/serverpool.go

@@ -143,7 +143,7 @@ func newServerPool(db ethdb.Database, dbPrefix []byte, server *p2p.Server, topic
 		pool.discSetPeriod = make(chan time.Duration, 1)
 		pool.discNodes = make(chan *discv5.Node, 100)
 		pool.discLookups = make(chan bool, 100)
-		go pool.server.DiscV5.SearchTopic(topic, pool.discSetPeriod, pool.discNodes, pool.discLookups)
+		go pool.server.DiscV5().SearchTopic(topic, pool.discSetPeriod, pool.discNodes, pool.discLookups)
 	}
 
 	go pool.eventLoop()

swarm/network/hive.go

@@ -127,23 +127,23 @@ func (self *Hive) Start(server *p2p.Server) error {
 			log.Debug("hive delegate to overlay driver: suggest addr to connect to")
 			// log.Trace("hive delegate to overlay driver: suggest addr to connect to")
 			addr, order, want := self.SuggestPeer()
-
+			if self.Discovery {
-			if addr != nil {
+				if addr != nil {
-				log.Info(fmt.Sprintf("========> connect to bee %v", addr))
+					log.Info(fmt.Sprintf("========> connect to bee %v", addr))
-				under, err := discover.ParseNode(string(addr.(Addr).Under()))
+					under, err := discover.ParseNode(string(addr.(Addr).Under()))
-				if err == nil {
+					if err == nil {
-					server.AddPeer(under)
+						server.AddPeer(under)
+					} else {
+						log.Error(fmt.Sprintf("===X====> connect to bee %v failed: invalid node URL: %v", addr, err))
+					}
 				} else {
-					log.Error(fmt.Sprintf("===X====> connect to bee %v failed: invalid node URL: %v", addr, err))
+					log.Trace("cannot suggest peers")
 				}
-			} else {
-				log.Trace("cannot suggest peers")
-			}
 
-			want = want && self.Discovery
+				if want {
-			if want {
+					log.Debug(fmt.Sprintf("========> request peers nearest %v", addr))
-				log.Debug(fmt.Sprintf("========> request peers nearest %v", addr))
+					RequestOrder(self.Overlay, uint8(order), self.PeersBroadcastSetSize, self.MaxPeersPerRequest)
-				RequestOrder(self.Overlay, uint8(order), self.PeersBroadcastSetSize, self.MaxPeersPerRequest)
+				}
 			}
 
 			log.Info(fmt.Sprintf("%v", self))

swarm/pss/client/client.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"sync"
-	"sync/atomic"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/event"
@@ -29,50 +28,49 @@ const (
 // RemotePort: port of node running websockets proxy to pss (0 = go-ethereum node default)
 // Secure: whether or not to use secure connection
 // SelfHost: local if host to connect from
-type PssClientConfig struct {
+type ClientConfig struct {
 	SelfHost   string
 	RemoteHost string
 	RemotePort int
 	Secure     bool
 }
 
-func NewPssClientConfig() *PssClientConfig {
+func NewClientConfig() *ClientConfig {
-	return &PssClientConfig{
+	return &ClientConfig{
 		SelfHost:   "localhost",
 		RemoteHost: "localhost",
 		RemotePort: 8546,
 	}
 }
 
-type PssClient struct {
+type Client struct {
 	localuri     string
 	remoteuri    string
 	ctx          context.Context
 	cancel       func()
 	subscription *rpc.ClientSubscription
 	topicsC      chan []byte
-	msgC         chan pss.PssAPIMsg
+	msgC         chan pss.APIMsg
 	quitC        chan struct{}
-	quitting     uint32
 	ws           *rpc.Client
 	lock         sync.Mutex
-	peerPool     map[pss.PssTopic]map[pot.Address]*pssRPCRW
+	peerPool     map[pss.Topic]map[pot.Address]*pssRPCRW
-	protos       map[pss.PssTopic]*p2p.Protocol
+	protos       map[pss.Topic]*p2p.Protocol
 }
 
 type pssRPCRW struct {
-	*PssClient
+	*Client
-	topic *pss.PssTopic
+	topic *pss.Topic
 	msgC  chan []byte
 	addr  pot.Address
 }
 
-func (self *PssClient) newpssRPCRW(addr pot.Address, topic *pss.PssTopic) *pssRPCRW {
+func (self *Client) newpssRPCRW(addr pot.Address, topic *pss.Topic) *pssRPCRW {
 	return &pssRPCRW{
-		PssClient: self,
+		Client: self,
-		topic:     topic,
+		topic:  topic,
-		msgC:      make(chan []byte),
+		msgC:   make(chan []byte),
-		addr:      addr,
+		addr:   addr,
 	}
 }
 
@@ -91,7 +89,7 @@ func (rw *pssRPCRW) WriteMsg(msg p2p.Msg) error {
 	log.Trace("got writemsg pssclient", "msg", msg)
 	rlpdata := make([]byte, msg.Size)
 	msg.Payload.Read(rlpdata)
-	pmsg, err := rlp.EncodeToBytes(pss.PssProtocolMsg{
+	pmsg, err := rlp.EncodeToBytes(pss.ProtocolMsg{
 		Code:    msg.Code,
 		Size:    msg.Size,
 		Payload: rlpdata,
@@ -99,13 +97,13 @@ func (rw *pssRPCRW) WriteMsg(msg p2p.Msg) error {
 	if err != nil {
 		return err
 	}
-	return rw.PssClient.ws.CallContext(rw.PssClient.ctx, nil, "pss_sendPss", rw.topic, pss.PssAPIMsg{
+	return rw.Client.ws.CallContext(rw.Client.ctx, nil, "pss_send", rw.topic, pss.APIMsg{
 		Addr: rw.addr.Bytes(),
 		Msg:  pmsg,
 	})
 }
 
-func NewPssClient(ctx context.Context, cancel func(), config *PssClientConfig) *PssClient {
+func NewClient(ctx context.Context, cancel func(), config *ClientConfig) *Client {
 	prefix := "ws"
 
 	if ctx == nil {
@@ -115,27 +113,15 @@ func NewPssClient(ctx context.Context, cancel func(), config *PssClientConfig) *
 		cancel = func() { return }
 	}
 
-	pssc := &PssClient{
+	pssc := &Client{
-		msgC:     make(chan pss.PssAPIMsg),
+		msgC:     make(chan pss.APIMsg),
 		quitC:    make(chan struct{}),
-		peerPool: make(map[pss.PssTopic]map[pot.Address]*pssRPCRW),
+		peerPool: make(map[pss.Topic]map[pot.Address]*pssRPCRW),
-		protos:   make(map[pss.PssTopic]*p2p.Protocol),
+		protos:   make(map[pss.Topic]*p2p.Protocol),
 		ctx:      ctx,
 		cancel:   cancel,
 	}
 
-	if config.RemoteHost == "" {
-		config.RemoteHost = "localhost"
-	}
-
-	if config.RemotePort == 0 {
-		config.RemotePort = defaultWSHost
-	}
-
-	if config.SelfHost == "" {
-		config.SelfHost = "localhost"
-	}
-
 	if config.Secure {
 		prefix = "wss"
 	}
@@ -146,23 +132,22 @@ func NewPssClient(ctx context.Context, cancel func(), config *PssClientConfig) *
 	return pssc
 }
 
-func NewPssClientWithRPC(ctx context.Context, client *rpc.Client) *PssClient {
+func NewClientWithRPC(ctx context.Context, client *rpc.Client) *Client {
-	return &PssClient{
+	return &Client{
-		msgC:     make(chan pss.PssAPIMsg),
+		msgC:     make(chan pss.APIMsg),
 		quitC:    make(chan struct{}),
-		peerPool: make(map[pss.PssTopic]map[pot.Address]*pssRPCRW),
+		peerPool: make(map[pss.Topic]map[pot.Address]*pssRPCRW),
-		protos:   make(map[pss.PssTopic]*p2p.Protocol),
+		protos:   make(map[pss.Topic]*p2p.Protocol),
 		ws:       client,
 		ctx:      ctx,
 	}
 }
 
-func (self *PssClient) shutdown() {
+func (self *Client) shutdown() {
-	atomic.StoreUint32(&self.quitting, 1)
 	self.cancel()
 }
 
-func (self *PssClient) Start() error {
+func (self *Client) Start() error {
 	if self.ws != nil {
 		return nil
 	}
@@ -177,11 +162,11 @@ func (self *PssClient) Start() error {
 	return nil
 }
 
-func (self *PssClient) RunProtocol(proto *p2p.Protocol) error {
+func (self *Client) RunProtocol(proto *p2p.Protocol) error {
 	topic := pss.NewTopic(proto.Name, int(proto.Version))
-	msgC := make(chan pss.PssAPIMsg)
+	msgC := make(chan pss.APIMsg)
 	self.peerPool[topic] = make(map[pot.Address]*pssRPCRW)
-	sub, err := self.ws.Subscribe(self.ctx, "pss", msgC, "receivePss", topic)
+	sub, err := self.ws.Subscribe(self.ctx, "pss", msgC, "receive", topic)
 	if err != nil {
 		return fmt.Errorf("pss event subscription failed: %v", err)
 	}
@@ -215,12 +200,12 @@ func (self *PssClient) RunProtocol(proto *p2p.Protocol) error {
 	return nil
 }
 
-func (self *PssClient) Stop() error {
+func (self *Client) Stop() error {
 	self.cancel()
 	return nil
 }
 
-func (self *PssClient) AddPssPeer(addr pot.Address, spec *protocols.Spec) {
+func (self *Client) AddPssPeer(addr pot.Address, spec *protocols.Spec) {
 	topic := pss.NewTopic(spec.Name, int(spec.Version))
 	if self.peerPool[topic][addr] == nil {
 		self.peerPool[topic][addr] = self.newpssRPCRW(addr, &topic)
@@ -230,31 +215,31 @@ func (self *PssClient) AddPssPeer(addr pot.Address, spec *protocols.Spec) {
 	}
 }
 
-func (self *PssClient) RemovePssPeer(addr pot.Address, spec *protocols.Spec) {
+func (self *Client) RemovePssPeer(addr pot.Address, spec *protocols.Spec) {
 	topic := pss.NewTopic(spec.Name, int(spec.Version))
 	delete(self.peerPool[topic], addr)
 }
 
-func (self *PssClient) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription {
+func (self *Client) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription {
 	log.Error("PSS client handles events internally, use the read functions instead")
 	return nil
 }
 
-func (self *PssClient) PeerCount() int {
+func (self *Client) PeerCount() int {
 	return len(self.peerPool)
 }
 
-func (self *PssClient) NodeInfo() *p2p.NodeInfo {
+func (self *Client) NodeInfo() *p2p.NodeInfo {
 	return nil
 }
 
-func (self *PssClient) PeersInfo() []*p2p.PeerInfo {
+func (self *Client) PeersInfo() []*p2p.PeerInfo {
 	return nil
 }
-func (self *PssClient) AddPeer(node *discover.Node) {
+func (self *Client) AddPeer(node *discover.Node) {
 	log.Error("Cannot add peer in PSS with discover.Node, need swarm overlay address")
 }
 
-func (self *PssClient) RemovePeer(node *discover.Node) {
+func (self *Client) RemovePeer(node *discover.Node) {
 	log.Error("Cannot remove peer in PSS with discover.Node, need swarm overlay address")
 }

swarm/pss/client/client_test.go

@@ -26,8 +26,8 @@ func init() {
 func TestRunProtocol(t *testing.T) {
 	quitC := make(chan struct{})
 	ps := pss.NewTestPss(nil)
-	ping := &pss.PssPing{
+	ping := &pss.Ping{
-		QuitC: make(chan struct{}),
+		C: make(chan struct{}),
 	}
 	proto := newProtocol(ping)
 	_, err := baseTester(t, proto, ps, nil, nil, quitC)
@@ -42,8 +42,8 @@ func TestIncoming(t *testing.T) {
 	ps := pss.NewTestPss(nil)
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
 	var addr []byte
-	ping := &pss.PssPing{
+	ping := &pss.Ping{
-		QuitC: make(chan struct{}),
+		C: make(chan struct{}),
 	}
 	proto := newProtocol(ping)
 	client, err := baseTester(t, proto, ps, ctx, cancel, quitC)
@@ -53,37 +53,28 @@ func TestIncoming(t *testing.T) {
 
 	client.ws.Call(&addr, "psstest_baseAddr")
 
-	code, _ := pss.PssPingProtocol.GetCode(&pss.PssPingMsg{})
+	code, _ := pss.PingProtocol.GetCode(&pss.PingMsg{})
-	rlpbundle, err := pss.NewProtocolMsg(code, &pss.PssPingMsg{
+	rlpbundle, err := pss.NewProtocolMsg(code, &pss.PingMsg{
 		Created: time.Now(),
 	})
 	if err != nil {
-		t.Fatalf("couldn't make pssmsg")
+		t.Fatalf("couldn't make pssmsg: %v", err)
 	}
 
-	pssenv := pss.PssEnvelope{
+	pssenv := pss.NewEnvelope(addr, pss.NewTopic(proto.Name, int(proto.Version)), rlpbundle)
-		From:    addr,
-		Topic:   pss.NewTopic(proto.Name, int(proto.Version)),
-		TTL:     pss.DefaultTTL,
-		Payload: rlpbundle,
-	}
 	pssmsg := pss.PssMsg{
 		To:      addr,
-		Payload: &pssenv,
+		Payload: pssenv,
 	}
 
 	ps.Process(&pssmsg)
 
-	go func() {
-		<-ping.QuitC
-		client.cancel()
-	}()
-
 	select {
 	case <-client.ctx.Done():
 		t.Fatalf("outgoing timed out or canceled")
-	default:
+	case <-ping.C:
 	}
+
 	quitC <- struct{}{}
 }
 
@@ -94,8 +85,8 @@ func TestOutgoing(t *testing.T) {
 	var addr []byte
 	var potaddr pot.Address
 
-	ping := &pss.PssPing{
+	ping := &pss.Ping{
-		QuitC: make(chan struct{}),
+		C: make(chan struct{}),
 	}
 	proto := newProtocol(ping)
 	client, err := baseTester(t, proto, ps, ctx, cancel, quitC)
@@ -106,25 +97,25 @@ func TestOutgoing(t *testing.T) {
 	client.ws.Call(&addr, "psstest_baseAddr")
 	copy(potaddr[:], addr)
 
-	msg := &pss.PssPingMsg{
+	msg := &pss.PingMsg{
 		Created: time.Now(),
 	}
 
-	topic := pss.NewTopic(pss.PssPingProtocol.Name, int(pss.PssPingProtocol.Version))
+	topic := pss.NewTopic(pss.PingProtocol.Name, int(pss.PingProtocol.Version))
-	client.AddPssPeer(potaddr, pss.PssPingProtocol)
+	client.AddPssPeer(potaddr, pss.PingProtocol)
 	nid, _ := discover.HexID("0x00")
 	p := p2p.NewPeer(nid, fmt.Sprintf("%v", potaddr), []p2p.Cap{})
-	pp := protocols.NewPeer(p, client.peerPool[topic][potaddr], pss.PssPingProtocol)
+	pp := protocols.NewPeer(p, client.peerPool[topic][potaddr], pss.PingProtocol)
 	pp.Send(msg)
 	select {
 	case <-client.ctx.Done():
 		t.Fatalf("outgoing timed out or canceled")
-	default:
+	case <-ping.C:
 	}
 	quitC <- struct{}{}
 }
 
-func baseTester(t *testing.T, proto *p2p.Protocol, ps *pss.Pss, ctx context.Context, cancel func(), quitC chan struct{}) (*PssClient, error) {
+func baseTester(t *testing.T, proto *p2p.Protocol, ps *pss.Pss, ctx context.Context, cancel func(), quitC chan struct{}) (*Client, error) {
 	var err error
 
 	client := newClient(t, ctx, cancel, quitC)
@@ -143,30 +134,30 @@ func baseTester(t *testing.T, proto *p2p.Protocol, ps *pss.Pss, ctx context.Cont
 	return client, nil
 }
 
-func newProtocol(ping *pss.PssPing) *p2p.Protocol {
+func newProtocol(ping *pss.Ping) *p2p.Protocol {
 
 	return &p2p.Protocol{
-		Name:    pss.PssPingProtocol.Name,
+		Name:    pss.PingProtocol.Name,
-		Version: pss.PssPingProtocol.Version,
+		Version: pss.PingProtocol.Version,
 		Length:  1,
 		Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-			pp := protocols.NewPeer(p, rw, pss.PssPingProtocol)
+			pp := protocols.NewPeer(p, rw, pss.PingProtocol)
-			pp.Run(ping.PssPingHandler)
+			pp.Run(ping.PingHandler)
 			return nil
 		},
 	}
 }
 
-func newClient(t *testing.T, ctx context.Context, cancel func(), quitC chan struct{}) *PssClient {
+func newClient(t *testing.T, ctx context.Context, cancel func(), quitC chan struct{}) *Client {
 
-	conf := NewPssClientConfig()
+	conf := NewClientConfig()
 
-	pssclient := NewPssClient(ctx, cancel, conf)
+	pssclient := NewClient(ctx, cancel, conf)
 
 	ps := pss.NewTestPss(nil)
 	srv := rpc.NewServer()
-	srv.RegisterName("pss", pss.NewPssAPI(ps))
+	srv.RegisterName("pss", pss.NewAPI(ps))
-	srv.RegisterName("psstest", pss.NewPssAPITest(ps))
+	srv.RegisterName("psstest", pss.NewAPITest(ps))
 	ws := srv.WebsocketHandler([]string{"*"})
 	uri := fmt.Sprintf("%s:%d", "localhost", 8546)
 

swarm/pss/common.go

@@ -13,33 +13,33 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
-type PssPingMsg struct {
+type PingMsg struct {
 	Created time.Time
 }
 
-type PssPing struct {
+type Ping struct {
-	QuitC chan struct{}
+	C chan struct{}
 }
 
-func (self *PssPing) PssPingHandler(msg interface{}) error {
+func (self *Ping) PingHandler(msg interface{}) error {
 	log.Warn("got ping", "msg", msg)
-	self.QuitC <- struct{}{}
+	self.C <- struct{}{}
 	return nil
 }
 
-var PssPingProtocol = &protocols.Spec{
+var PingProtocol = &protocols.Spec{
 	Name:       "psstest",
 	Version:    1,
 	MaxMsgSize: 10 * 1024 * 1024,
 	Messages: []interface{}{
-		PssPingMsg{},
+		PingMsg{},
 	},
 }
 
-var PssPingTopic = NewTopic(PssPingProtocol.Name, int(PssPingProtocol.Version))
+var PingTopic = NewTopic(PingProtocol.Name, int(PingProtocol.Version))
 
-func NewPssPingMsg(to []byte, spec *protocols.Spec, topic PssTopic, senderaddr []byte) PssMsg {
+func NewPingMsg(to []byte, spec *protocols.Spec, topic Topic, senderaddr []byte) PssMsg {
-	data := PssPingMsg{
+	data := PingMsg{
 		Created: time.Now(),
 	}
 	code, found := spec.GetCode(&data)
@@ -54,19 +54,19 @@ func NewPssPingMsg(to []byte, spec *protocols.Spec, topic PssTopic, senderaddr [
 
 	pssmsg := PssMsg{
 		To:      to,
-		Payload: NewPssEnvelope(senderaddr, topic, rlpbundle),
+		Payload: NewEnvelope(senderaddr, topic, rlpbundle),
 	}
 
 	return pssmsg
 }
 
-func NewPssPingProtocol(handler func(interface{}) error) *p2p.Protocol {
+func NewPingProtocol(handler func(interface{}) error) *p2p.Protocol {
 	return &p2p.Protocol{
-		Name:    PssPingProtocol.Name,
+		Name:    PingProtocol.Name,
-		Version: PssPingProtocol.Version,
+		Version: PingProtocol.Version,
-		Length:  uint64(PssPingProtocol.MaxMsgSize),
+		Length:  uint64(PingProtocol.MaxMsgSize),
 		Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-			pp := protocols.NewPeer(p, rw, PssPingProtocol)
+			pp := protocols.NewPeer(p, rw, PingProtocol)
 			log.Trace(fmt.Sprintf("running pss vprotocol on peer %v", p))
 			err := pp.Run(handler)
 			return err

swarm/pss/pss.go

@@ -2,14 +2,12 @@ package pss
 
 import (
 	"bytes"
-	//"encoding/binary"
 	"errors"
 	"fmt"
 	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
-	//"github.com/ethereum/go-ethereum/crypto/sha3"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/discover"
@@ -54,6 +52,8 @@ type pssCacheEntry struct {
 
 type pssDigest [digestLength]byte
 
+// implements node.Service
+//
 // pss provides sending messages to nodes without having to be directly connected to them.
 //
 // The messages are wrapped in a PssMsg structure and routed using the swarm kademlia routing.
@@ -67,15 +67,15 @@ type pssDigest [digestLength]byte
 // - 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[Topic]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
+	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
+	handlers        map[Topic]map[*Handler]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
+	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
+	cachettl        time.Duration                               // how long to keep messages in fwdcache
 	lock            sync.Mutex
 	dpa             *storage.DPA
-	debug		bool
+	debug           bool
 }
 
 func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
@@ -94,18 +94,16 @@ func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
 }
 
 // Creates a new Pss instance. A node should only need one of these
-//
-// TODO: error check overlay integrity
 func NewPss(k network.Overlay, dpa *storage.DPA, params *PssParams) *Pss {
 	return &Pss{
 		Overlay:  k,
-		peerPool: make(map[pot.Address]map[PssTopic]p2p.MsgReadWriter, PssPeerCapacity),
+		peerPool: make(map[pot.Address]map[Topic]p2p.MsgReadWriter, PssPeerCapacity),
 		fwdPool:  make(map[pot.Address]*protocols.Peer),
-		handlers: make(map[PssTopic]map[*PssHandler]bool),
+		handlers: make(map[Topic]map[*Handler]bool),
 		fwdcache: make(map[pssDigest]pssCacheEntry),
 		cachettl: params.Cachettl,
 		dpa:      dpa,
-		debug:	params.Debug,
+		debug:    params.Debug,
 	}
 }
 
@@ -145,7 +143,7 @@ func (self *Pss) APIs() []rpc.API {
 		rpc.API{
 			Namespace: "pss",
 			Version:   "0.1",
-			Service:   NewPssAPI(self),
+			Service:   NewAPI(self),
 			Public:    true,
 		},
 	}
@@ -153,32 +151,32 @@ func (self *Pss) APIs() []rpc.API {
 		apis = append(apis, rpc.API{
 			Namespace: "pss",
 			Version:   "0.1",
-			Service:   NewPssAPITest(self),
+			Service:   NewAPITest(self),
 			Public:    true,
 		})
 	}
 	return apis
 }
 
-// Takes the generated PssTopic of a protocol/chatroom etc, and links a handler function to it
+// Takes the generated Topic of a protocol/chatroom etc, and links a handler function to it
 // This allows the implementer to retrieve the right handler functions (invoke the right protocol)
 // for an incoming message by inspecting the topic on it.
 // a topic allows for multiple handlers
 // returns a deregister function which needs to be called to deregister the handler
 // (similar to event.Subscription.Unsubscribe())
-func (self *Pss) Register(topic *PssTopic, handler PssHandler) func() {
+func (self *Pss) Register(topic *Topic, handler Handler) func() {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	handlers := self.handlers[*topic]
 	if handlers == nil {
-		handlers = make(map[*PssHandler]bool)
+		handlers = make(map[*Handler]bool)
 		self.handlers[*topic] = handlers
 	}
 	handlers[&handler] = true
 	return func() { self.deregister(topic, &handler) }
 }
 
-func (self *Pss) deregister(topic *PssTopic, h *PssHandler) {
+func (self *Pss) deregister(topic *Topic, h *Handler) {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	handlers := self.handlers[*topic]
@@ -243,13 +241,12 @@ func (self *Pss) checkFwdCache(addr []byte, digest pssDigest) bool {
 	return false
 }
 
-func (self *Pss) getHandlers(topic PssTopic) map[*PssHandler]bool {
+func (self *Pss) getHandlers(topic Topic) map[*Handler]bool {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	return self.handlers[topic]
 }
 
-//
 func (self *Pss) handlePssMsg(msg interface{}) error {
 	pssmsg := msg.(*PssMsg)
 
@@ -280,12 +277,12 @@ func (self *Pss) Process(pssmsg *PssMsg) error {
 	return nil
 }
 
-// Sends a message using  The message could be anything at all, and will be handled by whichever handler function is mapped to PssTopic using *Pss.Register()
+// Sends a message using  The message could be anything at all, and will be handled by whichever handler function is mapped to Topic using *Pss.Register()
 //
 // The to address is a swarm overlay address
-func (self *Pss) Send(to []byte, topic PssTopic, msg []byte) error {
+func (self *Pss) SendRaw(to []byte, topic Topic, msg []byte) error {
 	sender := self.Overlay.BaseAddr()
-	pssenv := NewPssEnvelope(sender, topic, msg)
+	pssenv := NewEnvelope(sender, topic, msg)
 	pssmsg := &PssMsg{
 		To:      to,
 		Payload: pssenv,
@@ -354,7 +351,7 @@ func (self *Pss) Forward(msg *PssMsg) error {
 // Links a pss peer address and topic to a dedicated p2p.MsgReadWriter in the pss peerpool, and runs the specificed protocol on this p2p.MsgReadWriter and the specified peer
 //
 // The effect is that now we have a "virtual" protocol running on an artificial p2p.Peer, which can be looked up and piped to through Pss using swarm overlay address and topic
-func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, run func(*p2p.Peer, p2p.MsgReadWriter) error, topic PssTopic, rw p2p.MsgReadWriter) error {
+func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, run func(*p2p.Peer, p2p.MsgReadWriter) error, topic Topic, rw p2p.MsgReadWriter) error {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	self.addPeerTopic(addr, topic, rw)
@@ -366,15 +363,15 @@ func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, run func(*p2p.Peer, p2p.
 	return nil
 }
 
-func (self *Pss) addPeerTopic(id pot.Address, topic PssTopic, rw p2p.MsgReadWriter) error {
+func (self *Pss) addPeerTopic(id pot.Address, topic Topic, rw p2p.MsgReadWriter) error {
 	if self.peerPool[id] == nil {
-		self.peerPool[id] = make(map[PssTopic]p2p.MsgReadWriter, PssPeerTopicDefaultCapacity)
+		self.peerPool[id] = make(map[Topic]p2p.MsgReadWriter, PssPeerTopicDefaultCapacity)
 	}
 	self.peerPool[id][topic] = rw
 	return nil
 }
 
-func (self *Pss) removePeerTopic(rw p2p.MsgReadWriter, topic PssTopic) {
+func (self *Pss) removePeerTopic(rw p2p.MsgReadWriter, topic Topic) {
 	prw, ok := rw.(*PssReadWriter)
 	if !ok {
 		return
@@ -389,7 +386,10 @@ func (self *Pss) isSelfRecipient(msg *PssMsg) bool {
 	return bytes.Equal(msg.To, self.Overlay.BaseAddr())
 }
 
-func (self *Pss) isActive(id pot.Address, topic PssTopic) bool {
+func (self *Pss) isActive(id pot.Address, topic Topic) bool {
+	if self.peerPool[id] == nil {
+		return false
+	}
 	return self.peerPool[id][topic] != nil
 }
 
@@ -405,7 +405,7 @@ type PssReadWriter struct {
 	LastActive time.Time
 	rw         chan p2p.Msg
 	spec       *protocols.Spec
-	topic      *PssTopic
+	topic      *Topic
 }
 
 // Implements p2p.MsgReader
@@ -420,7 +420,7 @@ func (prw PssReadWriter) WriteMsg(msg p2p.Msg) error {
 	log.Warn("got writemsg pssclient", "msg", msg)
 	rlpdata := make([]byte, msg.Size)
 	msg.Payload.Read(rlpdata)
-	pmsg, err := rlp.EncodeToBytes(PssProtocolMsg{
+	pmsg, err := rlp.EncodeToBytes(ProtocolMsg{
 		Code:    msg.Code,
 		Size:    msg.Size,
 		Payload: rlpdata,
@@ -428,7 +428,7 @@ func (prw PssReadWriter) WriteMsg(msg p2p.Msg) error {
 	if err != nil {
 		return err
 	}
-	return prw.Pss.Send(prw.To.Bytes(), *prw.topic, pmsg)
+	return prw.SendRaw(prw.To.Bytes(), *prw.topic, pmsg)
 }
 
 // Injects a p2p.Msg into the MsgReadWriter, so that it appears on the associated p2p.MsgReader
@@ -442,13 +442,12 @@ func (prw PssReadWriter) injectMsg(msg p2p.Msg) error {
 type PssProtocol struct {
 	*Pss
 	proto *p2p.Protocol
-	topic *PssTopic
+	topic *Topic
 	spec  *protocols.Spec
 }
 
 // Constructor
-//func RegisterPssProtocol(pss *Pss, topic *PssTopic, spec *protocols.Spec, targetprotocol *p2p.Protocol) *PssProtocol {
+func RegisterPssProtocol(ps *Pss, topic *Topic, spec *protocols.Spec, targetprotocol *p2p.Protocol) error {
-func RegisterPssProtocol(ps *Pss, topic *PssTopic, spec *protocols.Spec, targetprotocol *p2p.Protocol) error {
 	pp := &PssProtocol{
 		Pss:   ps,
 		proto: targetprotocol,
@@ -461,7 +460,7 @@ func RegisterPssProtocol(ps *Pss, topic *PssTopic, spec *protocols.Spec, targetp
 
 func (self *PssProtocol) handle(msg []byte, p *p2p.Peer, senderAddr []byte) error {
 	hashoaddr := pot.NewHashAddressFromBytes(senderAddr).Address
-	if !self.Pss.isActive(hashoaddr, *self.topic) {
+	if !self.isActive(hashoaddr, *self.topic) {
 		rw := &PssReadWriter{
 			Pss:   self.Pss,
 			To:    hashoaddr,

swarm/pss/pss_test.go

@@ -6,7 +6,6 @@ import (
 	"encoding/hex"
 	"fmt"
 	"io/ioutil"
-	//"math/rand"
 	"os"
 	"sync"
 	"testing"
@@ -41,12 +40,10 @@ func init() {
 	log.Root().SetHandler(h)
 }
 
-func TestPssCache(t *testing.T) {
+func TestCache(t *testing.T) {
 	var err error
 	to, _ := hex.DecodeString("08090a0b0c0d0e0f1011121314150001020304050607161718191a1b1c1d1e1f")
 	oaddr, _ := hex.DecodeString("000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f")
-	//uaddr, _ := hex.DecodeString("101112131415161718191a1b1c1d1e1f000102030405060708090a0b0c0d0e0f")
-	//proofbytes := []byte{241, 172, 117, 105, 88, 154, 82, 33, 176, 188, 91, 244, 245, 85, 86, 16, 120, 232, 70, 45, 182, 188, 99, 103, 157, 3, 202, 121, 252, 21, 129, 22}
 	proofbytes, _ := hex.DecodeString("822fff7527f7ae630c1224921e50a7ca1b27324f00f3966623bd503780c7ab33")
 	ps := NewTestPss(oaddr)
 	pp := NewPssParams(false)
@@ -54,20 +51,20 @@ func TestPssCache(t *testing.T) {
 	datatwo := []byte("bar")
 	fwdaddr := network.RandomAddr()
 	msg := &PssMsg{
-		Payload: &PssEnvelope{
+		Payload: &Envelope{
 			TTL:     0,
 			From:    oaddr,
-			Topic:   PssPingTopic,
+			Topic:   PingTopic,
 			Payload: data,
 		},
 		To: to,
 	}
 
 	msgtwo := &PssMsg{
-		Payload: &PssEnvelope{
+		Payload: &Envelope{
 			TTL:     0,
 			From:    oaddr,
-			Topic:   PssPingTopic,
+			Topic:   PingTopic,
 			Payload: datatwo,
 		},
 		To: to,
@@ -133,7 +130,7 @@ func TestPssCache(t *testing.T) {
 	}
 }
 
-func TestPssRegisterHandler(t *testing.T) {
+func TestRegisterHandler(t *testing.T) {
 	var err error
 	addr := network.RandomAddr()
 	ps := NewTestPss(addr.OAddr)
@@ -151,13 +148,13 @@ func TestPssRegisterHandler(t *testing.T) {
 		return nil
 	}
 	deregister := ps.Register(&topic, checkMsg)
-	pssmsg := &PssMsg{Payload: NewPssEnvelope(from.OAddr, topic, payload)}
+	pssmsg := &PssMsg{Payload: NewEnvelope(from.OAddr, topic, payload)}
 	err = ps.Process(pssmsg)
 	if err != nil {
 		t.Fatal(err)
 	}
 	var i int
-	err = ps.Process(&PssMsg{Payload: NewPssEnvelope(from.OAddr, wrongtopic, payload)})
+	err = ps.Process(&PssMsg{Payload: NewEnvelope(from.OAddr, wrongtopic, payload)})
 	expErr := ""
 	if err == nil || err.Error() == expErr {
 		t.Fatalf("unhandled topic expected '%v', got '%v'", expErr, err)
@@ -172,25 +169,25 @@ func TestPssRegisterHandler(t *testing.T) {
 	}
 	deregister()
 	deregister2()
-	err = ps.Process(&PssMsg{Payload: NewPssEnvelope(from.OAddr, topic, payload)})
+	err = ps.Process(&PssMsg{Payload: NewEnvelope(from.OAddr, topic, payload)})
 	expErr = ""
 	if err == nil || err.Error() == expErr {
 		t.Fatalf("reregister handler expected %v, got %v", expErr, err)
 	}
 }
 
-func TestPssSimpleLinear(t *testing.T) {
+func TestSimpleLinear(t *testing.T) {
 	var err error
 	nodeconfig := adapters.RandomNodeConfig()
 	addr := network.NewAddrFromNodeID(nodeconfig.ID)
 	_ = p2ptest.NewTestPeerPool()
 	ps := NewTestPss(addr.Over())
 
-	ping := &PssPing{
+	ping := &Ping{
-		QuitC: make(chan struct{}),
+		C: make(chan struct{}),
 	}
 
-	err = RegisterPssProtocol(ps, &PssPingTopic, PssPingProtocol, NewPssPingProtocol(ping.PssPingHandler))
+	err = RegisterPssProtocol(ps, &PingTopic, PingProtocol, NewPingProtocol(ping.PingHandler))
 
 	if err != nil {
 		t.Fatalf("Failed to register virtual protocol in pss: %v", err)
@@ -198,7 +195,7 @@ func TestPssSimpleLinear(t *testing.T) {
 	run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 		id := p.ID()
 		pp := protocols.NewPeer(p, rw, pssSpec)
-		bp := &testPssOverlayConn{
+		bp := &testOverlayConn{
 			Peer: pp,
 			addr: network.ToOverlayAddr(id[:]),
 		}
@@ -212,7 +209,7 @@ func TestPssSimpleLinear(t *testing.T) {
 
 	pt := p2ptest.NewProtocolTester(t, nodeconfig.ID, 2, run)
 
-	msg := NewPssPingMsg(network.ToOverlayAddr(pt.IDs[0].Bytes()), PssPingProtocol, PssPingTopic, []byte{1, 2, 3})
+	msg := NewPingMsg(network.ToOverlayAddr(pt.IDs[0].Bytes()), PingProtocol, PingTopic, []byte{1, 2, 3})
 
 	exchange := p2ptest.Exchange{
 		Expects: []p2ptest.Expect{
@@ -237,27 +234,27 @@ func TestPssSimpleLinear(t *testing.T) {
 	}
 }
 
-func TestPssFullRandom50n(t *testing.T) {
+func TestFullRandom50n(t *testing.T) {
 	adapter := adapters.NewSimAdapter(services)
-	testPssFullRandom(t, adapter, 50, 50, 50)
+	testFullRandom(t, adapter, 50, 50, 50)
 }
 
-func TestPssFullRandom25n(t *testing.T) {
+func TestFullRandom25n(t *testing.T) {
 	adapter := adapters.NewSimAdapter(services)
-	testPssFullRandom(t, adapter, 25, 25, 25)
+	testFullRandom(t, adapter, 25, 25, 25)
 }
 
-func TestPssFullRandom10n(t *testing.T) {
+func TestFullRandom10n(t *testing.T) {
 	adapter := adapters.NewSimAdapter(services)
-	testPssFullRandom(t, adapter, 10, 10, 10)
+	testFullRandom(t, adapter, 10, 10, 10)
 }
 
-func TestPssFullRandom5n(t *testing.T) {
+func TestFullRandom5n(t *testing.T) {
 	adapter := adapters.NewSimAdapter(services)
-	testPssFullRandom(t, adapter, 5, 5, 5)
+	testFullRandom(t, adapter, 5, 5, 5)
 }
 
-func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int, fullnodecount int, msgcount int) {
+func testFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int, fullnodecount int, msgcount int) {
 	var i int
 	var msgfromids []discover.NodeID
 	var msgreceived []discover.NodeID
@@ -393,10 +390,10 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int
 			if err != nil {
 				return fmt.Errorf("error getting node client: %s", err)
 			}
-			msg := PssPingMsg{Created: time.Now()}
+			msg := PingMsg{Created: time.Now()}
-			code, _ := PssPingProtocol.GetCode(&PssPingMsg{})
+			code, _ := PingProtocol.GetCode(&PingMsg{})
 			pmsg, _ := NewProtocolMsg(code, msg)
-			client.CallContext(ctx, &rpcerr, "pss_sendPss", PssPingTopic, PssAPIMsg{
+			client.CallContext(ctx, &rpcerr, "pss_send", PingTopic, APIMsg{
 				Addr: fullpeers[msgtoids[ii]],
 				Msg:  pmsg,
 			})
@@ -467,8 +464,8 @@ func triggerChecks(ctx context.Context, wg *sync.WaitGroup, trigger *chan discov
 		return fmt.Errorf("error getting peer events for node %v: %s", id, err)
 	}
 
-	msgevents := make(chan PssAPIMsg)
+	msgevents := make(chan APIMsg)
-	msgsub, err := client.Subscribe(context.Background(), "pss", msgevents, "receivePss", PssPingTopic)
+	msgsub, err := client.Subscribe(context.Background(), "pss", msgevents, "receive", PingTopic)
 	if err != nil {
 		return fmt.Errorf("error getting msg events for node %v: %s", id, err)
 	}
@@ -543,10 +540,10 @@ func newServices() adapters.Services {
 			pssp := NewPssParams(true)
 			ps := NewPss(kademlia(ctx.Config.ID), dpa, pssp)
 
-			ping := &PssPing{
+			ping := &Ping{
-				QuitC: make(chan struct{}),
+				C: make(chan struct{}),
 			}
-			err = RegisterPssProtocol(ps, &PssPingTopic, PssPingProtocol, NewPssPingProtocol(ping.PssPingHandler))
+			err = RegisterPssProtocol(ps, &PingTopic, PingProtocol, NewPingProtocol(ping.PingHandler))
 			if err != nil {
 				log.Error("Couldnt register pss protocol", "err", err)
 				os.Exit(1)
@@ -557,10 +554,12 @@ func newServices() adapters.Services {
 		//"bzz": func(id discover.NodeID, snapshot []byte) node.Service {
 		"bzz": func(ctx *adapters.ServiceContext) (node.Service, error) {
 			addr := network.NewAddrFromNodeID(ctx.Config.ID)
+			hp := network.NewHiveParams()
+			hp.Discovery = true
 			config := &network.BzzConfig{
 				OverlayAddr:  addr.Over(),
 				UnderlayAddr: addr.Under(),
-				HiveParams:   network.NewHiveParams(),
+				HiveParams:   hp,
 			}
 			return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore), nil
 		},
@@ -573,22 +572,22 @@ type connmap struct {
 	lock    sync.Mutex
 }
 
-type testPssOverlayConn struct {
+type testOverlayConn struct {
 	*protocols.Peer
 	addr []byte
 }
 
-func (self *testPssOverlayConn) Address() []byte {
+func (self *testOverlayConn) Address() []byte {
 	return self.addr
 }
 
-func (self *testPssOverlayConn) Off() network.OverlayAddr {
+func (self *testOverlayConn) Off() network.OverlayAddr {
 	return self
 }
 
-func (self *testPssOverlayConn) Drop(err error) {
+func (self *testOverlayConn) Drop(err error) {
 }
 
-func (self *testPssOverlayConn) Update(o network.OverlayAddr) network.OverlayAddr {
+func (self *testOverlayConn) Update(o network.OverlayAddr) network.OverlayAddr {
 	return self
 }

swarm/pss/pssapi.go

@@ -11,18 +11,18 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/network"
 )
 
-// PssAPI is the RPC API module for Pss
+// API is the RPC API module for Pss
-type PssAPI struct {
+type API struct {
 	*Pss
 }
 
-// NewPssAPI constructs a PssAPI instance
+// NewAPI constructs a PssAPI instance
-func NewPssAPI(ps *Pss) *PssAPI {
+func NewAPI(ps *Pss) *API {
-	return &PssAPI{Pss: ps}
+	return &API{Pss: ps}
 }
 
 // NewMsg API endpoint creates an RPC subscription
-func (pssapi *PssAPI) ReceivePss(ctx context.Context, topic PssTopic) (*rpc.Subscription, error) {
+func (pssapi *API) Receive(ctx context.Context, topic Topic) (*rpc.Subscription, error) {
 	notifier, supported := rpc.NotifierFromContext(ctx)
 	if !supported {
 		return nil, fmt.Errorf("Subscribe not supported")
@@ -30,7 +30,7 @@ func (pssapi *PssAPI) ReceivePss(ctx context.Context, topic PssTopic) (*rpc.Subs
 
 	psssub := notifier.CreateSubscription()
 	handler := func(msg []byte, p *p2p.Peer, from []byte) error {
-		apimsg := &PssAPIMsg{
+		apimsg := &APIMsg{
 			Msg:  msg,
 			Addr: from,
 		}
@@ -39,7 +39,7 @@ func (pssapi *PssAPI) ReceivePss(ctx context.Context, topic PssTopic) (*rpc.Subs
 		}
 		return nil
 	}
-	deregf := pssapi.Pss.Register(&topic, handler)
+	deregf := pssapi.Register(&topic, handler)
 
 	go func() {
 		defer deregf()
@@ -55,39 +55,35 @@ func (pssapi *PssAPI) ReceivePss(ctx context.Context, topic PssTopic) (*rpc.Subs
 }
 
 // SendRaw sends the message (serialized into byte slice) to a peer with topic
-func (pssapi *PssAPI) SendPss(topic PssTopic, msg PssAPIMsg) error {
+func (pssapi *API) Send(topic Topic, msg APIMsg) error {
-	if pssapi.Pss.debug && bytes.Equal(msg.Addr, pssapi.Pss.BaseAddr()) {
+	if pssapi.debug && bytes.Equal(msg.Addr, pssapi.BaseAddr()) {
 		log.Warn("Pss debug enabled; send to self shortcircuit", "apimsg", msg, "topic", topic)
-		env := NewPssEnvelope(msg.Addr, topic, msg.Msg)
+		env := NewEnvelope(msg.Addr, topic, msg.Msg)
-		return pssapi.Pss.Process(&PssMsg{
+		return pssapi.Process(&PssMsg{
-			To: pssapi.Pss.BaseAddr(),
+			To:      pssapi.BaseAddr(),
 			Payload: env,
 		})
 	}
-	return pssapi.Pss.Send(msg.Addr, topic, msg.Msg)
+	return pssapi.SendRaw(msg.Addr, topic, msg.Msg)
-	/*if err != nil {
-		return fmt.Errorf("send error: %v", err)
-	}
-	return fmt.Errorf("ok sent")*/
 }
 
 // PssAPITest are temporary API calls for development use only
 // These symbols should not be included in production environment
-type PssAPITest struct {
+type APITest struct {
 	*Pss
 }
 
-// NewPssAPI constructs a PssAPI instance
+// NewAPI constructs a API instance
-func NewPssAPITest(ps *Pss) *PssAPITest {
+func NewAPITest(ps *Pss) *APITest {
-	return &PssAPITest{Pss: ps}
+	return &APITest{Pss: ps}
 }
 
 // temporary for access to overlay while faking kademlia healthy routines
-func (pssapitest *PssAPITest) GetForwarder(addr []byte) (fwd struct {
+func (pssapitest *APITest) GetForwarder(addr []byte) (fwd struct {
 	Addr  []byte
 	Count int
 }) {
-	pssapitest.Pss.Overlay.EachConn(addr, 255, func(op network.OverlayConn, po int, isproxbin bool) bool {
+	pssapitest.Overlay.EachConn(addr, 255, func(op network.OverlayConn, po int, isproxbin bool) bool {
 		if bytes.Equal(fwd.Addr, []byte{}) {
 			fwd.Addr = op.Address()
 		}
@@ -98,6 +94,6 @@ func (pssapitest *PssAPITest) GetForwarder(addr []byte) (fwd struct {
 }
 
 // BaseAddr gets our own overlayaddress
-func (pssapitest *PssAPITest) BaseAddr() ([]byte, error) {
+func (pssapitest *APITest) BaseAddr() ([]byte, error) {
 	return pssapitest.Pss.BaseAddr(), nil
 }

swarm/pss/simulations/service_test.go

@@ -31,7 +31,7 @@ func init() {
 // TestPssProtocol starts a pss network along with two test nodes which run
 // protocols via the pss network, connects those two test nodes and then
 // waits for them to handshake
-func TestPssProtocol(t *testing.T) {
+func TestProtocol(t *testing.T) {
 	// define the services
 	w := &testWrapper{}
 	services := adapters.Services{
@@ -141,7 +141,7 @@ func (t *testWrapper) newTestService(ctx *adapters.ServiceContext) (node.Service
 	}
 	return &testService{
 		id:         ctx.Config.ID,
-		pss:        client.NewPssClientWithRPC(context.Background(), rpcClient),
+		pss:        client.NewClientWithRPC(context.Background(), rpcClient),
 		handshakes: make(chan *testHandshake),
 	}, nil
 }
@@ -154,7 +154,7 @@ type testHandshake struct {
 // so that clients can wait for handshakes to complete
 type testService struct {
 	id         discover.NodeID
-	pss        *client.PssClient
+	pss        *client.Client
 	handshakes chan *testHandshake
 }
 
@@ -199,7 +199,7 @@ func (t *testService) run(_ *p2p.Peer, rw p2p.MsgReadWriter) error {
 }
 
 type TestAPI struct {
-	pss        *client.PssClient
+	pss        *client.Client
 	handshakes chan *testHandshake
 }
 

swarm/pss/types.go

@@ -21,21 +21,21 @@ const (
 // Defines params for Pss
 type PssParams struct {
 	Cachettl time.Duration
-	Debug bool
+	Debug    bool
 }
 
 // Initializes default params for Pss
 func NewPssParams(debug bool) *PssParams {
 	return &PssParams{
 		Cachettl: defaultDigestCacheTTL,
-		Debug: debug,
+		Debug:    debug,
 	}
 }
 
 // Encapsulates the message transported over pss.
 type PssMsg struct {
 	To      []byte
-	Payload *PssEnvelope
+	Payload *Envelope
 }
 
 func (msg *PssMsg) Serialize() []byte {
@@ -51,23 +51,23 @@ func (self *PssMsg) String() string {
 // Topic defines the context of a message being transported over pss
 // It is used by pss to determine what action is to be taken on an incoming message
 // Typically, one can map protocol handlers for the message payloads by mapping topic to them; see *Pss.Register()
-type PssTopic [TopicLength]byte
+type Topic [TopicLength]byte
 
-func (self *PssTopic) String() string {
+func (self *Topic) String() string {
 	return fmt.Sprintf("%x", self)
 }
 
 // Pre-Whisper placeholder, payload of PssMsg
-type PssEnvelope struct {
+type Envelope struct {
-	Topic   PssTopic
+	Topic   Topic
 	TTL     uint16
 	Payload []byte
 	From    []byte
 }
 
 // creates Pss envelope from sender address, topic and raw payload
-func NewPssEnvelope(addr []byte, topic PssTopic, payload []byte) *PssEnvelope {
+func NewEnvelope(addr []byte, topic Topic, payload []byte) *Envelope {
-	return &PssEnvelope{
+	return &Envelope{
 		From:    addr,
 		Topic:   topic,
 		TTL:     DefaultTTL,
@@ -76,7 +76,7 @@ func NewPssEnvelope(addr []byte, topic PssTopic, payload []byte) *PssEnvelope {
 }
 
 // encapsulates a protocol msg as PssEnvelope data
-type PssProtocolMsg struct {
+type ProtocolMsg struct {
 	Code       uint64
 	Size       uint32
 	Payload    []byte
@@ -85,7 +85,7 @@ type PssProtocolMsg struct {
 
 // PssAPIMsg is the type for messages, it extends the rlp encoded protocol Msg
 // with the Sender's overlay address
-type PssAPIMsg struct {
+type APIMsg struct {
 	Msg  []byte
 	Addr []byte
 }
@@ -100,7 +100,7 @@ func NewProtocolMsg(code uint64, msg interface{}) ([]byte, error) {
 	// previous attempts corrupted nested structs in the payload iself upon deserializing
 	// therefore we use two separate []byte fields instead of peerAddr
 	// TODO verify that nested structs cannot be used in rlp
-	smsg := &PssProtocolMsg{
+	smsg := &ProtocolMsg{
 		Code:    code,
 		Size:    uint32(len(rlpdata)),
 		Payload: rlpdata,
@@ -110,12 +110,12 @@ func NewProtocolMsg(code uint64, msg interface{}) ([]byte, error) {
 }
 
 // Message handler func for a topic
-type PssHandler func(msg []byte, p *p2p.Peer, from []byte) error
+type Handler func(msg []byte, p *p2p.Peer, from []byte) error
 
 // constructs a new PssTopic from a given name and version.
 //
 // Analogous to the name and version members of p2p.Protocol
-func NewTopic(s string, v int) (topic PssTopic) {
+func NewTopic(s string, v int) (topic Topic) {
 	h := sha3.NewKeccak256()
 	h.Write([]byte(s))
 	buf := make([]byte, TopicLength/8)
@@ -126,7 +126,7 @@ func NewTopic(s string, v int) (topic PssTopic) {
 }
 
 func ToP2pMsg(msg []byte) (p2p.Msg, error) {
-	payload := &PssProtocolMsg{}
+	payload := &ProtocolMsg{}
 	if err := rlp.DecodeBytes(msg, payload); err != nil {
 		return p2p.Msg{}, fmt.Errorf("pss protocol handler unable to decode payload as p2p message: %v", err)
 	}

swarm/swarm.go

@@ -21,7 +21,6 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"fmt"
-	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
@@ -37,26 +36,27 @@ import (
 	httpapi "github.com/ethereum/go-ethereum/swarm/api/http"
 	"github.com/ethereum/go-ethereum/swarm/fuse"
 	"github.com/ethereum/go-ethereum/swarm/network"
+	"github.com/ethereum/go-ethereum/swarm/pss"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
 // the swarm stack
 type Swarm struct {
-	config      *api.Config        // swarm configuration
+	config  *api.Config        // swarm configuration
-	api         *api.Api           // high level api layer (fs/manifest)
+	api     *api.Api           // high level api layer (fs/manifest)
-	dns         api.Resolver       // DNS registrar
+	dns     api.Resolver       // DNS registrar
-	storage     storage.ChunkStore // internal access to storage, common interface to cloud storage backends
+	storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends
-	dpa         *storage.DPA       // distributed preimage archive, the local API to the storage with document level storage/retrieval support
+	dpa     *storage.DPA       // distributed preimage archive, the local API to the storage with document level storage/retrieval support
-	cloud       storage.CloudStore // procurement, cloud storage backend (can multi-cloud)
+	cloud   storage.CloudStore // procurement, cloud storage backend (can multi-cloud)
-	hive        *network.Hive      // the logistic manager
+	//hive        *network.Hive      // the logistic manager
+	bzz         *network.Bzz       // hive and bzz protocol
 	backend     chequebook.Backend // simple blockchain Backend
 	privateKey  *ecdsa.PrivateKey
 	corsString  string
 	swapEnabled bool
-	pssEnabled  bool
-	pss         *network.Pss
 	lstore      *storage.LocalStore // local store, needs to store for releasing resources after node stopped
 	sfs         *fuse.SwarmFS       // need this to cleanup all the active mounts on node exit
+	pss         *pss.Pss
 }
 
 type SwarmAPI struct {
@@ -89,7 +89,6 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
 		backend:     backend,
 		privateKey:  config.Swap.PrivateKey(),
 		corsString:  cors,
-		pssEnabled:  pssEnabled,
 	}
 	log.Debug(fmt.Sprintf("Setting up Swarm service components"))
 
@@ -102,18 +101,26 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
 	log.Debug("Set up local db access (iterator/counter)")
 
 	kp := network.NewKadParams()
-
 	to := network.NewKademlia(
 		common.FromHex(config.BzzKey),
 		kp,
 	)
-	// set up the kademlia hive
-	self.hive = network.NewHive(
-		config.HiveParams, // configuration parameters
-		to,
-	)
-	log.Debug(fmt.Sprintf("Set up swarm network with Kademlia hive"))
 
+	bzzconfig := &network.BzzConfig{
+		UnderlayAddr: common.FromHex(config.PublicKey),
+		OverlayAddr:  common.FromHex(config.BzzKey),
+		HiveParams:   config.HiveParams,
+	}
+	self.bzz = network.NewBzz(bzzconfig, to, nil)
+
+	// set up the kademlia hive
+	//	self.hive = network.NewHive(
+	//		config.HiveParams, // configuration parameters
+	//		self.Kademlia,
+	//		stateStore,
+	//	)
+	//	log.Debug(fmt.Sprintf("Set up swarm network with Kademlia hive"))
+	//
 	// setup cloud storage internal access layer
 	self.storage = storage.NewNetStore(hash, self.lstore, nil, config.StoreParams)
 	log.Debug("-> swarm net store shared access layer to Swarm Chunk Store")
@@ -126,6 +133,12 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
 	self.dpa = storage.NewDPA(dpaChunkStore, self.config.ChunkerParams)
 	log.Debug(fmt.Sprintf("-> Content Store API"))
 
+	// Pss = postal service over swarm (devp2p over bzz)
+	if pssEnabled {
+		pssparams := pss.NewPssParams(false)
+		self.pss = pss.NewPss(to, self.dpa, pssparams)
+	}
+
 	// set up high level api
 	transactOpts := bind.NewKeyedTransactor(self.privateKey)
 
@@ -176,25 +189,16 @@ func (self *Swarm) Start(net *p2p.Server) error {
 
 	log.Warn(fmt.Sprintf("Starting Swarm service"))
 
-	self.hive.Start(
+	//	self.hive.Start(net)
-		net,
+	err := self.bzz.Start(net)
-		func() <-chan time.Time {
+	if err != nil {
-			return time.NewTicker(time.Second).C
+		log.Error("bzz failed", "err", err)
-		},
+		return err
-		nil,
+	}
-	)
+	log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", self.bzz.Hive.Overlay.BaseAddr()))
 
-	log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", self.hive.BaseAddr()))
+	if self.pss != nil {
-
+		self.pss.Start(net)
-	if self.pssEnabled {
-		pssparams := network.NewPssParams()
-		self.pss = network.NewPss(self.hive.Overlay, pssparams)
-
-		// for testing purposes, shold be removed in production environment!!
-		pingtopic, _ := network.MakeTopic("pss", 1)
-		self.pss.Register(pingtopic, self.pss.GetPingHandler())
-
-		log.Debug("Pss started: %v", self.pss)
 	}
 
 	self.dpa.Start()
@@ -222,7 +226,11 @@ func (self *Swarm) Start(net *p2p.Server) error {
 // stops all component services.
 func (self *Swarm) Stop() error {
 	self.dpa.Stop()
-	self.hive.Stop()
+	//self.hive.Stop()
+	self.bzz.Stop()
+	if self.pss != nil {
+		self.pss.Stop()
+	}
 	if ch := self.config.Swap.Chequebook(); ch != nil {
 		ch.Stop()
 		ch.Save()
@@ -236,53 +244,30 @@ func (self *Swarm) Stop() error {
 }
 
 // implements the node.Service interface
-func (self *Swarm) Protocols() []p2p.Protocol {
+func (self *Swarm) Protocols() (protos []p2p.Protocol) {
-	ct := network.BzzCodeMap()
+
-	if self.pssEnabled {
+	for _, p := range self.bzz.Protocols() {
-		ct.Register(1, &network.PssMsg{})
+		protos = append(protos, p)
 	}
-	ct.Register(2, network.DiscoveryMsgs...)
 
-	// srv := func(p network.Peer) error {
+	if self.pss != nil {
-	// 	if self.pssEnabled {
+		for _, p := range self.pss.Protocols() {
-	// 		p.Register(&network.PssMsg{}, func(msg interface{}) error {
+			protos = append(protos, p)
-	// 			pssmsg := msg.(*network.PssMsg)
+		}
+	}
+	// ct := network.BzzCodeMap()
+	// ct.Register(2, network.DiscoveryMsgs...)
+
+	//	proto := network.Bzz(
+	//		self.hive.Overlay.GetAddr().Over(),
+	//		self.hive.Overlay.GetAddr().Under(),
+	//		ct,
+	//		nil,
+	//		nil,
+	//		nil,
+	//	)
 	//
-	// 			if self.pss.IsSelfRecipient(pssmsg) {
+	return
-	// 				log.Trace("pss for us, yay! ... let's process!")
-	// 				env := pssmsg.Payload
-	// 				umsg := env.Payload
-	// 				f := self.pss.GetHandler(env.Topic)
-	// 				if f == nil {
-	// 					return fmt.Errorf("No registered handler for topic '%s'", env.Topic)
-	// 				}
-	// 				nid := discover.MustBytesID(env.SenderUAddr)
-	// 				p := p2p.NewPeer(nid.NodeID, fmt.Sprintf("%x", common.ByteLabel(nid.Bytes())), []p2p.Cap{})
-	// 				return f(umsg, p, env.SenderOAddr)
-	// 			} else {
-	// 				log.Trace("pss was for someone else :'( ... forwarding")
-	// 				return self.pss.Forward(pssmsg)
-	// 			}
-	// 			return nil
-	// 		})
-	// 	}
-	// 	self.hive.Add(p)
-	// 	p.DisconnectHook(func(err error) {
-	// 		self.hive.Remove(p)
-	// 	})
-	// 	return nil
-	// }
-
-	proto := network.Bzz(
-		self.hive.Overlay.GetAddr().Over(),
-		self.hive.Overlay.GetAddr().Under(),
-		ct,
-		nil,
-		nil,
-		nil,
-	)
-
-	return []p2p.Protocol{*proto}
 }
 
 // implements node.Service
@@ -301,7 +286,7 @@ func (self *Swarm) APIs() []rpc.API {
 		{
 			Namespace: "bzz",
 			Version:   "0.1",
-			Service:   api.NewControl(self.api, self.hive),
+			Service:   api.NewControl(self.api, self.bzz.Hive),
 			Public:    false,
 		},
 		{
@@ -333,13 +318,14 @@ func (self *Swarm) APIs() []rpc.API {
 		// {Namespace, Version, api.NewAdmin(self), false},
 	}
 
-	if self.pssEnabled {
+	for _, api := range self.bzz.APIs() {
-		apis = append(apis, rpc.API{
+		apis = append(apis, api)
-			Namespace: "eth",
+	}
-			Version:   "0.1/pss",
+
-			Service:   network.NewPssApi(self.pss),
+	if self.pss != nil {
-			Public:    true,
+		for _, api := range self.pss.APIs() {
-		})
+			apis = append(apis, api)
+		}
 	}
 
 	return apis