diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index cd94741be3..ac14f7f9e4 100644 --- a/swarm/network/kademlia.go +++ b/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, - MinProxBinSize: 2, - MinBinSize: 2, - MaxBinSize: 4, - RetryInterval: 4200000000, // 4.2 sec - MaxRetries: 42, - RetryExponent: 2, + MaxProxDisplay: defaultMaxProxDisplay, + MinProxBinSize: defaultMinProxBinSize, + MinBinSize: defaultMinBinSize, + MaxBinSize: defaultMaxBinSize, + RetryInterval: defaultRetryInterval, + MaxRetries: defaultMaxRetries, + 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 diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index e1e24e1f54..16b1130516 100644 --- a/swarm/pss/pss.go +++ b/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), - allowRaw: params.AllowRaw, + handlers: make(map[Topic]map[*handler]bool), 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 - return func() { p.deregister(topic, &handler) } + handlers[hndlr] = true + 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 { - return errors.New("raw message support disabled") - } + // if !p.allowRaw { + // 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 { - err := (*f)(payload, peer, asymmetric, keyid) + for h := range handlers { + 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 { - return errors.New("Raw messages not enabled") - } + //if !p.allowRaw { + // return errors.New("Raw messages not enabled") + //} pssMsgParams := &msgParams{ raw: true, } diff --git a/swarm/pss/types.go b/swarm/pss/types.go index 56c2c51dc0..7dc7a7bc40 100644 --- a/swarm/pss/types.go +++ b/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