swarm, swarm/pss, swarm/network: pssclient rw reads and writes from websocket

pss is now separate package
pss event feed removed, multipile handlers allowed instead
NOTE! pss tests are removed and will be rewritten
remove pssclient outbox, psscache bug fix
WIP tests reinstation

p2p/simulations/adapters/docker.go

@@ -42,8 +42,10 @@ func (d *DockerAdapter) Name() string {
 
 // NewNode returns a new DockerNode using the given config
 func (d *DockerAdapter) NewNode(config *NodeConfig) (Node, error) {
-	if _, exists := serviceFuncs[config.Service]; !exists {
+	for _, name := range config.Services {
-		return nil, fmt.Errorf("unknown node service %q", config.Service)
+		if _, exists := serviceFuncs[name]; !exists {
+			return nil, fmt.Errorf("unknown node service %q", name)
+		}
 	}
 
 	// generate the config
@@ -60,6 +62,7 @@ func (d *DockerAdapter) NewNode(config *NodeConfig) (Node, error) {
 		ExecNode: ExecNode{
 			ID:     config.Id,
 			Config: conf,
+			Services: config.Services,
 		},
 	}
 	node.newCmd = node.dockerCommand
@@ -81,12 +84,16 @@ func (n *DockerNode) dockerCommand() *exec.Cmd {
 	return exec.Command(
 		"sh", "-c",
 		fmt.Sprintf(
-			`exec docker run --interactive --env _P2P_NODE_CONFIG="${_P2P_NODE_CONFIG}" %s p2p-node %s %s`,
+			`exec docker run --interactive --env _P2P_NODE_CONFIG="${_P2P_NODE_CONFIG}" --env _P2P_NODE_KEY="${_P2P_NODE_KEY}" %s p2p-node %s %s`,
-			dockerImage, n.Config.Node.Service, n.ID.String(),
+			dockerImage, n.Services[0], n.ID.String(),
 		),
 	)
 }
 
+func (n *DockerNode) GetService(name string) node.Service {
+	return nil
+}
+
 // dockerImage is the name of the docker image
 const dockerImage = "p2p-node"
 

p2p/simulations/adapters/exec.go

@@ -46,8 +46,10 @@ func (e *ExecAdapter) Name() string {
 
 // NewNode returns a new ExecNode using the given config
 func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
-	if _, exists := serviceFuncs[config.Service]; !exists {
+	for _, name := range config.Services {
-		return nil, fmt.Errorf("unknown node service %q", config.Service)
+		if _, exists := serviceFuncs[name]; !exists {
+			return nil, fmt.Errorf("unknown node service %q", name)
+		}
 	}
 
 	// create the node directory using the first 12 characters of the ID
@@ -75,6 +77,7 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
 		ID:     config.Id,
 		Dir:    dir,
 		Config: conf,
+		Services: config.Services,
 	}
 	node.newCmd = node.execCommand
 	return node, nil
@@ -93,6 +96,7 @@ type ExecNode struct {
 	Config *execNodeConfig
 	Cmd    *exec.Cmd
 	Info   *p2p.NodeInfo
+	Services []string
 
 	client *rpc.Client
 	rpcMux *rpcMux
@@ -164,13 +168,18 @@ func (n *ExecNode) Start(snapshot []byte) (err error) {
 	return nil
 }
 
+
+func (n *ExecNode) GetService(name string) node.Service {
+	return nil
+}
+
 // execCommand returns a command which runs the node locally by exec'ing
 // the current binary but setting argv[0] to "p2p-node" so that the child
 // runs execP2PNode
 func (n *ExecNode) execCommand() *exec.Cmd {
 	return &exec.Cmd{
 		Path: reexec.Self(),
-		Args: []string{"p2p-node", n.Config.Node.Service, n.ID.String()},
+		Args: []string{"p2p-node", n.Services[0], n.ID.String()},
 	}
 }
 

p2p/simulations/adapters/inproc.go

@@ -24,6 +24,7 @@ import (
 	"sync"
 
 	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/discover"
@@ -55,6 +56,8 @@ func (s *SimAdapter) Name() string {
 
 // NewNode returns a new SimNode using the given config
 func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
+	var nodeprotos []p2p.Protocol
+	
 	s.mtx.Lock()
 	defer s.mtx.Unlock()
 
@@ -65,6 +68,7 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
 	}
 
 	// check the service is valid and initialize it
+/*
 	serviceFunc, exists := s.services[config.Service]
 	if !exists {
 		return nil, fmt.Errorf("unknown node service %q", config.Service)
@@ -75,9 +79,52 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
 		config:      config,
 		adapter:     s,
 		serviceFunc: serviceFunc,
+*/
+	//serviceFunc, exists := s.services[config.Service]
+	
+	//if !exists {
+	//	return nil, fmt.Errorf("unknown node service %q", config.Service)
+	//}
+	//service := serviceFunc(id)
+	
+	n, err := node.New(&node.Config{
+		P2P: p2p.Config{
+			PrivateKey:      config.PrivateKey,
+			MaxPeers:        math.MaxInt32,
+			NoDiscovery:     true,
+			Protocols:       nodeprotos,
+			Dialer:          s,
+			EnableMsgEvents: true,
+		},
+	})
+	if err != nil {
+		return nil, err
 	}
-	s.nodes[id.NodeID] = node
+	
-	return node, nil
+	services := make(map[string]node.Service)
+	
+	for name, servicefunc := range s.services {
+		service := servicefunc(id)
+		if err := n.Register(func(ctx *node.ServiceContext) (node.Service, error) {
+			return service, err
+		}); err != nil {
+			return nil, err
+		}
+		for _, proto := range service.Protocols() {
+			nodeprotos = append(nodeprotos, proto)
+		}
+		services[name] = service
+	}
+	
+	simnode := &SimNode{
+		Node:    n,
+		Id:      id,
+		services: services,
+		adapter:     s,
+		config:      config,
+	}
+	s.nodes[id.NodeID] = simnode
+	return simnode, nil
 }
 
 func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) {
@@ -113,7 +160,7 @@ type SimNode struct {
 	Id          *NodeId
 	config      *NodeConfig
 	adapter     *SimAdapter
-	serviceFunc ServiceFunc
+	services 	map[string]node.Service
 	node        *node.Node
 	running     node.Service
 	client      *rpc.Client
@@ -233,13 +280,6 @@ func (self *SimNode) Stop() error {
 	return nil
 }
 
-// Service returns the underlying node.Service
-func (self *SimNode) Service() node.Service {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-	return self.running
-}
-
 func (self *SimNode) Server() *p2p.Server {
 	self.lock.Lock()
 	defer self.lock.Unlock()
@@ -247,6 +287,11 @@ func (self *SimNode) Server() *p2p.Server {
 		return nil
 	}
 	return self.node.Server()
+
+// Service returns a underlying node.Service of the speficied type
+func (self *SimNode) GetService(servicename string) node.Service {
+	log.Warn("retrieving service", "name", servicename)
+	return self.services[servicename]
 }
 
 func (self *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription {

p2p/simulations/adapters/types.go

@@ -62,6 +62,9 @@ type Node interface {
 
 	// Snapshot creates a snapshot of the running service
 	Snapshot() ([]byte, error)
+	
+	// Gets a service by name
+	GetService(string) node.Service
 }
 
 // NodeAdapter is an object which creates Nodes to be used in a simulation
@@ -127,11 +130,11 @@ type NodeConfig struct {
 	// Name is a human friendly name for the node like "node01"
 	Name string
 
-	// Service is the name of the service which should be run when starting
+	// Services is the name of the services which should be run when starting
-	// the node (for SimNodes it should be the name of a service contained
+	// the node (for SimNodes it should be the names of services contained
-	// in SimAdapter.services, for other nodes it should be a service
+	// in SimAdapter.services, for other nodes it should be services
 	// registered by calling the RegisterService function)
-	Service string
+	Services []string
 }
 
 // nodeConfigJSON is used to encode and decode NodeConfig as JSON by converting
@@ -140,13 +143,13 @@ type nodeConfigJSON struct {
 	Id         string `json:"id"`
 	PrivateKey string `json:"private_key"`
 	Name       string `json:"name"`
-	Service    string `json:"service"`
+	Services   []string `json:"services"`
 }
 
 func (n *NodeConfig) MarshalJSON() ([]byte, error) {
 	confJSON := nodeConfigJSON{
 		Name:    n.Name,
-		Service: n.Service,
+		Services: n.Services,
 	}
 	if n.Id != nil {
 		confJSON.Id = n.Id.String()
@@ -180,7 +183,7 @@ func (n *NodeConfig) UnmarshalJSON(data []byte) error {
 	}
 
 	n.Name = confJSON.Name
-	n.Service = confJSON.Service
+	n.Services = confJSON.Services
 
 	return nil
 }

p2p/simulations/network.go

@@ -190,7 +190,7 @@ func (self *Msg) String() string {
 // NewNode adds a new node to the network with a random ID
 func (self *Network) NewNode() (*Node, error) {
 	conf := adapters.RandomNodeConfig()
-	conf.Service = self.DefaultService
+	conf.Services = append(conf.Services, self.DefaultService)
 	return self.NewNodeWithConfig(conf)
 }
 
@@ -203,8 +203,8 @@ func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error)
 	if conf.Name == "" {
 		conf.Name = fmt.Sprintf("node%02d", len(self.Nodes)+1)
 	}
-	if conf.Service == "" {
+	if len(conf.Services) == 0 {
-		conf.Service = self.DefaultService
+		conf.Services = append(conf.Services, self.DefaultService)
 	}
 
 	_, found := self.nodeMap[id.NodeID]

p2p/testing/protocolsession.go

@@ -57,7 +57,7 @@ func (self *ProtocolSession) trigger(trig Trigger) error {
 	if !ok {
 		return fmt.Errorf("trigger: peer %v does not exist (1- %v)", trig.Peer, len(self.Ids))
 	}
-	mockNode, ok := simNode.Service().(*mockNode)
+	mockNode, ok := simNode.GetService("mock").(*mockNode)
 	if !ok {
 		return fmt.Errorf("trigger: peer %v is not a mock", trig.Peer)
 	}
@@ -92,7 +92,7 @@ func (self *ProtocolSession) expect(exp Expect) error {
 	if !ok {
 		return fmt.Errorf("trigger: peer %v does not exist (1- %v)", exp.Peer, len(self.Ids))
 	}
-	mockNode, ok := simNode.Service().(*mockNode)
+	mockNode, ok := simNode.GetService("mock").(*mockNode)
 	if !ok {
 		return fmt.Errorf("trigger: peer %v is not a mock", exp.Peer)
 	}

p2p/testing/protocoltester.go

@@ -29,7 +29,7 @@ func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(*p2p.P
 	}
 	adapter := adapters.NewSimAdapter(services)
 	net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{})
-	if _, err := net.NewNodeWithConfig(&adapters.NodeConfig{Id: id, Service: "test"}); err != nil {
+	if _, err := net.NewNodeWithConfig(&adapters.NodeConfig{Id: id, Services: []string{"test"}}); err != nil {
 		panic(err.Error())
 	}
 	if err := net.Start(id); err != nil {
@@ -41,7 +41,7 @@ func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(*p2p.P
 	peerIDs := make([]*adapters.NodeId, n)
 	for i := 0; i < n; i++ {
 		peers[i] = adapters.RandomNodeConfig()
-		peers[i].Service = "mock"
+		peers[i].Services = append(peers[i].Services, "mock")
 		peerIDs[i] = peers[i].Id
 	}
 	events := make(chan *p2p.PeerEvent, 1000)

swarm/network/pssapi.go

@@ -1,66 +0,0 @@
-package network
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/ethereum/go-ethereum/log"
-	"github.com/ethereum/go-ethereum/rpc"
-)
-
-type PssApi struct {
-	*Pss
-}
-
-func NewPssApi(ps *Pss) *PssApi {
-	return &PssApi{Pss: ps}
-}
-
-func (self *PssApi) NewMsg(ctx context.Context, topic PssTopic) (*rpc.Subscription, error) {
-	notifier, supported := rpc.NotifierFromContext(ctx)
-	if !supported {
-		return nil, fmt.Errorf("Subscribe not supported")
-	}
-
-	sub := notifier.CreateSubscription()
-
-	ch := make(chan []byte)
-	psssub, err := self.Pss.Subscribe(&topic, ch)
-	if err != nil {
-		return nil, fmt.Errorf("pss subscription topic %v (rpc sub id %v) failed: %v", topic, sub.ID, err)
-	}
-
-	go func(topic PssTopic) error {
-		for {
-			select {
-			case msg := <-ch:
-				if err := notifier.Notify(sub.ID, msg); err != nil {
-					log.Warn(fmt.Sprintf("notification on pss sub topic %v rpc (sub %v) msg %v failed!", topic, sub.ID, msg))
-					return err
-				}
-			case err := <-psssub.Err():
-				log.Warn(fmt.Sprintf("caught subscription error in pss sub topic: %v", topic, err))
-				return err
-			case <-notifier.Closed():
-				log.Warn(fmt.Sprintf("rpc sub notifier closed"))
-				psssub.Unsubscribe()
-				return nil
-			case err := <-sub.Err():
-				log.Warn(fmt.Sprintf("rpc sub closed: %v", err))
-				psssub.Unsubscribe()
-				return nil
-			}
-		}
-		return nil
-	}(topic)
-
-	return sub, nil
-}
-
-func (self *PssApi) SendRaw(to []byte, topic PssTopic, msg []byte) error {
-	err := self.Pss.Send(to, topic, msg)
-	if err != nil {
-		return fmt.Errorf("send error: %v", err)
-	}
-	return fmt.Errorf("ok sent")
-}

swarm/pss/client.go

@@ -0,0 +1,228 @@
+package pss 
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+	"github.com/ethereum/go-ethereum/p2p/protocols"
+	"github.com/ethereum/go-ethereum/pot"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+const (
+	inboxCapacity = 3000
+	outboxCapacity = 100
+	addrLen = common.HashLength
+)
+
+// implements p2p.Server
+// implements net.Conn
+type PssClient struct {
+	localuri string
+	remoteuri  string
+	ctx context.Context
+	cancel func()
+	subscription *rpc.ClientSubscription
+	topicsC chan []byte
+	msgC chan PssAPIMsg
+	quitC chan struct{}
+	quitting uint32
+	ws *rpc.Client
+	lock sync.Mutex
+	peerPool map[PssTopic]map[pot.Address]*pssRPCRW
+	protos []*p2p.Protocol
+}
+
+type pssRPCRW struct {
+	*PssClient
+	topic *PssTopic
+	spec *protocols.Spec
+	msgC chan []byte
+	addr pot.Address
+}
+
+func (self *PssClient) newpssRPCRW(addr pot.Address, spec *protocols.Spec, topic *PssTopic) *pssRPCRW {
+	return &pssRPCRW {
+		PssClient: self,
+		topic: topic,
+		spec: spec,
+		msgC: make(chan []byte),
+		addr: addr,
+	}
+}
+
+func (rw *pssRPCRW) ReadMsg() (p2p.Msg, error) {
+	msg := <- rw.msgC
+	log.Warn("pssrpcrw read", "msg", msg)
+	pmsg, err := ToP2pMsg(msg)
+	if err != nil {
+		return p2p.Msg{}, err
+	}
+	
+	return pmsg, nil
+}
+
+func (rw *pssRPCRW) WriteMsg(msg p2p.Msg) error {
+	
+	ifc, found := rw.spec.NewMsg(msg.Code)
+	if !found {
+		return fmt.Errorf("could not find interface for msg #%d", msg.Code)
+	}
+	msg.Decode(ifc)
+	pmsg, err := newProtocolMsg(msg.Code, ifc)
+	if err != nil {
+		return fmt.Errorf("Could not render protocolmessage", "error", err)
+	}
+	
+	return rw.PssClient.ws.CallContext(rw.PssClient.ctx, nil, "pss_sendRaw", rw.topic, PssAPIMsg{
+		Addr: rw.addr.Bytes(),
+		Msg: pmsg,
+	})
+	
+}
+
+// remotehost: hostname of node running websockets proxy to pss (default localhost)
+// remoteport: port of node running websockets proxy to pss (0 = go-ethereum node default)
+// secure: whether or not to use secure connection
+// originhost: local if host to connect from
+
+func NewPssClient(ctx context.Context, cancel func(), remotehost string, remoteport int, secure bool, originhost string) *PssClient {
+	prefix := "ws"
+	
+	if ctx == nil {
+		ctx = context.Background()
+		cancel = func() {return}
+	}
+	pssc := &PssClient{
+		msgC: make(chan PssAPIMsg),
+		quitC: make(chan struct{}),
+		peerPool: make(map[PssTopic]map[pot.Address]*pssRPCRW),
+		ctx: ctx,
+		cancel: cancel,
+	}
+
+	if remotehost == "" {
+		remotehost = "localhost"
+	}
+	
+	if remoteport == 0 {
+		remoteport = node.DefaultWSPort
+	}
+	
+	if originhost == "" {
+		originhost = "localhost"
+	}
+	
+	if secure {
+		prefix = "wss"
+	}
+	
+	pssc.remoteuri = fmt.Sprintf("%s://%s:%d", prefix, remotehost, remoteport)
+	pssc.localuri = fmt.Sprintf("%s://%s", prefix, originhost)
+	
+	return pssc
+}
+
+func (self *PssClient) shutdown() {
+	atomic.StoreUint32(&self.quitting, 1)
+	self.cancel()
+}
+
+func (self *PssClient) Start() error {
+	log.Debug("Dialing ws", "src", self.localuri, "dst", self.remoteuri)
+	ws, err := rpc.DialWebsocket(self.ctx, self.remoteuri, self.localuri)
+	if err != nil {
+		return fmt.Errorf("Couldnt dial pss websocket: %v", err)
+	}
+	
+	self.ws = ws
+	
+	return nil
+}
+
+func (self *PssClient) RunProtocol(proto *p2p.Protocol, spec *protocols.Spec) error {
+	topic := NewTopic(spec.Name, int(spec.Version))
+	msgC := make(chan PssAPIMsg)
+	self.peerPool[topic] = make(map[pot.Address]*pssRPCRW)
+	sub, err := self.ws.Subscribe(self.ctx, "pss", msgC, "newMsg", topic)
+	if err != nil {
+		return fmt.Errorf("pss event subscription failed: %v", err)
+	}
+	
+	self.subscription = sub
+	
+	// dispatch incoming messages
+	go func() {
+		for {
+			select {
+				case msg := <- msgC:
+					var addr pot.Address
+					copy(addr[:], msg.Addr)
+					if self.peerPool[topic][addr] == nil {
+						self.peerPool[topic][addr] = self.newpssRPCRW(addr, spec, &topic)
+						nid, _ := discover.HexID("0x00")
+						p := p2p.NewPeer(nid, fmt.Sprintf("%v", addr), []p2p.Cap{})
+						go proto.Run(p, self.peerPool[topic][addr])
+					}
+					go func() {
+						self.peerPool[topic][addr].msgC <- msg.Msg
+					}()
+				case <-self.quitC:
+					self.shutdown()
+					return
+			}
+		}
+	}()
+	
+	self.protos = append(self.protos, proto)
+	return nil
+}
+
+func (self *PssClient) Stop() error {
+	self.cancel()
+	return nil
+}
+
+func (self *PssClient) AddPssPeer(addr pot.Address, spec *protocols.Spec) {
+	topic := NewTopic(spec.Name, int(spec.Version))
+	if self.peerPool[topic][addr] == nil {
+		self.peerPool[topic][addr] = self.newpssRPCRW(addr, spec, &topic)
+	}
+}
+
+func (self *PssClient) RemovePssPeer(addr pot.Address, spec *protocols.Spec) {
+	topic := NewTopic(spec.Name, int(spec.Version))	
+	delete(self.peerPool[topic], addr)
+}
+
+func (self *PssClient) 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 {
+	return len(self.peerPool)
+}
+
+func (self *PssClient) NodeInfo() *p2p.NodeInfo {
+	return nil
+}
+
+func (self *PssClient) PeersInfo() []*p2p.PeerInfo {
+	return nil
+}
+func (self *PssClient) 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) {
+	log.Error("Cannot remove peer in PSS with discover.Node, need swarm overlay address")
+}

swarm/pss/client_test.go

@@ -0,0 +1,173 @@
+package pss
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+	"github.com/ethereum/go-ethereum/p2p/protocols"
+	"github.com/ethereum/go-ethereum/pot"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+func init() {
+	h := log.CallerFileHandler(log.StreamHandler(os.Stderr, log.TerminalFormat(true)))
+	log.Root().SetHandler(h)
+}
+
+func TestRunProtocol(t *testing.T) {
+	quitC := make(chan struct{})
+	pss := newTestPss(nil)
+	ping := &pssPing{
+		quitC: make(chan struct{}),
+	}
+	proto := newProtocol(ping)	
+	_, err := baseTester(t, proto, pss, nil, nil, quitC)
+	if err != nil {
+		t.Fatalf(err.Error())
+	}
+	quitC <- struct{}{}
+}
+
+func TestIncoming(t *testing.T) {
+	quitC := make(chan struct{})
+	pss := newTestPss(nil)
+	ctx, cancel := context.WithCancel(context.Background())
+	var addr []byte
+	ping := &pssPing{
+		quitC: make(chan struct{}),
+	}
+	proto := newProtocol(ping)	
+	client, err := baseTester(t, proto, pss, ctx, cancel, quitC)
+	if err != nil {
+		t.Fatalf(err.Error())
+	}
+	
+	client.ws.Call(&addr, "pss_baseAddr")
+
+	code, _ := pssPingProtocol.GetCode(&pssPingMsg{})
+	rlpbundle, err := newProtocolMsg(code, &pssPingMsg{
+		Created: time.Now(),
+	})
+	if err != nil {
+		t.Fatalf("couldn't make pssmsg")
+	}
+
+	pssenv := PssEnvelope{
+		From: addr,
+		Topic:       NewTopic(proto.Name, int(proto.Version)),
+		TTL:         DefaultTTL,
+		Payload:     rlpbundle,
+	}
+	pssmsg := PssMsg{
+		To: addr,
+		Payload: &pssenv,
+	}
+	
+	pss.Process(&pssmsg)
+	
+	go func() {
+		<-ping.quitC
+		client.cancel()
+	}()
+	
+	<-client.ctx.Done()
+	quitC <- struct{}{}
+}
+
+func TestOutgoing(t *testing.T) {
+	quitC := make(chan struct{})
+	pss := newTestPss(nil)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond * 250)
+	var addr []byte
+	var potaddr pot.Address
+	
+	ping := &pssPing{
+		quitC: make(chan struct{}),
+	}
+	proto := newProtocol(ping)	
+	client, err := baseTester(t, proto, pss, ctx, cancel, quitC)
+	if err != nil {
+		t.Fatalf(err.Error())
+	}
+	
+	client.ws.Call(&addr, "pss_baseAddr")
+	copy(potaddr[:], addr)
+					
+	msg := &pssPingMsg{
+		Created: time.Now(),
+	}
+	
+	topic := NewTopic(pssPingProtocol.Name, int(pssPingProtocol.Version))
+	client.AddPssPeer(potaddr, pssPingProtocol)
+	nid, _ := discover.HexID("0x00")
+	p := p2p.NewPeer(nid, fmt.Sprintf("%v", potaddr), []p2p.Cap{})
+	pp := protocols.NewPeer(p, client.peerPool[topic][potaddr], pssPingProtocol)
+	pp.Send(msg)
+	<-client.ctx.Done()
+	quitC <- struct{}{}
+}
+
+func baseTester(t *testing.T, proto *p2p.Protocol, pss *Pss, ctx context.Context, cancel func(), quitC chan struct{}) (*PssClient, error) {
+	var err error
+	
+	client := newClient(t, pss, ctx, cancel, quitC)
+	
+	err = client.Start()
+	if err != nil {
+		return nil, err
+	}
+	
+	err = client.RunProtocol(proto, pssPingProtocol)
+	
+	if err != nil {
+		return nil, err
+	}
+	
+	return client, nil
+}
+
+func newProtocol(ping *pssPing) *p2p.Protocol {
+	
+	return &p2p.Protocol{
+		Name: pssPingProtocol.Name,
+		Version: pssPingProtocol.Version,
+		Length: 1,
+		Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
+			pp := protocols.NewPeer(p, rw, pssPingProtocol)
+			pp.Run(ping.pssPingHandler)
+			return nil
+		},
+	}
+}
+
+func newClient(t *testing.T, pss *Pss, ctx context.Context, cancel func(), quitC chan struct{}) *PssClient {
+	pssclient := NewPssClient(ctx, cancel, "", 0, false, "")
+	
+	srv := rpc.NewServer()
+	srv.RegisterName("pss", NewPssAPI(pss))
+	ws := srv.WebsocketHandler([]string{"*"})
+	uri := fmt.Sprintf("%s:%d", node.DefaultWSHost, node.DefaultWSPort)
+	
+	sock, err := net.Listen("tcp", uri)
+	if err != nil {
+		t.Fatalf("Tcp (recv) on %s failed: %v", uri, err)
+	}
+	
+	go func() {
+		http.Serve(sock, ws)
+	}()
+
+	go func() {
+		<-quitC
+		sock.Close()
+	}()	
+	return pssclient
+} 

swarm/pss/common.go

@@ -0,0 +1,66 @@
+package pss
+
+import (
+	"io/ioutil"
+	"os"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p/protocols"
+	"github.com/ethereum/go-ethereum/swarm/network"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)	
+
+type pssPingMsg struct {
+	Created time.Time
+}
+
+type pssPing struct {
+	quitC chan struct{}
+}
+
+func (self *pssPing) pssPingHandler(msg interface{}) error {
+	log.Warn("got ping", "msg", msg)
+	self.quitC <- struct{}{}
+	return nil
+}
+
+var pssPingProtocol = &protocols.Spec{
+	Name:       "psstest",
+	Version:    1,
+	MaxMsgSize: 10 * 1024 * 1024,
+	Messages: []interface{}{
+		pssPingMsg{},
+	},
+}
+
+func newTestPss(addr []byte) *Pss {	
+	if addr == nil {
+		addr = network.RandomAddr().OAddr
+	}
+	
+	// set up storage
+	cachedir, err := ioutil.TempDir("", "pss-cache")
+	if err != nil {
+		log.Error("create pss cache tmpdir failed", "error", err)
+		os.Exit(1)
+	}
+	dpa, err := storage.NewLocalDPA(cachedir)
+	if err != nil {
+		log.Error("local dpa creation failed", "error", err)
+		os.Exit(1)
+	}
+	
+	// set up routing
+	kp := network.NewKadParams()
+	kp.MinProxBinSize = 3
+
+	// create pss
+	pp := NewPssParams()
+
+	overlay := network.NewKademlia(addr, kp)
+	ps := NewPss(overlay, dpa, pp)
+
+	return ps
+}
+

swarm/pss/pss.go

@@ -1,4 +1,4 @@
-package network
+package pss
 
 import (
 	"bytes"
@@ -9,13 +9,16 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/event"
+	"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/protocols"
+	"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/pot"
 	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
@@ -25,17 +28,13 @@ const (
 	TopicResolverLength         = 8
 	PssPeerCapacity             = 256
 	PssPeerTopicDefaultCapacity = 8
-	digestLength                = 64
+	digestLength                = 32
 	digestCapacity              = 256
 	defaultDigestCacheTTL       = time.Second
-	pingTopicName               = "pss"
-	pingTopicVersion            = 1
 )
 
 var (
-	errorNoForwarder   = errors.New("no available forwarders in routing table")
 	errorForwardToSelf = errors.New("forward to self")
-	errorBlockByCache  = errors.New("message found in blocking cache")
 )
 
 // Defines params for Pss
@@ -51,50 +50,14 @@ func NewPssParams() *PssParams {
 }
 
 // Encapsulates the message transported over pss.
-//
-// Warning: do not access the To-member directly. Use *PssMsg.GetRecipient() and *PssMsg.SetRecipient() instead.
 type PssMsg struct {
-	// (we need the To-member exported for type inference)
 	To      []byte
-	Payload pssEnvelope
+	Payload *PssEnvelope
-}
-
-// Retrieve the remote peer receipient address of the message
-func (self *PssMsg) GetRecipient() []byte {
-	return self.To
-}
-
-// Set the remote peer recipient address of the message
-func (self *PssMsg) SetRecipient(to []byte) {
-	self.To = to
 }
 
 // String representation of PssMsg
 func (self *PssMsg) String() string {
-	return fmt.Sprintf("PssMsg: Recipient: %x", common.ByteLabel(self.GetRecipient()))
+	return fmt.Sprintf("PssMsg: Recipient: %x", common.ByteLabel(self.To))
-}
-
-// Pre-Whisper placeholder
-type pssEnvelope struct {
-	Topic       PssTopic
-	TTL         uint16
-	Payload     []byte
-	SenderOAddr []byte
-	// SenderUAddr []byte
-}
-
-// Pre-Whisper placeholder
-type pssPayload struct {
-	Code       uint64
-	Size       uint32
-	Data       []byte
-	ReceivedAt time.Time
-}
-
-// Pre-Whisper placeholder
-type pssCacheEntry struct {
-	expiresAt    time.Time
-	receivedFrom []byte
 }
 
 // Topic defines the context of a message being transported over pss
@@ -102,14 +65,70 @@ type pssCacheEntry struct {
 // Typically, one can map protocol handlers for the message payloads by mapping topic to them; see *Pss.Register()
 type PssTopic [TopicLength]byte
 
-// Pre-Whisper placeholder
+func (self *PssTopic) String() string {
-type pssDigest uint32
+	return fmt.Sprintf("%x", self)
+}
+
+// Pre-Whisper placeholder, payload of PssMsg
+type PssEnvelope struct {
+	Topic       PssTopic
+	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 {
+	return &PssEnvelope{
+		From: 		 addr,
+		Topic:       topic,
+		TTL:         DefaultTTL,
+		Payload:     payload,
+	}
+}
+
+
+func (msg *PssMsg) serialize() []byte {
+	rlpdata, _ := rlp.EncodeToBytes(msg)
+	/*buf := bytes.NewBuffer(nil)
+	buf.Write(self.PssEnvelope.Topic[:])
+	buf.Write(self.PssEnvelope.Payload)
+	buf.Write(self.PssEnvelope.From)
+	return buf.Bytes()*/
+	return rlpdata
+}
+
+
+var pssTransportProtocol = &protocols.Spec{
+	Name:       "pss",
+	Version:    1,
+	MaxMsgSize: 10 * 1024 * 1024,
+	Messages: []interface{}{
+		PssMsg{},
+	},
+}
+
+// encapsulates a protocol msg as PssEnvelope data
+type PssProtocolMsg struct {
+	Code       uint64
+	Size       uint32
+	Payload    []byte
+	ReceivedAt time.Time
+}
+
+type pssCacheEntry struct {
+	expiresAt    time.Time
+	receivedFrom []byte
+}
+
+type pssDigest [digestLength]byte
+
+// Message handler func for a topic
+type pssHandler func(msg []byte, p *p2p.Peer, from []byte) error
 
 // 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.
-// The structure is used by normal incoming message handlers on the nodes to determine which action to take, forward or process.
-// Thus it is up to the implementer to write a handler, and link the PssMsg to this appropriate handler.
 //
 // The top-level Pss object provides:
 //
@@ -120,69 +139,117 @@ type pssDigest uint32
 // - a dispatcher lookup, mapping protocols to topics
 // - a message cache to spot messages that previously have been forwarded
 type Pss struct {
-	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]*PssReadWriter // keep track of all virtual p2p.Peers we are currently speaking to
+	handlers map[PssTopic]map[*pssHandler]bool              // topic and version based pss payload handlers
-	peerPool map[pot.Address]map[PssTopic]p2p.MsgReadWriter     // keep track of all virtual p2p.Peers we are currently speaking to
+	fwdcache map[pssDigest]pssCacheEntry                    // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg
-	handlers map[PssTopic]func([]byte, *p2p.Peer, []byte) error // topic and version based pss payload handlers
+	cachettl time.Duration                                  // how long to keep messages in fwdcache
-	events   map[PssTopic]*event.Feed                           // subscriptions for each topic
-	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
-	hasher   func(string) storage.Hasher                        // hasher to digest message to cache
-	baseAddr []byte
 	lock     sync.Mutex
+	dpa		*storage.DPA
 }
 
-func (self *Pss) hashMsg(msg *PssMsg) pssDigest {
+func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
-	hasher := self.hasher("SHA3")()
+	swg := &sync.WaitGroup{}
-	hasher.Reset()
+	wwg := &sync.WaitGroup{}
-	hasher.Write(msg.GetRecipient())
+	buf := bytes.NewReader(msg.serialize())
-	// hasher.Write(msg.Payload.SenderUAddr)
+	key, err := self.dpa.Store(buf, int64(buf.Len()), swg, wwg)
-	hasher.Write(msg.Payload.SenderOAddr)
+	if err != nil {
-	hasher.Write(msg.Payload.Topic[:])
+		log.Warn("Could not store in swarm", "err", err)
-	hasher.Write(msg.Payload.Payload)
+		return pssDigest{}, err
-	b := hasher.Sum([]byte{})
+	}
-	return pssDigest(binary.BigEndian.Uint32(b))
+	log.Trace("Stored msg in swarm", "key", key)
+	digest := pssDigest{}
+	copy(digest[:], key[:digestLength])
+	return digest, nil
 }
 
 // Creates a new Pss instance. A node should only need one of these
 //
-// TODO error check overlay integrity
+// TODO: error check overlay integrity
-func NewPss(k Overlay, params *PssParams) *Pss {
+func NewPss(k network.Overlay, dpa *storage.DPA, params *PssParams) *Pss {
-	baseAddr := k.BaseAddr()
 	return &Pss{
 		Overlay: k,
-		//peerPool: make(map[pot.Address]map[PssTopic]*PssReadWriter, PssPeerCapacity),
 		peerPool: make(map[pot.Address]map[PssTopic]p2p.MsgReadWriter, PssPeerCapacity),
-		handlers: make(map[PssTopic]func([]byte, *p2p.Peer, []byte) error),
+		handlers: make(map[PssTopic]map[*pssHandler]bool),
-		events:   make(map[PssTopic]*event.Feed),
 		fwdcache: make(map[pssDigest]pssCacheEntry),
 		cachettl: params.Cachettl,
-		hasher:   storage.MakeHashFunc,
+		dpa: dpa,
-		baseAddr: baseAddr,
 	}
 }
 
-func (p *Pss) Run(peer *bzzPeer) error {
+func (self *Pss) Start(srv *p2p.Server) error {
-	return peer.Run(p.HandleMsg)
-}
-
-func (p *Pss) HandleMsg(m interface{}) error {
-	msg, ok := m.(*PssMsg)
-	if !ok {
-		return fmt.Errorf("unknown pss protocol message type: %T", m)
-	}
-	_ = msg
-	// TODO: handle the message
 	return nil
 }
 
+func (self *Pss) Stop() error {
+	return nil
+}
+
+func (self *Pss) Protocols() []p2p.Protocol {
+	return []p2p.Protocol{
+		p2p.Protocol{
+			Name:    pssTransportProtocol.Name,
+			Version: pssTransportProtocol.Version,
+			Length:  pssTransportProtocol.Length(),
+			Run:     func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
+						pp := protocols.NewPeer(p, rw, pssTransportProtocol)
+						pp.Run(self.handlePssMsg)
+						return nil
+					},
+		},
+	}
+}
+
+func (self *Pss) APIs() []rpc.API {
+	return []rpc.API{
+		rpc.API {
+			Namespace: "pss",
+			Version:   "0.1",
+			Service:   NewPssAPI(self),
+			Public:    true,
+		},
+	}
+}
+
+// Takes the generated PssTopic 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() {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	handlers := self.handlers[topic]
+	if handlers == nil {
+		handlers = make(map[*pssHandler]bool)
+		self.handlers[topic] = handlers
+	}
+	handlers[&handler] = true
+	return func() { self.deregister(topic, &handler) }
+}
+
+func (self *Pss) deregister(topic PssTopic, h *pssHandler) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	handlers := self.handlers[topic]
+	if len(handlers) == 1 {
+		delete(self.handlers, topic)
+		return
+	}
+	delete(handlers, h)
+}
+
 // enables to set address of node, to avoid backwards forwarding
 //
 // 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 := self.hashMsg(msg)
+	digest, err := self.storeMsg(msg)
+	if err != nil {
+		return err
+	}
 	return self.addFwdCacheSender(addr, digest)
 }
 
@@ -228,35 +295,115 @@ func (self *Pss) checkFwdCache(addr []byte, digest pssDigest) bool {
 	return false
 }
 
-// Takes the generated PssTopic of a protocol, and links a handler function to it
+func (self *Pss) getHandlers(topic PssTopic) map[*pssHandler]bool {
-// This allows the implementer to retrieve the right handler function (invoke the right protocol) for an incoming message by inspecting the topic on it.
-func (self *Pss) Register(topic PssTopic, handler func(msg []byte, p *p2p.Peer, from []byte) error) error {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-	self.handlers[topic] = func(msg []byte, p *p2p.Peer, from []byte) error {
-		self.alertSubscribers(&topic, msg)
-		return handler(msg, p, from)
-	}
-	self.registerFeed(topic)
-	return nil
-}
-
-func (self *Pss) Subscribe(topic *PssTopic, ch chan []byte) (event.Subscription, error) {
-	_, ok := self.events[*topic]
-	if !ok {
-		return nil, fmt.Errorf("No feed registered for topic %v", topic)
-	}
-	sub := self.events[*topic].Subscribe(ch)
-	log.Trace("new pss subscribe", "topic", topic, "sub", sub)
-	return sub, nil
-}
-
-func (self *Pss) GetHandler(topic PssTopic) func([]byte, *p2p.Peer, []byte) error {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	return self.handlers[topic]
 }
 
+//
+func (self *Pss) handlePssMsg(msg interface{}) error {
+	pssmsg := msg.(*PssMsg)
+
+	if !self.isSelfRecipient(pssmsg) {
+		log.Trace("pss was for someone else :'( ... forwarding")
+		return self.Forward(pssmsg)
+	}
+	log.Trace("pss for us, yay! ... let's process!")
+	return self.Process(pssmsg)
+}
+
+// processes a message with self as recipient
+func (self *Pss) Process(pssmsg *PssMsg) error {
+	env := pssmsg.Payload
+	payload := env.Payload
+	handlers := self.getHandlers(env.Topic)
+	if len(handlers) == 0 {
+		return fmt.Errorf("No registered handler for topic '%s'", env.Topic)
+	}
+	nid, _ := discover.HexID("0x00")
+	p := p2p.NewPeer(nid, fmt.Sprintf("%x", env.From), []p2p.Cap{})
+	for f := range handlers {
+		err := (*f)(payload, p, env.From)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Sends a message using pss. The message could be anything at all, and will be handled by whichever handler function is mapped to PssTopic using *Pss.Register()
+//
+// The to address is a swarm overlay address
+func (self *Pss) Send(to []byte, topic PssTopic, msg []byte) error {
+	sender := self.Overlay.BaseAddr()
+	pssenv := NewPssEnvelope(sender, topic, msg)
+	pssmsg := &PssMsg{
+		To:      to,
+		Payload: pssenv,
+	}
+	return self.Forward(pssmsg)
+}
+
+// Forwards a pss message to the peer(s) closest to the to address
+//
+// Handlers that want to pass on a message should call this directly
+func (self *Pss) Forward(msg *PssMsg) error {
+
+	if self.isSelfRecipient(msg) {
+		return errorForwardToSelf
+	}
+
+	digest, err := self.storeMsg(msg)
+	if err != nil {
+		log.Warn(fmt.Sprintf("could not store message %v to cache: %v", msg, err))
+	}
+
+	if self.checkFwdCache(nil, digest) {
+		log.Trace(fmt.Sprintf("pss relay block-cache match: FROM %x TO %x", common.ByteLabel(self.Overlay.BaseAddr()), common.ByteLabel(msg.To)))
+		return nil
+	}
+
+	// TODO:check integrity of message
+	sent := 0
+
+	// 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)
+		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) {
+			log.Info(fmt.Sprintf("%v: peer already forwarded to", sendMsg))
+			return true
+		}
+		err := p.Send(msg)
+		if err != nil {
+			log.Warn(fmt.Sprintf("%v: failed forwarding: %v", sendMsg, err))
+			return true
+		}
+		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 {
+			return false
+		}
+		log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ByteLabel(p.Over())))
+		return true
+	})
+
+	if sent == 0 {
+		log.Error("PSS: unable to forward to any peers")
+		return nil
+	}
+
+	self.addFwdCacheExpire(digest)
+	return nil
+}
+
 // 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
@@ -267,121 +414,32 @@ func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, run adapters.RunProtocol
 	go func() {
 		err := run(p, rw)
 		log.Warn(fmt.Sprintf("pss vprotocol quit on addr %v topic %v: %v", addr, topic, err))
+		self.removePeerTopic(rw, topic)
 	}()
 	return nil
 }
 
-// Removes a pss peer from the pss peerpool
-func (self *Pss) RemovePeer(id pot.Address) {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-	self.peerPool[id] = nil
-	return
-}
-
 func (self *Pss) addPeerTopic(id pot.Address, topic PssTopic, rw p2p.MsgReadWriter) error {
-	if self.peerPool[id][topic] == nil {
+	if self.peerPool[id] == nil {
 		self.peerPool[id] = make(map[PssTopic]p2p.MsgReadWriter, PssPeerTopicDefaultCapacity)
 	}
 	self.peerPool[id][topic] = rw
 	return nil
 }
 
-func (self *Pss) removePeerTopic(id pot.Address, topic PssTopic) {
+func (self *Pss) removePeerTopic(rw p2p.MsgReadWriter, topic PssTopic) {
-	self.peerPool[id][topic] = nil
+	prw, ok := rw.(*PssReadWriter)
-	return
+	if !ok {
+		return
+	}
+	delete(self.peerPool[prw.To], topic)
+	if len(self.peerPool[prw.To]) == 0 {
+		delete(self.peerPool, prw.To)
+	}
 }
 
 func (self *Pss) isActive(id pot.Address, topic PssTopic) bool {
-	if self.peerPool[id][topic] == nil {
+	return self.peerPool[id][topic] != nil
-		return false
-	}
-	return true
-}
-
-func (self *Pss) registerFeed(topic PssTopic) {
-	self.events[topic] = &event.Feed{}
-}
-
-func (self *Pss) alertSubscribers(topic *PssTopic, msg []byte) error {
-	feed, ok := self.events[*topic]
-	if !ok {
-		return fmt.Errorf("No subscriptions registered for topic %v", topic)
-	}
-	numsent := feed.Send(msg)
-	log.Trace(fmt.Sprintf("pss sent to %d subscribers", numsent))
-	return nil
-}
-
-// Sends a message using pss. The message could be anything at all, and will be handled by whichever handler function is mapped to PssTopic using *Pss.Register()
-//
-// The to address is a swarm overlay address
-func (self *Pss) Send(to []byte, topic PssTopic, msg []byte) error {
-
-	pssenv := pssEnvelope{
-		SenderOAddr: self.baseAddr,
-		// SenderUAddr: self.baseAddr.Under(),
-		Topic:   topic,
-		TTL:     DefaultTTL,
-		Payload: msg,
-	}
-
-	pssmsg := &PssMsg{
-		Payload: pssenv,
-	}
-	pssmsg.SetRecipient(to)
-
-	return self.Forward(pssmsg)
-}
-
-// Forwards a pss message to the peer(s) closest to the to address
-//
-// Handlers that want to pass on a message should call this directly
-func (self *Pss) Forward(msg *PssMsg) error {
-
-	if self.IsSelfRecipient(msg) {
-		return errorForwardToSelf
-	}
-
-	digest := self.hashMsg(msg)
-
-	if self.checkFwdCache(nil, digest) {
-		log.Trace(fmt.Sprintf("pss relay block-cache match: FROM %x TO %x", common.ByteLabel(self.baseAddr), common.ByteLabel(msg.GetRecipient())))
-		//return errorBlockByCache
-		return nil
-	}
-
-	// TODO:check integrity of message
-
-	sent := 0
-
-	// send with kademlia
-	// find the closest peer to the recipient and attempt to send
-	self.Overlay.EachConn(msg.GetRecipient(), 256, func(p OverlayConn, po int, isproxbin bool) bool {
-		if self.checkFwdCache(p.Address(), digest) {
-			log.Warn(fmt.Sprintf("BOUNCE DEFER PSS-relay FROM %x TO %x THRU %x:", common.ByteLabel(self.baseAddr), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.Address())))
-			return true
-		}
-		log.Warn(fmt.Sprintf("Attempting PSS-relay FROM %x TO %x THRU %x", common.ByteLabel(self.baseAddr), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.Address())))
-		err := p.(Peer).Send(msg)
-		if err != nil {
-			log.Warn(fmt.Sprintf("FAILED PSS-relay FROM %x TO %x THRU %x: %v", common.ByteLabel(self.baseAddr), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.Address()), err))
-			return true
-		}
-		sent++
-		if bytes.Equal(msg.GetRecipient(), p.Address()) || !isproxbin {
-			return false
-		}
-		log.Trace(fmt.Sprintf("%x is in proxbin, so we continue sending", common.ByteLabel(p.Address())))
-		return true
-	})
-	if sent == 0 {
-		return fmt.Errorf("PSS Was not able to send to any peers")
-	} else {
-		self.addFwdCacheExpire(digest)
-	}
-
-	return nil
 }
 
 // Convenience object that:
@@ -392,19 +450,17 @@ func (self *Pss) Forward(msg *PssMsg) error {
 // Implements p2p.MsgReadWriter
 type PssReadWriter struct {
 	*Pss
-	RecipientOAddr pot.Address
+	To         pot.Address
-	LastActive     time.Time
+	LastActive time.Time
-	rw             chan p2p.Msg
+	rw         chan p2p.Msg
-	spec           *protocols.Spec
+	spec		*protocols.Spec
-	topic          *PssTopic
+	topic      *PssTopic
 }
 
 // Implements p2p.MsgReader
 func (prw PssReadWriter) ReadMsg() (p2p.Msg, error) {
 	msg := <-prw.rw
-
 	log.Trace(fmt.Sprintf("pssrw readmsg: %v", msg))
-
 	return msg, nil
 }
 
@@ -417,11 +473,11 @@ func (prw PssReadWriter) WriteMsg(msg p2p.Msg) error {
 	}
 	msg.Decode(ifc)
 
-	to := prw.RecipientOAddr.Bytes()
+	pmsg, err := newProtocolMsg(msg.Code, ifc)
-
+	if err != nil {
-	pmsg, _ := makeMsg(msg.Code, ifc)
+		return err
-
+	}
-	return prw.Pss.Send(to, *prw.topic, pmsg)
+	return prw.Pss.Send(prw.To.Bytes(), *prw.topic, pmsg)
 }
 
 // Injects a p2p.Msg into the MsgReadWriter, so that it appears on the associated p2p.MsgReader
@@ -432,52 +488,40 @@ func (prw PssReadWriter) injectMsg(msg p2p.Msg) error {
 }
 
 // Convenience object for passing messages in and out of the p2p layer
-type pssProtocol struct {
+type PssProtocol struct {
 	*Pss
-	virtualProtocol *p2p.Protocol
+	proto *p2p.Protocol
 	topic           *PssTopic
-	spec            *protocols.Spec
+	spec			*protocols.Spec
 }
 
 // Constructor
-func NewPssProtocol(pss *Pss, topic *PssTopic, spec *protocols.Spec, targetprotocol *p2p.Protocol) *pssProtocol {
+func NewPssProtocol(pss *Pss, topic *PssTopic, spec *protocols.Spec, targetprotocol *p2p.Protocol) *PssProtocol {
-	pp := &pssProtocol{
+	pp := &PssProtocol{
 		Pss:             pss,
-		virtualProtocol: targetprotocol,
+		proto: targetprotocol,
 		topic:           topic,
-		spec:            spec,
+		spec:			 spec,
 	}
 	return pp
 }
 
-// Retrieves a convenience method for passing an incoming message into the p2p layer
+func (self *PssProtocol) handle(msg []byte, p *p2p.Peer, senderAddr []byte) error {
-//
-// If the implementer wishes to use the p2p.Protocol (or p2p/protocols) message handling, this handler can be directly registered as a handler for the PssMsg structure
-func (self *pssProtocol) GetHandler() func([]byte, *p2p.Peer, []byte) error {
-	return self.handle
-}
-
-func (self *pssProtocol) handle(msg []byte, p *p2p.Peer, senderAddr []byte) error {
 	hashoaddr := pot.NewHashAddressFromBytes(senderAddr).Address
 	if !self.isActive(hashoaddr, *self.topic) {
 		rw := &PssReadWriter{
-			Pss:            self.Pss,
+			Pss:   self.Pss,
-			RecipientOAddr: hashoaddr,
+			To:    hashoaddr,
-			rw:             make(chan p2p.Msg),
+			rw:    make(chan p2p.Msg),
-			spec:           self.spec,
+			spec:	self.spec,
-			topic:          self.topic,
+			topic: self.topic,
 		}
-		self.Pss.AddPeer(p, hashoaddr, self.virtualProtocol.Run, *self.topic, rw)
+		self.Pss.AddPeer(p, hashoaddr, self.proto.Run, *self.topic, rw)
 	}
 
-	payload := &pssPayload{}
+	pmsg, err := ToP2pMsg(msg)
-	rlp.DecodeBytes(msg, payload)
+	if err != nil {
-
+		return fmt.Errorf("could not decode pssmsg")
-	pmsg := p2p.Msg{
-		Code:       payload.Code,
-		Size:       uint32(len(payload.Data)),
-		ReceivedAt: time.Now(),
-		Payload:    bytes.NewBuffer(payload.Data),
 	}
 
 	vrw := self.Pss.peerPool[hashoaddr][*self.topic].(*PssReadWriter)
@@ -486,26 +530,11 @@ func (self *pssProtocol) handle(msg []byte, p *p2p.Peer, senderAddr []byte) erro
 	return nil
 }
 
-func (self *Pss) IsSelfRecipient(msg *PssMsg) bool {
+func (self *Pss) isSelfRecipient(msg *PssMsg) bool {
-	if bytes.Equal(msg.GetRecipient(), self.baseAddr) {
+	return bytes.Equal(msg.To, self.Overlay.BaseAddr())
-		return true
-	}
-	return false
 }
 
-func (self *Pss) GetPingHandler() func([]byte, *p2p.Peer, []byte) error {
+func newProtocolMsg(code uint64, msg interface{}) ([]byte, error) {
-	pingtopic, _ := MakeTopic(pingTopicName, pingTopicVersion)
-	return func(msg []byte, p *p2p.Peer, from []byte) error {
-		if bytes.Equal([]byte("ping"), msg) {
-			log.Trace(fmt.Sprintf("swarm pss ping from %x sending pong", common.ByteLabel(from)))
-			self.Send(from, pingtopic, []byte("pong"))
-		}
-		return nil
-	}
-}
-
-// Pre-Whisper placeholder
-func makeMsg(code uint64, msg interface{}) ([]byte, error) {
 
 	rlpdata, err := rlp.EncodeToBytes(msg)
 	if err != nil {
@@ -515,30 +544,39 @@ func makeMsg(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 := &pssPayload{
+	smsg := &PssProtocolMsg{
 		Code: code,
 		Size: uint32(len(rlpdata)),
-		Data: rlpdata,
+		Payload: rlpdata,
 	}
 
-	rlpbundle, err := rlp.EncodeToBytes(smsg)
+	return rlp.EncodeToBytes(smsg)
-	if err != nil {
-		return nil, err
-	}
-
-	return rlpbundle, nil
 }
 
-// Compiles a new PssTopic from a given name and version.
+// constructs a new PssTopic from a given name and version.
 //
 // Analogous to the name and version members of p2p.Protocol
-func MakeTopic(s string, v int) (PssTopic, error) {
+func NewTopic(s string, v int) (topic PssTopic) {
-	t := [TopicLength]byte{}
+	h := sha3.NewKeccak256()
-	if len(s)+4 <= TopicLength {
+	h.Write([]byte(s))
-		copy(t[4:len(s)+4], s)
+	buf := make([]byte, TopicLength / 8)
-	} else {
+	binary.PutUvarint(buf, uint64(v))
-		return t, fmt.Errorf("topic '%t' too long", s)
+	h.Write(buf)
-	}
+	copy(topic[:], h.Sum(buf)[:])
-	binary.PutVarint(t[:4], int64(v))
+	return topic
-	return t, nil
+}
+
+
+func ToP2pMsg(msg []byte) (p2p.Msg, error) {
+	payload := &PssProtocolMsg{}
+	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)
+	}
+	
+	return p2p.Msg{
+		Code:       payload.Code,
+		Size:       uint32(len(payload.Payload)),
+		ReceivedAt: time.Now(),
+		Payload:    bytes.NewBuffer(payload.Payload),
+	}, nil
 }

swarm/pss/pss_test.go

@@ -1,23 +1,399 @@
-package network
+package pss
 
 import (
+	"bytes"
+	"context"
+	"encoding/hex"
+	"fmt"
+	"io/ioutil"
 	"os"
+	"testing"
+	"time"
 	
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p"
+	"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/swarm/network"
+	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
 const (
-	protocolName    = "foo"
+	pssServiceName 	= "pss"
-	protocolVersion = 42
+	bzzServiceName 	= "bzz"
 )
 
+var topic PssTopic = NewTopic(pssPingProtocol.Name, int(pssPingProtocol.Version))
+
+var services = newServices()
+
 func init() {
+<<<<<<< HEAD:swarm/network/pss_test.go
 	h := log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(true)))
 	//
 	// h := log.CallerFileHandler(log.StreamHandler(os.Stderr, log.TerminalFormat(true)))
+=======
+	adapters.RegisterServices(services)
+	h := log.CallerFileHandler(log.StreamHandler(os.Stderr, log.TerminalFormat(true)))
+>>>>>>> 9c47957... swarm, swarm/pss, swarm/network: pssclient rw reads and writes from websocket:swarm/pss/pss_test.go
 	log.Root().SetHandler(h)
 }
 
+func TestPssCache(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("ad312dca94df401555cfdeb85a6a1f87fb8f240f08dc36af246bd9d4d41efd89")
+	ps := newTestPss(oaddr)
+	pp := NewPssParams()
+	data := []byte("foo")
+	datatwo := []byte("bar")
+	fwdaddr := network.RandomAddr()
+	msg := &PssMsg{
+		Payload: &PssEnvelope{
+			TTL: 0,
+			From: oaddr,
+			Topic:   topic,
+			Payload: data,
+		},
+		To: to,
+	}
+	
+	msgtwo := &PssMsg{
+		Payload: &PssEnvelope{
+			TTL:  0,
+			From: oaddr,
+			Topic:   topic,
+			Payload: datatwo,
+		},
+		To: to,
+	}
+
+	digest, err := ps.storeMsg(msg)
+	if err != nil {
+		t.Fatalf("could not store cache msgone: %v", err)
+	}
+	digesttwo, err := ps.storeMsg(msgtwo)
+	if err != nil {
+		t.Fatalf("could not store cache msgtwo: %v", err)
+	}
+	
+	if !bytes.Equal(digest[:], proofbytes) {
+		t.Fatalf("digest - got: %x, expected: %x", digest, proofbytes)
+	}
+
+	if digest == digesttwo {
+		t.Fatalf("different msgs return same crc: %d", digesttwo)
+	}
+
+	// check the sender cache
+	err = ps.addFwdCacheSender(fwdaddr.Over(), digest)
+	if err != nil {
+		t.Fatalf("write to pss sender cache failed: %v", err)
+	}
+
+	if !ps.checkFwdCache(fwdaddr.Over(), digest) {
+		t.Fatalf("message %v should have SENDER record in cache but checkCache returned false", msg)
+	}
+
+	if ps.checkFwdCache(fwdaddr.Over(), digesttwo) {
+		t.Fatalf("message %v should NOT have SENDER record in cache but checkCache returned true", msgtwo)
+	}
+
+	// check the expire cache
+	err = ps.addFwdCacheExpire(digest)
+	if err != nil {
+		t.Fatalf("write to pss expire cache failed: %v", err)
+	}
+
+	if !ps.checkFwdCache(nil, digest) {
+		t.Fatalf("message %v should have EXPIRE record in cache but checkCache returned false", msg)
+	}
+
+	if ps.checkFwdCache(nil, digesttwo) {
+		t.Fatalf("message %v should NOT have EXPIRE record in cache but checkCache returned true", msgtwo)
+	}
+
+	time.Sleep(pp.Cachettl)
+	if ps.checkFwdCache(nil, digest) {
+		t.Fatalf("message %v should have expired from cache but checkCache returned true", msg)
+	}
+
+	err = ps.AddToCache(fwdaddr.Over(), msgtwo)
+	if err != nil {
+		t.Fatalf("public accessor cache write failed: %v", err)
+	}
+
+	if !ps.checkFwdCache(fwdaddr.Over(), digesttwo) {
+		t.Fatalf("message %v should have SENDER record in cache but checkCache returned false", msgtwo)
+	}
+}
+
+func TestPssRegisterHandler(t *testing.T) {
+	var err error
+	addr := network.RandomAddr()
+	ps := newTestPss(addr.OAddr)
+	from := network.RandomAddr()
+	payload := []byte("payload")
+	topic := NewTopic(pssTransportProtocol.Name, int(pssTransportProtocol.Version))
+	wrongtopic := NewTopic("foo", 42)
+	checkMsg := func(msg []byte, p *p2p.Peer, sender []byte) error {
+		if !bytes.Equal(from.OAddr, sender) {
+			return fmt.Errorf("sender mismatch. expected %x, got %x", from.OAddr, sender)
+		}
+		if !bytes.Equal(msg, payload) {
+			return fmt.Errorf("sender mismatch. expected %x, got %x", msg, payload)
+		}
+		return nil
+	}
+	deregister := ps.Register(topic, checkMsg)
+	pssmsg := &PssMsg{Payload: NewPssEnvelope(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)})
+	expErr := ""
+	if err == nil || err.Error() == expErr {
+		t.Fatalf("unhandled topic expected '%v', got '%v'", expErr, err)
+	}
+	deregister2 := ps.Register(topic, func(msg []byte, p *p2p.Peer, sender []byte) error { i++; return nil })
+	err = ps.Process(pssmsg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if i != 1 {
+		t.Fatalf("second registerer handler did not run")
+	}
+	deregister()
+	deregister2()
+	err = ps.Process(&PssMsg{Payload: NewPssEnvelope(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) {
+	nodeconfig := adapters.RandomNodeConfig()
+	addr := network.NewAddrFromNodeId(nodeconfig.Id)
+	pss := newTestPss(addr.OAddr)
+	
+	pt := p2ptest.NewProtocolTester(t, nodeconfig.Id, 2, pss.Protocols()[0].Run)
+	
+	return []p2ptest.Exchange{
+		p2ptest.Exchange{
+			Expects: []p2ptest.Expect{
+				p2ptest.Expect{
+					Code: 0,
+					Msg:  lhs,
+					Peer: id,
+				},
+			},
+			Triggers: []p2ptest.Trigger{
+				p2ptest.Trigger{
+					Code: 0,
+					Msg:  ,
+					Peer: id,
+				},
+			},
+		},
+	}
+}
+
+
+func TestPssFullRandom10_5_5(t *testing.T) {
+	adapter := adapters.NewSimAdapter(services)
+	testPssFullRandom(t, adapter, 10, 5, 5)
+}
+
+func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int, fullnodecount int, msgcount int) {
+	var lastid *adapters.NodeId = nil
+	nodeCount := 5
+	net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
+		Id:             "0",
+		DefaultService: bzzServiceName,
+	})
+	defer net.Shutdown()
+	
+	trigger := make(chan *adapters.NodeId)
+	ids := make([]*adapters.NodeId, nodeCount)
+	
+	for i := 0; i < nodeCount; i++ {
+		nodeconfig := adapters.RandomNodeConfig()
+		nodeconfig.Services = []string{"bzz", "pss"}
+		node, err := net.NewNodeWithConfig(nodeconfig)
+		if err != nil {
+			t.Fatalf("error starting node: %s", err)
+		}
+
+		if err := net.Start(node.ID()); err != nil {
+			t.Fatalf("error starting node %s: %s", node.ID().Label(), err)
+		}
+		
+		if err := triggerChecks(trigger, net, node.ID()); err != nil {
+			t.Fatal("error triggering checks for node %s: %s", node.ID().Label(), err)
+		}
+		ids[i] = node.ID()
+	}
+	
+	// run a simulation which connects the 10 nodes in a ring and waits
+	// for full peer discovery
+	action := func(ctx context.Context) error {
+		for i, id := range ids {
+			var peerId *adapters.NodeId
+			if i == 0 {
+				peerId = ids[len(ids)-1]
+			} else {
+				peerId = ids[i-1]
+			}
+			if err := net.Connect(id, peerId); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+	check := func(ctx context.Context, id *adapters.NodeId) (bool, error) {
+		select {
+		case <-ctx.Done():
+			return false, ctx.Err()
+		default:
+		}
+
+		node := net.GetNode(id)
+		if node == nil {
+			return false, fmt.Errorf("unknown node: %s", id)
+		}
+		client, err := node.Client()
+		if err != nil {
+			return false, fmt.Errorf("error getting node client: %s", err)
+		}
+		
+		log.Debug("in check", "node", id)
+		
+		if lastid != nil {
+			//msg := pssPingMsg{Created: time.Now(),}
+			client.CallContext(context.Background(), nil, "pss_sendRaw", topic, PssAPIMsg{
+				Addr: lastid.Bytes(),
+				Msg: []byte{1,2,3},
+			})
+		}
+		lastid = id
+	
+		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: ids,
+			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))
+	
+	time.Sleep(time.Second * 2)
+}
+
+// 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 {
+	node := net.GetNode(id)
+	if node == nil {
+		return fmt.Errorf("unknown node: %s", id)
+	}
+	client, err := node.Client()
+	if err != nil {
+		return err
+	}
+	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():
+				if err != nil {
+					log.Error(fmt.Sprintf("error getting peer events for node %v", id), "err", err)
+				}
+				return
+			}
+		}
+	}()
+	return nil
+}
+
+func newServices() adapters.Services {
+	
+	bzzs := make(map[*adapters.NodeId]*network.Bzz)
+	
+	adaptersservices := make(map[string]adapters.ServiceFunc)
+	
+	adaptersservices["bzz"] = func(id *adapters.NodeId) node.Service {
+		// setup hive
+		addr := network.NewAddrFromNodeId(id)
+
+		config := &network.BzzConfig{
+			OverlayAddr:  addr.Over(),
+			UnderlayAddr: addr.Under(),
+			KadParams:    network.NewKadParams(),
+			HiveParams:   network.NewHiveParams(),
+		}
+
+		config.KadParams.MinProxBinSize = 2
+		config.KadParams.MaxBinSize = 3
+		config.KadParams.MinBinSize = 1
+		config.KadParams.MaxRetries = 1000
+		config.KadParams.RetryExponent = 2
+		config.KadParams.RetryInterval = 1000000
+
+		config.HiveParams.KeepAliveInterval = time.Second
+
+		bzzs[id] = network.NewBzz(config)
+	
+		return bzzs[id]
+	}
+	
+	adaptersservices["pss"] = func(id *adapters.NodeId) node.Service {
+		// pss setup
+		cachedir, err := ioutil.TempDir("", "pss-cache")
+		if err != nil {
+			log.Error("create pss cache tmpdir failed", "error", err)
+			return nil
+		}
+		dpa, err := storage.NewLocalDPA(cachedir)
+		if err != nil {
+			log.Error("local dpa creation failed", "error", err)
+			return nil
+		}
+		pssp := NewPssParams()
+		return NewPss(bzzs[id].Kademlia, dpa, pssp)
+	}
+	
+	return adaptersservices
+}
+
 /*
 // example protocol implementation peer
 // message handlers are methods of this
@@ -127,143 +503,10 @@ func (self *pssTestService) Run(peer *bzzPeer) error {
 	defer self.node.Remove(peer)
 	return peer.Run(self.msgFunc)
 }
+*/
 
-func TestPssCache(t *testing.T) {
+/*
-	var err error
-	to, _ := hex.DecodeString("08090a0b0c0d0e0f1011121314150001020304050607161718191a1b1c1d1e1f")
-	oaddr, _ := hex.DecodeString("000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f")
-	uaddr, _ := hex.DecodeString("101112131415161718191a1b1c1d1e1f000102030405060708090a0b0c0d0e0f")
-	ps := makePss(oaddr)
-	pp := NewPssParams()
-	topic, _ := MakeTopic(protocolName, protocolVersion)
-	data := []byte("foo")
-	fwdaddr := RandomAddr()
-	msg := &PssMsg{
-		Payload: pssEnvelope{
-			TTL:         0,
-			SenderOAddr: oaddr,
-			// SenderUAddr: uaddr,
-			Topic:   topic,
-			Payload: data,
-		},
-	}
-	msg.SetRecipient(to)
 
-	msgtwo := &PssMsg{
-		Payload: pssEnvelope{
-			TTL:         0,
-			SenderOAddr: oaddr,
-			// SenderUAddr: oaddr,
-			Topic:   topic,
-			Payload: data,
-		},
-	}
-	msgtwo.SetRecipient(to)
-
-	digest := ps.hashMsg(msg)
-	digesttwo := ps.hashMsg(msgtwo)
-
-	if digest != 3595343914 {
-		t.Fatalf("digest - got: %d, expected: %d", digest, 3595343914)
-	}
-
-	if digest == digesttwo {
-		t.Fatalf("different msgs return same crc: %d", digesttwo)
-	}
-
-	// check the sender cache
-	err = ps.addFwdCacheSender(fwdaddr.Over(), digest)
-	if err != nil {
-		t.Fatalf("write to pss sender cache failed: %v", err)
-	}
-
-	if !ps.checkFwdCache(fwdaddr.Over(), digest) {
-		t.Fatalf("message %v should have SENDER record in cache but checkCache returned false", msg)
-	}
-
-	if ps.checkFwdCache(fwdaddr.Over(), digesttwo) {
-		t.Fatalf("message %v should NOT have SENDER record in cache but checkCache returned true", msgtwo)
-	}
-
-	// check the expire cache
-	err = ps.addFwdCacheExpire(digest)
-	if err != nil {
-		t.Fatalf("write to pss expire cache failed: %v", err)
-	}
-
-	if !ps.checkFwdCache(nil, digest) {
-		t.Fatalf("message %v should have EXPIRE record in cache but checkCache returned false", msg)
-	}
-
-	if ps.checkFwdCache(nil, digesttwo) {
-		t.Fatalf("message %v should NOT have EXPIRE record in cache but checkCache returned true", msgtwo)
-	}
-
-	time.Sleep(pp.Cachettl)
-	if ps.checkFwdCache(nil, digest) {
-		t.Fatalf("message %v should have expired from cache but checkCache returned true", msg)
-	}
-
-	err = ps.AddToCache(fwdaddr.Over(), msgtwo)
-	if err != nil {
-		t.Fatalf("public accessor cache write failed: %v", err)
-	}
-
-	if !ps.checkFwdCache(fwdaddr.Over(), digesttwo) {
-		t.Fatalf("message %v should have SENDER record in cache but checkCache returned false", msgtwo)
-	}
-}
-
-func TestPssRegisterHandler(t *testing.T) {
-	var topic PssTopic
-	var err error
-	addr := RandomAddr()
-	ps := makePss(addr.Under())
-
-	topic, _ = MakeTopic(protocolName, protocolVersion)
-	err = ps.Register(topic, func(msg []byte, p *p2p.Peer, sender []byte) error { return nil })
-	if err != nil {
-		t.Fatalf("couldnt register protocol 'foo' v 42: %v", err)
-	}
-
-	topic, _ = MakeTopic(protocolName, protocolVersion)
-	err = ps.Register(topic, func(msg []byte, p *p2p.Peer, sender []byte) error { return nil })
-	if err == nil {
-		t.Fatalf("register protocol 'abc..789' v 65536 should have failed")
-	}
-}
-
-func TestPssFullRandom10_10_5(t *testing.T) {
-	testPssFullRandom(t, 10, 10, 5)
-}
-
-func TestPssFullRandom50_50_5(t *testing.T) {
-	testPssFullRandom(t, 50, 50, 5)
-}
-
-func TestPssFullRandom50_50_25(t *testing.T) {
-	testPssFullRandom(t, 50, 50, 25)
-}
-
-func TestPssFullRandom10_100_50(t *testing.T) {
-	testPssFullRandom(t, 10, 100, 50)
-}
-
-func TestPssFullRandom50_100_50(t *testing.T) {
-	testPssFullRandom(t, 50, 100, 50)
-}
-
-func TestPssFullRandom100_100_5(t *testing.T) {
-	testPssFullRandom(t, 100, 100, 5)
-}
-
-func TestPssFullRandom100_100_25(t *testing.T) {
-	testPssFullRandom(t, 100, 100, 25)
-}
-
-func TestPssFullRandom100_100_50(t *testing.T) {
-	testPssFullRandom(t, 100, 100, 50)
-}
 
 func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes int) {
 	var action func(ctx context.Context) error
@@ -970,13 +1213,73 @@ func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, trigge
 }
 
 func makePss(addr []byte) *Pss {
-	kp := NewKadParams()
+	
+	// set up storage
+	cachedir, err := ioutil.TempDir("", "pss-cache")
+	if err != nil {
+		log.Error("create pss cache tmpdir failed", "error", err)
+		os.Exit(1)
+	}
+	
+	dpa, err := storage.NewLocalDPA(cachedir)
+	if err != nil {
+		log.Error("local dpa creation failed", "error", err)
+		os.Exit(1)
+	}
+	// cannot use pyramidchunker as it still lacks joinfunc TestPssRegisterHandler(t *testing.T) {
+	addr := RandomAddr()
+	ps := newTestPss(addr.UnderlayAddr())
+	from := RandomAddr()
+	payload := []byte("payload")
+	topic := NewTopic(protocolName, protocolVersion)
+	checkMsg := func(msg []byte, p *p2p.Peer, sender []byte) error {
+		if !bytes.Equal(from.OverlayAddr(), sender) {
+			return fmt.Errorf("sender mismatch. expected %x, got %x", from.OverlayAddr(), sender)
+		}
+		if !bytes.Equal(msg, payload) {
+			return fmt.Errorf("sender mismatch. expected %x, got %x", msg, payload)
+		}
+		if !bytes.Equal(from.UnderlayAddr(), p.ID()) {
+			return fmt.Errorf("sender mismatch. expected %x, got %x", from.UnderlayAddr(), p.ID())
+		}
+	}
+	deregister := ps.Register(topic, checkMsg)
+	pssmsg := &PssMsg{Data: NewPssEnvelope(from, topic, payload)}
+	err = ps.Process(pssmsg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var i int
+	err = ps.Process(&PssMsg{Data: NewPssEnvelope(from, []byte("topic"), payload)})
+	expErr := ""
+	if err == nil || err.Error() != expErr {
+		t.Fatalf("unhandled topic expected %v, got %v", expErr, err)
+	}
+	deregister2 := ps.Register(topic, func(msg []byte, p *p2p.Peer, sender []byte) error { i++; return nil })
+	ps.Process(pssmsg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if i != 1 {
+		t.Fatalf("second registerer handler did not run")
+	}
+	deregister()
+	deregister2()
+	err = ps.Process(&PssMsg{Data: NewPssEnvelope(from, topic, payload)})
+	expErr = ""
+	if err == nil || err.Error() != expErr {
+		t.Fatalf("reregister handler expected %v, got %v", expErr, err)
+	}
+}
+	// dpa.Chunker = storage.NewPyramidChunker(storage.NewChunkerParams())
+	
+	kp := network.NewKadParams()
 	kp.MinProxBinSize = 3
 
 	pp := NewPssParams()
 
-	overlay := NewKademlia(addr, kp)
+	overlay := network.NewKademlia(addr, kp)
-	ps := NewPss(overlay, pp)
+	ps := NewPss(overlay, dpa, pp)
 	//overlay.Prune(time.Tick(time.Millisecond * 250))
 	return ps
 }

swarm/pss/pssapi.go

@@ -0,0 +1,76 @@
+package pss
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+// PssAPI is the RPC API module for Pss
+type PssAPI struct {
+	*Pss
+}
+
+// NewPssAPI constructs a PssAPI instance
+func NewPssAPI(ps *Pss) *PssAPI {
+	return &PssAPI{Pss: ps}
+}
+
+// PssAPIMsg is the type for messages, it extends the rlp encoded protocol Msg
+// with the Sender's overlay address
+type PssAPIMsg struct {
+	Msg  []byte
+	Addr []byte
+}
+
+// NewMsg API endpoint creates an RPC subscription
+func (pssapi *PssAPI) NewMsg(ctx context.Context, topic PssTopic) (*rpc.Subscription, error) {
+	notifier, supported := rpc.NotifierFromContext(ctx)
+	if !supported {
+		return nil, fmt.Errorf("Subscribe not supported")
+	}
+
+	psssub := notifier.CreateSubscription()
+	handler := func(msg []byte, p *p2p.Peer, from []byte) error {
+		apimsg := &PssAPIMsg{
+			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))
+		}
+		return nil
+	}
+	deregf := pssapi.Pss.Register(topic, handler)
+
+	go func() {
+		defer deregf()
+		//defer psssub.Unsubscribe()
+		select {
+		case err := <-psssub.Err():
+			log.Warn(fmt.Sprintf("caught subscription error in pss sub topic: %v", topic, err))
+		case <-notifier.Closed():
+			log.Warn(fmt.Sprintf("rpc sub notifier closed"))
+		}
+	}()
+
+	return psssub, nil
+}
+
+// SendRaw sends the message (serialised 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 {
+		return fmt.Errorf("send error: %v", err)
+	}
+	return fmt.Errorf("ok sent")
+}
+
+// BaseAddr gets our own overlayaddress
+func (pssapi *PssAPI) BaseAddr() ([]byte, error) { 
+	log.Warn("inside baseaddr")
+	return pssapi.Pss.Overlay.BaseAddr(), nil
+}

swarm/swarm.go

@@ -38,6 +38,7 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/fuse"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/storage"
+	"github.com/ethereum/go-ethereum/swarm/pss"
 )
 
 // the swarm stack