swarm/pss, swarm/network: WIP Rawmsg per handler + handler to struct

swarm/network/kademlia.go

@@ -29,6 +29,16 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/pot"
 )
 
+const (
+	defaultMaxProxDisplay = 16
+	defaultMinProxBinSize = 2
+	defaultMinBinSize     = 2
+	defaultMaxBinSize     = 4
+	defaultRetryInterval  = 4200000000 // 4.2 sec
+	defaultMaxRetries     = 42
+	defaultRetryExponent  = 2
+)
+
 /*
 
 Taking the proximity order relative to a fix point x classifies the points in
@@ -68,13 +78,13 @@ type KadParams struct {
 // NewKadParams returns a params struct with default values
 func NewKadParams() *KadParams {
 	return &KadParams{
-		MaxProxDisplay: 16,
+		MaxProxDisplay: defaultMaxProxDisplay,
-		MinProxBinSize: 2,
+		MinProxBinSize: defaultMinProxBinSize,
-		MinBinSize:     2,
+		MinBinSize:     defaultMinBinSize,
-		MaxBinSize:     4,
+		MaxBinSize:     defaultMaxBinSize,
-		RetryInterval:  4200000000, // 4.2 sec
+		RetryInterval:  defaultRetryInterval,
-		MaxRetries:     42,
+		MaxRetries:     defaultMaxRetries,
-		RetryExponent:  2,
+		RetryExponent:  defaultRetryExponent,
 	}
 }
 
@@ -289,6 +299,7 @@ func (k *Kademlia) On(p *Peer) (uint8, bool) {
 // neighbourhood depth on each change.
 // Not receiving from the returned channel will block On function
 // when the neighbourhood depth is changed.
+// TODO: Why is this exported, and if it should be; why can't we have more subscribers than one?
 func (k *Kademlia) NeighbourhoodDepthC() <-chan int {
 	k.lock.Lock()
 	defer k.lock.Unlock()
@@ -430,6 +441,12 @@ func (k *Kademlia) eachAddr(base []byte, o int, f func(*BzzAddr, int, bool) bool
 // the nearest neighbour set with cardinality >= MinProxBinSize
 // if there is altogether less than MinProxBinSize peers it returns 0
 // caller must hold the lock
+func (k *Kademlia) NeighbourhoodDepth() (depth int) {
+	k.lock.RLock()
+	defer k.lock.Unlock()
+	return k.neighbourhoodDepth()
+}
+
 func (k *Kademlia) neighbourhoodDepth() (depth int) {
 	if k.conns.Size() < k.MinProxBinSize {
 		return 0

swarm/pss/pss.go

@@ -136,9 +136,8 @@ type Pss struct {
 	symKeyDecryptCacheCapacity int       // max amount of symkeys to keep.
 
 	// message handling
-	handlers   map[Topic]map[*Handler]bool // topic and version based pss payload handlers. See pss.Handle()
+	handlers   map[Topic]map[*handler]bool // topic and version based pss payload handlers. See pss.Handle()
 	handlersMu sync.RWMutex
-	allowRaw   bool
 	hashPool   sync.Pool
 
 	// process
@@ -180,8 +179,7 @@ func NewPss(k *network.Kademlia, params *PssParams) (*Pss, error) {
 		symKeyDecryptCache:         make([]*string, params.SymKeyCacheCapacity),
 		symKeyDecryptCacheCapacity: params.SymKeyCacheCapacity,
 
-		handlers: make(map[Topic]map[*Handler]bool),
+		handlers: make(map[Topic]map[*handler]bool),
-		allowRaw: params.AllowRaw,
 		hashPool: sync.Pool{
 			New: func() interface{} {
 				return storage.MakeHashFunc(storage.DefaultHash)()
@@ -313,18 +311,18 @@ func (p *Pss) PublicKey() *ecdsa.PublicKey {
 //
 // Returns a deregister function which needs to be called to
 // deregister the handler,
-func (p *Pss) Register(topic *Topic, handler Handler) func() {
+func (p *Pss) Register(topic *Topic, hndlr *handler) func() {
 	p.handlersMu.Lock()
 	defer p.handlersMu.Unlock()
 	handlers := p.handlers[*topic]
 	if handlers == nil {
-		handlers = make(map[*Handler]bool)
+		handlers = make(map[*handler]bool)
 		p.handlers[*topic] = handlers
 	}
-	handlers[&handler] = true
+	handlers[hndlr] = true
-	return func() { p.deregister(topic, &handler) }
+	return func() { p.deregister(topic, hndlr) }
 }
-func (p *Pss) deregister(topic *Topic, h *Handler) {
+func (p *Pss) deregister(topic *Topic, hndlr *handler) {
 	p.handlersMu.Lock()
 	defer p.handlersMu.Unlock()
 	handlers := p.handlers[*topic]
@@ -332,11 +330,11 @@ func (p *Pss) deregister(topic *Topic, h *Handler) {
 		delete(p.handlers, *topic)
 		return
 	}
-	delete(handlers, h)
+	delete(handlers, hndlr)
 }
 
 // get all registered handlers for respective topics
-func (p *Pss) getHandlers(topic Topic) map[*Handler]bool {
+func (p *Pss) getHandlers(topic Topic) map[*handler]bool {
 	p.handlersMu.RLock()
 	defer p.handlersMu.RUnlock()
 	return p.handlers[topic]
@@ -392,15 +390,16 @@ func (p *Pss) process(pssmsg *PssMsg) error {
 	var payload []byte
 	var from *PssAddress
 	var asymmetric bool
+	var raw bool
 	var keyid string
 	var keyFunc func(envelope *whisper.Envelope) (*whisper.ReceivedMessage, string, *PssAddress, error)
 
 	envelope := pssmsg.Payload
 	psstopic := Topic(envelope.Topic)
 	if pssmsg.isRaw() {
-		if !p.allowRaw {
+		//		if !p.allowRaw {
-			return errors.New("raw message support disabled")
+		//			return errors.New("raw message support disabled")
-		}
+		//		}
 		payload = pssmsg.Payload.Data
 	} else {
 		if pssmsg.isSym() {
@@ -414,7 +413,6 @@ func (p *Pss) process(pssmsg *PssMsg) error {
 		if err != nil {
 			return errors.New("Decryption failed")
 		}
-		payload = recvmsg.Payload
 	}
 
 	if len(pssmsg.To) < addressLength {
@@ -422,19 +420,22 @@ func (p *Pss) process(pssmsg *PssMsg) error {
 			return err
 		}
 	}
-	p.executeHandlers(psstopic, payload, from, asymmetric, keyid)
+	p.executeHandlers(psstopic, payload, from, raw, asymmetric, keyid)
 
 	return nil
 
 }
 
-func (p *Pss) executeHandlers(topic Topic, payload []byte, from *PssAddress, asymmetric bool, keyid string) {
+func (p *Pss) executeHandlers(topic Topic, payload []byte, from *PssAddress, raw bool, asymmetric bool, keyid string) {
 	handlers := p.getHandlers(topic)
 	peer := p2p.NewPeer(enode.ID{}, fmt.Sprintf("%x", from), []p2p.Cap{})
-	for f := range handlers {
+	for h := range handlers {
-		err := (*f)(payload, peer, asymmetric, keyid)
+		if !h.raw && raw {
+			continue
+		}
+		err := (h.f)(payload, peer, asymmetric, keyid)
 		if err != nil {
-			log.Warn("Pss handler %p failed: %v", f, err)
+			log.Warn("Pss handler %p failed: %v", h.f, err)
 		}
 	}
 }
@@ -684,9 +685,9 @@ func (p *Pss) enqueue(msg *PssMsg) error {
 //
 // Will fail if raw messages are disallowed
 func (p *Pss) SendRaw(address PssAddress, topic Topic, msg []byte) error {
-	if !p.allowRaw {
+	//if !p.allowRaw {
-		return errors.New("Raw messages not enabled")
+	//	return errors.New("Raw messages not enabled")
-	}
+	//}
 	pssMsgParams := &msgParams{
 		raw: true,
 	}

swarm/pss/types.go

@@ -159,9 +159,34 @@ func (msg *PssMsg) String() string {
 }
 
 // Signature for a message handler function for a PssMsg
-//
 // Implementations of this type are passed to Pss.Register together with a topic,
-type Handler func(msg []byte, p *p2p.Peer, asymmetric bool, keyid string) error
+type HandlerFunc func(msg []byte, p *p2p.Peer, asymmetric bool, keyid string) error
+
+// Handler defines code to be executed upon reception of content.
+type handler struct {
+	f    HandlerFunc
+	raw  bool // if true, will allow raw messages to be handled
+	prox bool // if true, explicit recipient address will be truncated to minproxsize depth
+}
+
+// NewHandler returns a new message handler
+func NewHandler(f HandlerFunc) *handler {
+	return &handler{
+		f: f,
+	}
+}
+
+// WithRaw is a chainable method that allows raw messages to be handled.
+func (h *handler) WithRaw() *handler {
+	h.raw = true
+	return h
+}
+
+// WithProxBin is a chainable method that allows sending messages with full addresses to neighbourhoods using the kademlia depth as reference
+func (h *handler) WithProxBin() *handler {
+	h.prox = true
+	return h
+}
 
 // the stateStore handles saving and loading PSS peers and their corresponding keys
 // it is currently unimplemented