diff --git a/p2p/simulations/adapters/docker.go b/p2p/simulations/adapters/docker.go
index a9460df0df..7b58e7cfeb 100644
--- a/p2p/simulations/adapters/docker.go
+++ b/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 {
-		return nil, fmt.Errorf("unknown node service %q", config.Service)
+	for _, name := range config.Services {
+		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`,
-			dockerImage, n.Config.Node.Service, n.ID.String(),
+			`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.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"
 
diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go
index 88d99b22b5..376c03c6db 100644
--- a/p2p/simulations/adapters/exec.go
+++ b/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 {
-		return nil, fmt.Errorf("unknown node service %q", config.Service)
+	for _, name := range config.Services {
+		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()},
 	}
 }
 
diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go
index a714564019..5c1c38aec4 100644
--- a/p2p/simulations/adapters/inproc.go
+++ b/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 {
diff --git a/p2p/simulations/adapters/types.go b/p2p/simulations/adapters/types.go
index e7f15ba8f7..dacf3a2111 100644
--- a/p2p/simulations/adapters/types.go
+++ b/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
-	// the node (for SimNodes it should be the name of a service contained
-	// in SimAdapter.services, for other nodes it should be a service
+	// Services is the name of the services which should be run when starting
+	// the node (for SimNodes it should be the names of services contained
+	// 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
 }
diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go
index 77cd4b9cbd..84d06e089e 100644
--- a/p2p/simulations/network.go
+++ b/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 == "" {
-		conf.Service = self.DefaultService
+	if len(conf.Services) == 0 {
+		conf.Services = append(conf.Services, self.DefaultService)
 	}
 
 	_, found := self.nodeMap[id.NodeID]
diff --git a/p2p/testing/protocolsession.go b/p2p/testing/protocolsession.go
index 2de849885d..98c0bf7f2a 100644
--- a/p2p/testing/protocolsession.go
+++ b/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)
 	}
diff --git a/p2p/testing/protocoltester.go b/p2p/testing/protocoltester.go
index 3b104bd12f..e34fcf0265 100644
--- a/p2p/testing/protocoltester.go
+++ b/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)
diff --git a/swarm/network/pssapi.go b/swarm/network/pssapi.go
deleted file mode 100644
index 82b45e8b03..0000000000
--- a/swarm/network/pssapi.go
+++ /dev/null
@@ -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")
-}
diff --git a/swarm/pss/client.go b/swarm/pss/client.go
new file mode 100644
index 0000000000..d71d7928a9
--- /dev/null
+++ b/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")
+}
diff --git a/swarm/pss/client_test.go b/swarm/pss/client_test.go
new file mode 100644
index 0000000000..05fcd2f542
--- /dev/null
+++ b/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
+} 
diff --git a/swarm/pss/common.go b/swarm/pss/common.go
new file mode 100644
index 0000000000..7b985a6fea
--- /dev/null
+++ b/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
+}
+
diff --git a/swarm/network/pss.go b/swarm/pss/pss.go
similarity index 50%
rename from swarm/network/pss.go
rename to swarm/pss/pss.go
index d713c1d764..dea193e89a 100644
--- a/swarm/network/pss.go
+++ b/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
-}
-
-// 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
+	Payload *PssEnvelope
 }
 
 // String representation of PssMsg
 func (self *PssMsg) String() string {
-	return fmt.Sprintf("PssMsg: Recipient: %x", common.ByteLabel(self.GetRecipient()))
-}
-
-// 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
+	return fmt.Sprintf("PssMsg: Recipient: %x", common.ByteLabel(self.To))
 }
 
 // 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
-type pssDigest uint32
+func (self *PssTopic) String() string {
+	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
-
-	//peerPool map[pot.Address]map[PssTopic]*PssReadWriter // keep track of all virtual p2p.Peers we are currently speaking to
-	peerPool map[pot.Address]map[PssTopic]p2p.MsgReadWriter     // keep track of all virtual p2p.Peers we are currently speaking to
-	handlers map[PssTopic]func([]byte, *p2p.Peer, []byte) error // topic and version based pss payload handlers
-	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
+	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
+	handlers map[PssTopic]map[*pssHandler]bool              // topic and version based pss payload handlers
+	fwdcache map[pssDigest]pssCacheEntry                    // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg
+	cachettl time.Duration                                  // how long to keep messages in fwdcache
 	lock     sync.Mutex
+	dpa		*storage.DPA
 }
 
-func (self *Pss) hashMsg(msg *PssMsg) pssDigest {
-	hasher := self.hasher("SHA3")()
-	hasher.Reset()
-	hasher.Write(msg.GetRecipient())
-	// hasher.Write(msg.Payload.SenderUAddr)
-	hasher.Write(msg.Payload.SenderOAddr)
-	hasher.Write(msg.Payload.Topic[:])
-	hasher.Write(msg.Payload.Payload)
-	b := hasher.Sum([]byte{})
-	return pssDigest(binary.BigEndian.Uint32(b))
+func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
+	swg := &sync.WaitGroup{}
+	wwg := &sync.WaitGroup{}
+	buf := bytes.NewReader(msg.serialize())
+	key, err := self.dpa.Store(buf, int64(buf.Len()), swg, wwg)
+	if err != nil {
+		log.Warn("Could not store in swarm", "err", err)
+		return pssDigest{}, err
+	}
+	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
-func NewPss(k Overlay, params *PssParams) *Pss {
-	baseAddr := k.BaseAddr()
+// 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]*PssReadWriter, PssPeerCapacity),
 		peerPool: make(map[pot.Address]map[PssTopic]p2p.MsgReadWriter, PssPeerCapacity),
-		handlers: make(map[PssTopic]func([]byte, *p2p.Peer, []byte) error),
-		events:   make(map[PssTopic]*event.Feed),
+		handlers: make(map[PssTopic]map[*pssHandler]bool),
 		fwdcache: make(map[pssDigest]pssCacheEntry),
 		cachettl: params.Cachettl,
-		hasher:   storage.MakeHashFunc,
-		baseAddr: baseAddr,
+		dpa: dpa,
 	}
 }
 
