swarm/network: pss 0.1

General:
    - Message encapsulation is temporary until integration with whisper

    Incoming message handling:
    - Implements a dispatcher for registering handler functions to pss topics
    - Handlers can be any sort of function, and can send/receive on p2p.MsgReadWriter
    - Actual p2p.Protocol handling now as pluggable convenience function

Sending and routing:
    - Added temporary cache solution based on message hash digest
    - Send and forward split up as separate methods
    - Send is now payload agnostic (p2p.Protocol replies massaged in p2p.MsgWriter)

Tests:
    - TestPssRandom...
	Sends PSS between randomly selected nodes in different population magnitudes
    - TestPssFullLinearEcho
        A->B->C->B->A ping/pong over pss, where A and C are running pss
    - TestPssFullRandom50Pct
        10 nodes where 5 are running pss, 5 sends to/from random nodes
        (messages currently get stuck in kad. routing, discarded by cache block)
    - Protocoltester tests status unknown

cmd/swarm/main.go

@@ -116,6 +116,9 @@ var (
 	SwarmUploadMimeType = cli.StringFlag{
 		Name:  "mime",
 		Usage: "force mime type",
+	PssEnabledFlag = cli.BoolFlag{
+		Name: "pss",
+		Usage: "Enable pss (message passing over swarm)",
 	}
 	CorsStringFlag = cli.StringFlag{
 		Name:  "corsdomain",
@@ -260,6 +263,8 @@ Cleans database of corrupted entries.
 		SwarmUploadDefaultPath,
 		SwarmUpFromStdinFlag,
 		SwarmUploadMimeType,
+		// pss flags
+		PssEnabledFlag,
 	}
 	app.Flags = append(app.Flags, debug.Flags...)
 	app.Before = func(ctx *cli.Context) error {
@@ -347,7 +352,8 @@ func registerBzzService(ctx *cli.Context, stack *node.Node) {
 	}
 	swapEnabled := ctx.GlobalBool(SwarmSwapEnabledFlag.Name)
 	syncEnabled := ctx.GlobalBoolT(SwarmSyncEnabledFlag.Name)
-
+	pssEnabled := ctx.GlobalBool(PssEnabledFlag.Name)
+	
 	ethapi := ctx.GlobalString(EthAPIFlag.Name)
 	cors := ctx.GlobalString(CorsStringFlag.Name)
 
@@ -361,7 +367,7 @@ func registerBzzService(ctx *cli.Context, stack *node.Node) {
 		} else {
 			swapEnabled = false
 		}
-		return swarm.NewSwarm(ctx, client, bzzconfig, swapEnabled, syncEnabled, cors)
+		return swarm.NewSwarm(ctx, client, bzzconfig, swapEnabled, syncEnabled, cors, pssEnabled)
 	}
 	if err := stack.Register(boot); err != nil {
 		utils.Fatalf("Failed to register the Swarm service: %v", err)

common/format.go

@@ -23,6 +23,10 @@ import (
 	"time"
 )
 
+const (
+	LabelLength    = 8
+)
+
 // PrettyDuration is a pretty printed version of a time.Duration value that cuts
 // the unnecessary precision off from the formatted textual representation.
 type PrettyDuration time.Duration
@@ -38,3 +42,14 @@ func (d PrettyDuration) String() string {
 	}
 	return label
 }
+
+// useful for segfault safe reduction of log clutter
+func ByteLabel(addr []byte) (b [LabelLength]byte) {
+	//b = make([LabelLength]byte)
+	if len(addr) < LabelLength {
+		copy(b[LabelLength - len(addr) - 1:len(addr)], addr[:])
+	} else {
+		copy(b[:], addr[:LabelLength])
+	}
+	return b
+}

p2p/protocols/protocol.go

@@ -32,7 +32,6 @@ package protocols
 import (
 	"fmt"
 	"reflect"
-	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
@@ -122,6 +121,15 @@ type CodeMap struct {
 	messages   map[reflect.Type]uint64 // index of types to codes, for sending by type
 }
 
+func (self *CodeMap) GetInterface(code uint64) (interface{}, bool) {
+	if int(code) > len(self.codes)-1 {
+		return nil, false
+	}
+	typ := self.codes[code]
+	val := reflect.New(typ)
+	return val.Interface(), true
+}
+
 func (self *CodeMap) GetCode(msg interface{}) (uint64, bool) {
 	code, found := self.messages[reflect.TypeOf(msg)]
 	return code, found
@@ -268,21 +276,8 @@ func (self *Peer) Send(msg interface{}) error {
 		return errorf(ErrInvalidMsgType, "%v", code)
 	}
 	log.Trace(fmt.Sprintf("=> msg #%d TO %v : %v", code, self.ID(), msg))
-	go func() {
+	go p2p.Send(self.rw, uint64(code), msg)
-		self.wErrc <- p2p.Send(self.rw, uint64(code), msg)
+	return nil
-	}()
-	var err error
-	select {
-	case err = <-self.wErrc:
-		if err == nil {
-			return nil
-		}
-	case <-time.NewTimer(3000 * time.Millisecond).C:
-		err = fmt.Errorf("write timeout")
-	}
-	err = errorf(ErrWrite, "(msg code: %v): %v", code, err)
-	self.Drop(err)
-	return err
 }
 
 func (self *Peer) DisconnectHook(f func(error)) {

p2p/testing/protocolsession.go

@@ -2,6 +2,9 @@ package testing
 
 import (
 	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -13,6 +16,7 @@ import (
 type ProtocolSession struct {
 	TestNodeAdapter
 	Ids []*adapters.NodeId
+	ignore []uint64
 }
 
 type TestMessenger interface {
@@ -21,6 +25,7 @@ type TestMessenger interface {
 }
 
 type TestNodeAdapter interface {
+	p2p.Server
 	GetPeer(id *adapters.NodeId) *adapters.Peer
 }
 
@@ -64,6 +69,10 @@ func NewProtocolSession(na TestNodeAdapter, ids []*adapters.NodeId) *ProtocolSes
 	return ps
 }
 
+func (self *ProtocolSession) SetIgnoreCodes(ignore ...uint64) {
+	self.ignore = ignore
+}
+
 // trigger sends messages from peers
 func (self *ProtocolSession) trigger(trig Trigger) error {
 	peer := self.GetPeer(trig.Peer)
@@ -109,8 +118,39 @@ func (self *ProtocolSession) expect(exp Expect) error {
 
 	errc := make(chan error)
 	go func() {
-		log.Trace(fmt.Sprintf("waiting for msg, %v", exp.Msg))
+		var err error
-		errc <- p2p.ExpectMsg(peer, exp.Code, exp.Msg)
+		ignored := true
+		log.Trace("waiting for msg", "code", exp.Code, "msg", exp.Msg)
+		for ignored {
+			ignored = false
+			err = p2p.ExpectMsg(peer, exp.Code, exp.Msg)
+			// frail, but we can't know what code expectmsg got otherwise
+			// can we do better error reporting in p2p.ExpectMsg()?
+			if err != nil {
+				if strings.Contains(err.Error(), "code") {
+					re, _ := regexp.Compile("got ([0-9]+),")
+					match := re.FindStringSubmatch(err.Error())
+					if len(match) > 1 {
+						for _, codetoignore := range self.ignore {
+							codewegot, err := strconv.ParseUint(match[1], 10, 64)
+							if err == nil {
+								if codetoignore == codewegot {
+									ignored = true
+									log.Trace("ignore msg with wrong code", "received", codewegot, "expected", exp.Code)
+									break
+								}
+							} else {
+								log.Warn("expectmsg errormsg parse error?!")
+							}
+						} 
+					} else {
+						log.Warn("expectmsg errormsg parse error?!")
+						break
+					}
+				}
+			}
+		}
+		errc <- err
 	}()
 
 	t := exp.Timeout

swarm/api/config.go

@@ -47,13 +47,14 @@ type Config struct {
 	*storage.ChunkerParams
 	*network.HiveParams
 	Swap *swap.SwapParams
-	*network.SyncParams
+	//*network.SyncParams
 	Path      string
 	Port      string
 	PublicKey string
 	BzzKey    string
 	EnsRoot   common.Address
 	NetworkId uint64
+	*network.PssParams
 }
 
 // config is agnostic to where private key is coming from
@@ -72,10 +73,11 @@ func NewConfig(path string, contract common.Address, prvKey *ecdsa.PrivateKey, n
 	keyhex := crypto.Keccak256Hash(pubkey).Hex()
 
 	self = &Config{
-		SyncParams:    network.NewSyncParams(dirpath),
+		//SyncParams:    network.NewSyncParams(dirpath),
-		HiveParams:    network.NewHiveParams(dirpath),
+		HiveParams:    network.NewHiveParams(),
 		ChunkerParams: storage.NewChunkerParams(),
 		StoreParams:   storage.NewStoreParams(dirpath),
+		PssParams:	   network.NewPssParams(),
 		Port:          port,
 		Path:          dirpath,
 		Swap:          swap.DefaultSwapParams(contract, prvKey),

swarm/api/testapi.go

@@ -29,18 +29,6 @@ func NewControl(api *Api, hive *network.Hive) *Control {
 	return &Control{api, hive}
 }
 
-func (self *Control) BlockNetworkRead(on bool) {
-	self.hive.BlockNetworkRead(on)
-}
-
-func (self *Control) SyncEnabled(on bool) {
-	self.hive.SyncEnabled(on)
-}
-
-func (self *Control) SwapEnabled(on bool) {
-	self.hive.SwapEnabled(on)
-}
-
 func (self *Control) Hive() string {
 	return self.hive.String()
 }

swarm/api/ws/server.go

@@ -0,0 +1,54 @@
+package ws
+
+import (
+	"net"
+	
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+// Server is the basic configuration needs for the HTTP server and also
+// includes CORS settings.
+type Server struct {
+	//Addr       string
+	CorsString string
+	Endpoint	string
+}
+
+// startWS initializes and starts the websocket RPC endpoint.
+func StartWSServer(apis []rpc.API, server *Server) error {
+	
+	// Generate the whitelist based on the allowed modules
+	/*whitelist := make(map[string]bool)
+	for _, module := range modules {
+		whitelist[module] = true
+	}*/
+	// Register all the APIs exposed by the services
+	handler := rpc.NewServer()
+	for _, api := range apis {
+		//if whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {
+			if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
+				return err
+			}
+			glog.V(logger.Debug).Infof("WebSocket registered %T under '%s'", api.Service, api.Namespace)
+		//}
+	}
+	// All APIs registered, start the HTTP listener
+	var (
+		listener net.Listener
+		err      error
+	)
+	if listener, err = net.Listen("tcp", server.Endpoint); err != nil {
+		return err
+	}
+	rpc.NewWSServer(server.CorsString, handler).Serve(listener)
+	glog.V(logger.Info).Infof("WebSocket endpoint opened: ws://%s", server.Endpoint)
+
+	// All listeners booted successfully
+	//n.wsEndpoint = endpoint
+	//n.wsListener = listener
+	//n.wsHandler = handler
+
+	return nil
+}

swarm/api/ws/server_test.go

@@ -0,0 +1,62 @@
+package ws
+
+import (
+	"context"
+	"testing"
+	"time"
+	
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+func init() {
+	glog.SetV(logger.Detail)
+	glog.SetToStderr(true)
+}
+
+type TestResult struct {
+	Foo string `json:"foo"`
+}
+
+func TestStartWSServer(t *testing.T) {
+	ep := "localhost:8099"
+	server := &Server{
+		Endpoint: ep,
+		CorsString: "*",
+	}
+	apis := []rpc.API{
+		{
+			Namespace: "pss",
+			Version:   "0.1",
+			Service:   makeFakeAPIHandler(),
+			Public:    true,
+		},
+	}
+	go func() {
+		err := StartWSServer(apis, server)
+		t.Logf("wsserver exited: %v", err)
+	}()
+	
+	time.Sleep(time.Second)
+	
+	client, err := rpc.DialWebsocket(context.Background(), "ws://" + ep, "ws://localhost")
+	if err != nil {
+		t.Fatalf("could not connect: %v", err)
+	} else {
+		t.Logf("client: %v", client)
+		client.Call(&TestResult{}, "pss_test")
+	}
+	
+}
+
+func makeFakeAPIHandler() *FakeAPIHandler {
+	return &FakeAPIHandler{}
+}
+
+type FakeAPIHandler struct {
+}
+
+func (self *FakeAPIHandler) Test() {
+	glog.V(logger.Detail).Infof("in fakehandler Test()")
+}

swarm/network/discovery.go

@@ -53,7 +53,9 @@ func (self *discPeer) NotifyPeer(p Peer, po uint8) error {
 	resp := &peersMsg{
 		Peers: []*peerAddr{&peerAddr{OAddr: p.OverlayAddr(), UAddr: p.UnderlayAddr()}}, // perhaps the PeerAddr interface is unnecessary generalization
 	}
-	return self.Send(resp)
+	self.Send(resp)
+	//return err
+	return nil
 }
 
 // NotifyProx sends a subPeers Msg to the receiver notifying them about
@@ -61,7 +63,9 @@ func (self *discPeer) NotifyPeer(p Peer, po uint8) error {
 // or first empty row)
 // callback for overlay driver
 func (self *discPeer) NotifyProx(po uint8) error {
-	return self.Send(&subPeersMsg{ProxLimit: po})
+	self.Send(&subPeersMsg{ProxLimit: po})
+	//return err
+	return nil
 }
 
 /*
@@ -115,7 +119,7 @@ func (self *discPeer) handleSubPeersMsg(msg interface{}) error {
 	self.proxLimit = spm.ProxLimit
 	if !self.sentPeers {
 		var peers []*peerAddr
-		self.overlay.EachLivePeer(self.OverlayAddr(), 255, func(p Peer, po int) bool {
+		self.overlay.EachLivePeer(self.OverlayAddr(), 255, func(p Peer, po int, isproxbin bool) bool {
 			if uint8(po) < self.proxLimit {
 				return false
 			}
@@ -161,7 +165,7 @@ func (self *discPeer) handleGetPeersMsg(msg interface{}) error {
 	var peers []*peerAddr
 	req := msg.(*getPeersMsg)
 	i := 0
-	self.overlay.EachLivePeer(self.OverlayAddr(), int(req.Order), func(n Peer, po int) bool {
+	self.overlay.EachLivePeer(self.OverlayAddr(), int(req.Order), func(n Peer, po int, isproxbin bool) bool {
 		i++
 		// only send peers we have not sent before in this session
 		if self.seen(n) {
@@ -177,7 +181,8 @@ func (self *discPeer) handleGetPeersMsg(msg interface{}) error {
 	resp := &peersMsg{
 		Peers: peers,
 	}
-	return self.Send(resp)
+	self.Send(resp)
+	return nil
 }
 
 func RequestOrder(k Overlay, order, broadcastSize, maxPeers uint8) {
@@ -186,10 +191,10 @@ func RequestOrder(k Overlay, order, broadcastSize, maxPeers uint8) {
 		Max:   maxPeers,
 	}
 	var i uint8
-	var err error
+	//var err error
-	k.EachLivePeer(nil, 255, func(n Peer, po int) bool {
+	k.EachLivePeer(nil, 255, func(n Peer, po int, isproxbin bool) bool {
 		log.Trace(fmt.Sprintf("%T sent to %v", req, n.ID()))
-		err = n.Send(req)
+		err := n.Send(req)
 		if err == nil {
 			i++
 			if i >= broadcastSize {

swarm/network/hive.go

@@ -45,12 +45,13 @@ type Overlay interface {
 	On(Peer)
 	Off(Peer)
 
-	EachLivePeer([]byte, int, func(Peer, int) bool)
+	EachLivePeer([]byte, int, func(Peer, int, bool) bool)
 	EachPeer([]byte, int, func(PeerAddr, int) bool)
 
 	SuggestPeer() (PeerAddr, int, bool)
 
 	String() string
+	GetAddr() PeerAddr
 }
 
 // Hive implements the PeerPool interface
@@ -126,7 +127,7 @@ func (self *Hive) Start(server p2p.Server, af func() <-chan time.Time) error {
 
 			want = want && self.Discovery
 			if want {
-				RequestOrder(self.Overlay, uint8(order), self.PeersBroadcastSetSize, self.MaxPeersPerRequest)
+				go RequestOrder(self.Overlay, uint8(order), self.PeersBroadcastSetSize, self.MaxPeersPerRequest)
 			}
 
 			select {

swarm/network/hive_test.go

@@ -82,7 +82,8 @@ func TestRegisterAndConnect(t *testing.T) {
 		},
 		ticker: make(chan time.Time),
 	}
-	pp.Start(tc.connect, tc.ping)
+	//pp.Start(tc.connect, tc.ping)
+	pp.Start(s.TestNodeAdapter, tc.ping)
 	defer pp.Stop()
 	tc.ticker <- time.Now()
 

swarm/network/kademlia.go

@@ -196,7 +196,13 @@ func NewKadPeer(na PeerAddr) *KadPeer {
 	}
 }
 
-// Register enters each PeerAddr as kademlia peers into the
+// retrieve the base address
+// which is the overlayaddress used by peers to reach us
+func (self *Kademlia) GetAddr() PeerAddr {
+	return self.addr
+}
+
+// Register(nas) enters each PeerAddr as kademlia peers into the
 // database of known peers
 func (self *Kademlia) Register(nas ...PeerAddr) error {
 	label := fmt.Sprintf("%x", RandomAddr().OverlayAddr())
@@ -288,7 +294,7 @@ type ByteAddr struct {
 // EachLivePeer(base, po, f) is an iterator applying f to each live peer
 // that has proximity order po or less as measured from the base
 // if base is nil, kademlia base address is used
-func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer, int) bool) {
+func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer, int, bool) bool) {
 	var p pot.PotVal
 	if base == nil {
 		p = pot.PotVal(self.addr)
@@ -299,7 +305,11 @@ func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer, int) bool) {
 		if po > o {
 			return true
 		}
-		return f(val.(*KadPeer).Peer, po)
+		isproxbin := false
+		if l, _ := p.PO(val, 0); l >= self.proxLimit() {
+			isproxbin = true
+		}
+		return f(val.(*KadPeer).Peer, po, isproxbin)
 	})
 }
 

swarm/network/pss.go

@@ -2,50 +2,482 @@ package network
 
 import (
 	"bytes"
+	"encoding/binary"
+	"errors"
 	"fmt"
+	"sync"
+	"time"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/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/swarm/storage"
 )
 
-type Pss struct {
+const (
-	Overlay
+	DefaultTTL                  = 6000
-	LocalAddr []byte
+	TopicLength                 = 32
-	C         chan []byte
+	TopicResolverLength         = 8
+	PssPeerCapacity             = 256
+	PssPeerTopicDefaultCapacity = 8
+	digestLength                = 64
+	digestCapacity              = 256
+	defaultDigestCacheTTL       = time.Second
+)
+
+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
+type PssParams struct {
+	Cachettl time.Duration
 }
 
-func NewPss(k Overlay, addr []byte) *Pss {
+// Initializes default params for Pss
-	return &Pss{
+func NewPssParams() *PssParams {
-		Overlay:   k,
+	return &PssParams{
-		LocalAddr: addr,
+		Cachettl: defaultDigestCacheTTL,
-		C:         make(chan []byte),
 	}
 }
 
+// Encapsulates the message transported over pss.
+//
+// Warning: do not access the To-member directly. Use *PssMsg.GetRecipient() and *PssMsg.SetRecipient() instead.
 type PssMsg struct {
-	To   []byte
+	// (we need the To-member exported for type inference)
-	Data []byte
+	To      []byte
+	Payload pssEnvelope
 }
 
-func (pm *PssMsg) String() string {
+// Retrieve the remote peer receipient address of the message
-	return fmt.Sprintf("PssMsg: Recipient: %v", pm.To)
+func (self *PssMsg) GetRecipient() []byte {
+	return self.To
 }
 
-func (ps *Pss) HandlePssMsg(msg interface{}) error {
+// Set the remote peer recipient address of the message
-	pssmsg := msg.(*PssMsg)
+func (self *PssMsg) SetRecipient(to []byte) {
-	to := pssmsg.To
+	self.To = to
-	if bytes.Equal(to, ps.LocalAddr) {
+}
-		log.Trace(fmt.Sprintf("Pss to us, yay! %v", to))
+
-		ps.C <- pssmsg.Data
+// 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
+}
+
+// Topic defines the context of a message being transported over pss
+// It is used by pss to determine what action is to be taken on an incoming message
+// Typically, one can map protocol handlers for the message payloads by mapping topic to them; see *Pss.Register()
+type PssTopic [TopicLength]byte
+
+// Pre-Whisper placeholder
+type pssDigest uint32
+
+// 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:
+//
+// - access to the swarm overlay and routing (kademlia)
+// - a collection of remote overlay addresses mapped to MsgReadWriters, representing the virtually connected peers
+// - a collection of remote underlay address, mapped to the overlay addresses above
+// - a method to send a message to specific overlayaddr
+// - 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
+	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
+	lock     sync.Mutex
+}
+
+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))
+}
+
+// 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 {
+	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),
+		fwdcache: make(map[pssDigest]pssCacheEntry),
+		cachettl: params.Cachettl,
+		hasher:   storage.MakeHashFunc,
+	}
+}
+
+// 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)
+	return self.addFwdCacheSender(addr, digest)
+}
+
+func (self *Pss) addFwdCacheSender(addr []byte, digest pssDigest) error {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	var entry pssCacheEntry
+	var ok bool
+	if entry, ok = self.fwdcache[digest]; !ok {
+		entry = pssCacheEntry{}
+	}
+	entry.receivedFrom = addr
+	self.fwdcache[digest] = entry
+	return nil
+}
+
+func (self *Pss) addFwdCacheExpire(digest pssDigest) error {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	var entry pssCacheEntry
+	var ok bool
+	if entry, ok = self.fwdcache[digest]; !ok {
+		entry = pssCacheEntry{}
+	}
+	entry.expiresAt = time.Now().Add(self.cachettl)
+	self.fwdcache[digest] = entry
+	return nil
+}
+
+func (self *Pss) checkFwdCache(addr []byte, digest pssDigest) bool {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	entry, ok := self.fwdcache[digest]
+	if ok {
+		if entry.expiresAt.After(time.Now()) {
+			log.Debug(fmt.Sprintf("unexpired cache for digest %x", digest))
+			return true
+		} else if entry.expiresAt.IsZero() && bytes.Equal(addr, entry.receivedFrom) {
+			log.Debug(fmt.Sprintf("sendermatch %x for digest %x", common.ByteLabel(addr), digest))
+			return true
+		}
+	}
+	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] = handler
+	return nil
+}
+
+// Retrieves the handler function registered by *Pss.Register()
+func (self *Pss) GetHandler(topic PssTopic) func([]byte, *p2p.Peer, []byte) error {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	return self.handlers[topic]
+}
+
+// Links a pss peer address and topic to a dedicated p2p.MsgReadWriter in the pss peerpool, and runs the specificed protocol on this p2p.MsgReadWriter and the specified peer
+//
+// The effect is that now we have a "virtual" protocol running on an artificial p2p.Peer, which can be looked up and piped to through Pss using swarm overlay address and topic
+func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, protocall adapters.ProtoCall, topic PssTopic, rw p2p.MsgReadWriter) error {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	self.addPeerTopic(addr, topic, rw)
+	go func() {
+		err := protocall(p, rw)
+		log.Warn(fmt.Sprintf("pss vprotocol quit on addr %v topic %v: %v", addr, topic, err))
+	}()
+	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 {
+		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) isActive(id pot.Address, topic PssTopic) bool {
+	if self.peerPool[id][topic] == nil {
+		return false
+	}
+	return true
+}
+
+// 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.Overlay.GetAddr().OverlayAddr(),
+		SenderUAddr: self.Overlay.GetAddr().UnderlayAddr(),
+		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.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient())))
+		//return errorBlockByCache
 		return nil
 	}
 
-	ps.EachLivePeer(to, 255, func(p Peer, po int) bool {
+	// TODO:check integrity of message
-		err := p.Send(pssmsg)
+
-		if err != nil {
+	sent := 0
+
+	// send with kademlia
+	// find the closest peer to the recipient and attempt to send
+	self.Overlay.EachLivePeer(msg.GetRecipient(), 256, func(p Peer, po int, isproxbin bool) bool {
+		if self.checkFwdCache(p.OverlayAddr(), digest) {
+			log.Warn(fmt.Sprintf("BOUNCE DEFER PSS-relay FROM %x TO %x THRU %x:", common.ByteLabel(self.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.OverlayAddr())))
 			return true
 		}
-		return false
+		log.Warn(fmt.Sprintf("Attempting PSS-relay FROM %x TO %x THRU %x", common.ByteLabel(self.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.OverlayAddr())))
+		err := p.Send(msg)
+		if err != nil {
+			log.Warn(fmt.Sprintf("FAILED PSS-relay FROM %x TO %x THRU %x: %v", common.ByteLabel(self.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.OverlayAddr()), err))
+			return true
+		}
+		sent++
+		if bytes.Equal(msg.GetRecipient(), p.OverlayAddr()) || !isproxbin {
+			return false
+		}
+		log.Trace(fmt.Sprintf("%x is in proxbin, so we continue sending", common.ByteLabel(p.OverlayAddr())))
+		return true
 	})
+	if sent == 0 {
+		log.Warn("PSS Was not able to send to any peers")
+	} else {
+		self.addFwdCacheExpire(digest)
+	}
 
 	return nil
 }
+
+// Convenience object that:
+//
+// - allows passing of the unwrapped PssMsg payload to the p2p level message handlers
+// - interprets outgoing p2p.Msg from the p2p level to pass in to *Pss.Send()
+//
+// Implements p2p.MsgReadWriter
+type PssReadWriter struct {
+	*Pss
+	RecipientOAddr pot.Address
+	LastActive     time.Time
+	rw             chan p2p.Msg
+	ct             *protocols.CodeMap
+	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
+}
+
+// Implements p2p.MsgWriter
+func (prw PssReadWriter) WriteMsg(msg p2p.Msg) error {
+	log.Trace(fmt.Sprintf("pssrw writemsg: %v", msg))
+	ifc, found := prw.ct.GetInterface(msg.Code)
+	if !found {
+		return fmt.Errorf("Writemsg couldn't find matching interface for code %d", msg.Code)
+	}
+	msg.Decode(ifc)
+
+	to := prw.RecipientOAddr.Bytes()
+
+	pmsg, _ := makeMsg(msg.Code, ifc)
+
+	return prw.Pss.Send(to, *prw.topic, pmsg)
+}
+
+// Injects a p2p.Msg into the MsgReadWriter, so that it appears on the associated p2p.MsgReader
+func (prw PssReadWriter) injectMsg(msg p2p.Msg) error {
+	log.Trace(fmt.Sprintf("pssrw injectmsg: %v", msg))
+	prw.rw <- msg
+	return nil
+}
+
+// Convenience object for passing messages in and out of the p2p layer
+type PssProtocol struct {
+	*Pss
+	virtualProtocol *p2p.Protocol
+	topic           *PssTopic
+	ct              *protocols.CodeMap
+}
+
+// Constructor
+func NewPssProtocol(pss *Pss, topic *PssTopic, ct *protocols.CodeMap, targetprotocol *p2p.Protocol) *PssProtocol {
+	pp := &PssProtocol{
+		Pss:             pss,
+		virtualProtocol: targetprotocol,
+		topic:           topic,
+		ct:              ct,
+	}
+	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 {
+	hashoaddr := pot.NewHashAddressFromBytes(senderAddr).Address
+	if !self.isActive(hashoaddr, *self.topic) {
+		rw := &PssReadWriter{
+			Pss:            self.Pss,
+			RecipientOAddr: hashoaddr,
+			rw:             make(chan p2p.Msg),
+			ct:             self.ct,
+			topic:          self.topic,
+		}
+		self.Pss.AddPeer(p, hashoaddr, self.virtualProtocol.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),
+	}
+
+	vrw := self.Pss.peerPool[hashoaddr][*self.topic].(*PssReadWriter)
+	vrw.injectMsg(pmsg)
+
+	return nil
+}
+
+func (ps *Pss) isSelfRecipient(msg *PssMsg) bool {
+	if bytes.Equal(msg.GetRecipient(), ps.Overlay.GetAddr().OverlayAddr()) {
+		return true
+	}
+	return false
+}
+
+// Pre-Whisper placeholder
+func makeMsg(code uint64, msg interface{}) ([]byte, error) {
+
+	rlpdata, err := rlp.EncodeToBytes(msg)
+	if err != nil {
+		return nil, err
+	}
+
+	// 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{
+		Code: code,
+		Size: uint32(len(rlpdata)),
+		Data: rlpdata,
+	}
+
+	rlpbundle, err := rlp.EncodeToBytes(smsg)
+	if err != nil {
+		return nil, err
+	}
+
+	return rlpbundle, nil
+}
+
+// Compiles 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
+}

swarm/network/test_overlay.go

@@ -28,6 +28,13 @@ func (self *testOverlay) Register(nas ...PeerAddr) error {
 	return self.register(nas...)
 }
 
+func (self *testOverlay) GetAddr() PeerAddr {
+	return &peerAddr{
+		OAddr: self.addr,
+		UAddr: []byte{},
+	}
+}
+
 func (self *testOverlay) register(nas ...PeerAddr) error {
 	for _, na := range nas {
 		tna := &testPeerAddr{PeerAddr: na}
@@ -95,14 +102,14 @@ func (self *testOverlay) off(po []*testPeerAddr) (nas []PeerAddr) {
 	return nas
 }
 
-func (self *testOverlay) EachLivePeer(base []byte, o int, f func(Peer, int) bool) {
+func (self *testOverlay) EachLivePeer(base []byte, o int, f func(Peer, int, bool) bool) {
 	if base == nil {
 		base = self.addr
 	}
 	for i := o; i < len(self.pos); i++ {
 		for _, na := range self.pos[i] {
 			if na.Peer != nil {
-				if !f(na.Peer, o) {
+				if !f(na.Peer, o, false) {
 					return
 				}
 			}

swarm/swarm.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"fmt"
+	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
@@ -31,6 +32,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/adapters"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethereum/go-ethereum/swarm/api"
@@ -45,18 +47,21 @@ type Swarm struct {
 	config      *api.Config            // swarm configuration
 	api         *api.Api               // high level api layer (fs/manifest)
 	dns         api.Resolver           // DNS registrar
-	dbAccess    *network.DbAccess      // access to local chunk db iterator and storage counter
+	//dbAccess    *network.DbAccess      // access to local chunk db iterator and storage counter
 	storage     storage.ChunkStore     // internal access to storage, common interface to cloud storage backends
 	dpa         *storage.DPA           // distributed preimage archive, the local API to the storage with document level storage/retrieval support
-	depo        network.StorageHandler // remote request handler, interface between bzz protocol and the storage
+	//depo        network.StorageHandler // remote request handler, interface between bzz protocol and the storage
 	cloud       storage.CloudStore     // procurement, cloud storage backend (can multi-cloud)
 	hive        *network.Hive          // the logistic manager
 	backend     chequebook.Backend     // simple blockchain Backend
 	privateKey  *ecdsa.PrivateKey
 	corsString  string
 	swapEnabled bool
+	pssEnabled  bool
+	pss			*network.Pss
 	lstore      *storage.LocalStore // local store, needs to store for releasing resources after node stopped
 	sfs         *fuse.SwarmFS       // need this to cleanup all the active mounts on node exit
+	na			*adapters.NodeAdapter
 }
 
 type SwarmAPI struct {
@@ -75,7 +80,7 @@ func (self *Swarm) API() *SwarmAPI {
 
 // creates a new swarm service instance
 // implements node.Service
-func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.Config, swapEnabled, syncEnabled bool, cors string) (self *Swarm, err error) {
+func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.Config, swapEnabled, syncEnabled bool, cors string, pssEnabled bool) (self *Swarm, err error) {
 	if bytes.Equal(common.FromHex(config.PublicKey), storage.ZeroKey) {
 		return nil, fmt.Errorf("empty public key")
 	}
@@ -89,6 +94,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
 		backend:     backend,
 		privateKey:  config.Swap.PrivateKey(),
 		corsString:  cors,
+		pssEnabled:	pssEnabled,
 	}
 	log.Debug(fmt.Sprintf("Setting up Swarm service components"))
 
@@ -99,31 +105,35 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
 	}
 
 	// setup local store
-	log.Debug(fmt.Sprintf("Set up local storage"))
+	//glog.V(logger.Debug).Infof("Set up local storage")
-
-	self.dbAccess = network.NewDbAccess(self.lstore)
-	log.Debug(fmt.Sprintf("Set up local db access (iterator/counter)"))
 
+	//self.dbAccess = network.NewDbAccess(self.lstore)
+	glog.V(logger.Debug).Infof("Set up local db access (iterator/counter)")
+	
+	kp := network.NewKadParams()
+	
+	to := network.NewKademlia(
+		common.FromHex(config.BzzKey),
+		kp,
+	)
 	// set up the kademlia hive
 	self.hive = network.NewHive(
-		common.HexToHash(self.config.BzzKey), // key to hive (kademlia base address)
 		config.HiveParams,                    // configuration parameters
-		swapEnabled,                          // SWAP enabled
+		to,
-		syncEnabled,                          // syncronisation enabled
 	)
 	log.Debug(fmt.Sprintf("Set up swarm network with Kademlia hive"))
 
 	// setup cloud storage backend
-	cloud := network.NewForwarder(self.hive)
+	//cloud := network.NewForwarder(self.hive)
-	log.Debug(fmt.Sprintf("-> set swarm forwarder as cloud storage backend"))
+	//glog.V(logger.Debug).Infof("-> set swarm forwarder as cloud storage backend")
+	
 	// setup cloud storage internal access layer
-
+	self.storage = storage.NewNetStore(hash, self.lstore, nil, config.StoreParams)
-	self.storage = storage.NewNetStore(hash, self.lstore, cloud, config.StoreParams)
+	glog.V(logger.Debug).Infof("-> swarm net store shared access layer to Swarm Chunk Store")
-	log.Debug(fmt.Sprintf("-> swarm net store shared access layer to Swarm Chunk Store"))
 
 	// set up Depo (storage handler = cloud storage access layer for incoming remote requests)
-	self.depo = network.NewDepo(hash, self.lstore, self.storage)
+	// self.depo = network.NewDepo(hash, self.lstore, self.storage)
-	log.Debug(fmt.Sprintf("-> REmote Access to CHunks"))
+	// glog.V(logger.Debug).Infof("-> REmote Access to CHunks")
 
 	// set up DPA, the cloud storage local access layer
 	dpaChunkStore := storage.NewDpaChunkStore(self.lstore, self.storage)
@@ -188,13 +198,23 @@ func (self *Swarm) Start(net p2p.Server) error {
 	}
 
 	log.Warn(fmt.Sprintf("Starting Swarm service"))
+	
+	glog.V(logger.Warn).Infof("Starting Swarm service")
 	self.hive.Start(
-		discover.PubkeyID(&net.PrivateKey.PublicKey),
-		func() string { return net.ListenAddr },
 		connectPeer,
+		func () <-chan time.Time{
+			return time.NewTicker(time.Second).C
+		},
 	)
+
 	log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", self.hive.Addr()))
 
+	if self.pssEnabled {
+		pssparams := network.NewPssParams()
+		self.pss = network.NewPss(self.hive.Overlay, pssparams)
+		glog.V(logger.Info).Infof("Pss started: %v", self.pss)
+	}
+	
 	self.dpa.Start()
 	log.Debug(fmt.Sprintf("Swarm DPA started"))
 
@@ -235,11 +255,37 @@ func (self *Swarm) Stop() error {
 
 // implements the node.Service interface
 func (self *Swarm) Protocols() []p2p.Protocol {
-	proto, err := network.Bzz(self.depo, self.backend, self.hive, self.dbAccess, self.config.Swap, self.config.SyncParams, self.config.NetworkId)
+	//proto, err := network.Bzz(self.depo, self.backend, self.hive, self.dbAccess, self.config.Swap, self.config.SyncParams, self.config.NetworkId)
-	if err != nil {
+	ct := network.BzzCodeMap()
+	for _, m := range network.DiscoveryMsgs {
+		ct.Register(m)
+	}
+	if self.pssEnabled {
+		ct.Register(&network.PssMsg{})
+	}
+	
+	srv := func(p network.Peer) error {
+		if self.pssEnabled {
+			//p.Register(&PssMsg{}, self.pssFunc)
+			glog.V(logger.Warn).Infof("pss is enabled, but handler not yet implemented - it won't work yet, sorry")
+		}
+		self.hive.Add(p)
+		p.DisconnectHook(func(err error) {
+			self.hive.Remove(p)
+		})
 		return nil
 	}
-	return []p2p.Protocol{proto}
+	
+	proto := network.Bzz(
+		self.hive.Overlay.GetAddr().OverlayAddr(),
+		self.hive.Overlay.GetAddr().UnderlayAddr(),
+		ct,
+		srv,
+		nil,
+		nil,
+	)
+	
+	return []p2p.Protocol{*proto}
 }
 
 // implements node.Service
@@ -302,7 +348,7 @@ func (self *Swarm) SetChequebook(ctx context.Context) error {
 	}
 	log.Info(fmt.Sprintf("new chequebook set (%v): saving config file, resetting all connections in the hive", self.config.Swap.Contract.Hex()))
 	self.config.Save()
-	self.hive.DropAll()
+	//self.hive.DropAll()
 	return nil
 }
 
@@ -332,7 +378,7 @@ func NewLocalSwarm(datadir, port string) (self *Swarm, err error) {
 
 	return
 }
-
+	
 // serialisable info about swarm
 type Info struct {
 	*api.Config