-func (p *Pss) Run(peer *bzzPeer) 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
+func (self *Pss) Start(srv *p2p.Server) error {
 	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
-// 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 {
+func (self *Pss) getHandlers(topic PssTopic) map[*pssHandler]bool {
 	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) {
-	self.peerPool[id][topic] = nil
-	return
+func (self *Pss) removePeerTopic(rw p2p.MsgReadWriter, topic PssTopic) {
+	prw, ok := rw.(*PssReadWriter)
+	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 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
+	return self.peerPool[id][topic] != 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
-	LastActive     time.Time
-	rw             chan p2p.Msg
-	spec           *protocols.Spec
-	topic          *PssTopic
+	To         pot.Address
+	LastActive time.Time
+	rw         chan p2p.Msg
+	spec		*protocols.Spec
+	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, _ := makeMsg(msg.Code, ifc)
-
-	return prw.Pss.Send(to, *prw.topic, pmsg)
+	pmsg, err := newProtocolMsg(msg.Code, ifc)
+	if err != nil {
+		return err
+	}
+	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 {
-	pp := &pssProtocol{
+func NewPssProtocol(pss *Pss, topic *PssTopic, spec *protocols.Spec, targetprotocol *p2p.Protocol) *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
-//
-// 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 {
+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,
-			RecipientOAddr: hashoaddr,
-			rw:             make(chan p2p.Msg),
-			spec:           self.spec,
-			topic:          self.topic,
+			Pss:   self.Pss,
+			To:    hashoaddr,
+			rw:    make(chan p2p.Msg),
+			spec:	self.spec,
+			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{}
-	rlp.DecodeBytes(msg, payload)
-
-	pmsg := p2p.Msg{
-		Code:       payload.Code,
-		Size:       uint32(len(payload.Data)),
-		ReceivedAt: time.Now(),
-		Payload:    bytes.NewBuffer(payload.Data),
+	pmsg, err := ToP2pMsg(msg)
+	if err != nil {
+		return fmt.Errorf("could not decode pssmsg")
 	}
 
 	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 {
-	if bytes.Equal(msg.GetRecipient(), self.baseAddr) {
-		return true
-	}
-	return false
+func (self *Pss) isSelfRecipient(msg *PssMsg) bool {
+	return bytes.Equal(msg.To, self.Overlay.BaseAddr())
 }
 
-func (self *Pss) GetPingHandler() func([]byte, *p2p.Peer, []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) {
+func newProtocolMsg(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)
-	if err != nil {
-		return nil, err
-	}
-
-	return rlpbundle, nil
+	return rlp.EncodeToBytes(smsg)
 }
 
-// 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) {
-	t := [TopicLength]byte{}
-	if len(s)+4 <= TopicLength {
-		copy(t[4:len(s)+4], s)
-	} else {
-		return t, fmt.Errorf("topic '%t' too long", s)
-	}
-	binary.PutVarint(t[:4], int64(v))
-	return t, nil
+func NewTopic(s string, v int) (topic PssTopic) {
+	h := sha3.NewKeccak256()
+	h.Write([]byte(s))
+	buf := make([]byte, TopicLength / 8)
+	binary.PutUvarint(buf, uint64(v))
+	h.Write(buf)
+	copy(topic[:], h.Sum(buf)[:])
+	return topic
+}
+
+
+func ToP2pMsg(msg []byte) (p2p.Msg, error) {
+	payload := &PssProtocolMsg{}
+	if err := rlp.DecodeBytes(msg, payload); err != nil {
+		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
 }
diff --git a/swarm/network/pss_test.go b/swarm/pss/pss_test.go
similarity index 73%
rename from swarm/network/pss_test.go
rename to swarm/pss/pss_test.go
index 3c059ac972..d7c371ad5e 100644
--- a/swarm/network/pss_test.go
+++ b/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"
-	protocolVersion = 42
+	pssServiceName 	= "pss"
+	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)
-	ps := NewPss(overlay, pp)
+	overlay := network.NewKademlia(addr, kp)
+	ps := NewPss(overlay, dpa, pp)
 	//overlay.Prune(time.Tick(time.Millisecond * 250))
 	return ps
 }
diff --git a/swarm/pss/pssapi.go b/swarm/pss/pssapi.go
new file mode 100644
index 0000000000..2943d8d9ea
--- /dev/null
+++ b/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
+}
diff --git a/swarm/swarm.go b/swarm/swarm.go
index 3bad8b8e1e..6676a4e944 100644
--- a/swarm/swarm.go
+++ b/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