diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 95cacd7da5..a2ce5de75a 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -104,8 +104,6 @@ type Pss struct { paddingByteSize int capstring string outbox chan *PssMsg - outboxCapacity int - outboxCounter int // keys and peers pubKeyPool map[string]map[Topic]*pssPeer // mapping of hex public keys to peer address by topic. @@ -152,7 +150,6 @@ func NewPss(k network.Overlay, params *PssParams) *Pss { paddingByteSize: defaultPaddingByteSize, capstring: cap.String(), outbox: make(chan *PssMsg, defaultOutboxCapacity), - outboxCapacity: defaultOutboxCapacity, pubKeyPool: make(map[string]map[Topic]*pssPeer), symKeyPool: make(map[string]map[Topic]*pssPeer), @@ -331,15 +328,16 @@ func (self *Pss) handlePssMsg(msg interface{}) error { var err error if !self.isSelfPossibleRecipient(pssmsg) { log.Trace("pss was for someone else :'( ... forwarding", "pss", common.ToHex(self.BaseAddr())) - if !self.enqueue(pssmsg) { - return errors.New("outbox full!") + if err := self.enqueue(pssmsg); err != nil { + return err } } log.Trace("pss for us, yay! ... let's process!", "pss", common.ToHex(self.BaseAddr())) - if !self.process(pssmsg) { - if !self.enqueue(pssmsg) { - return errors.New("outbox full!") + if err := self.process(pssmsg); err != nil { + qerr := self.enqueue(pssmsg) + if qerr != nil { + err = fmt.Errorf("%s + %s", err, qerr) } } return err @@ -351,7 +349,7 @@ func (self *Pss) handlePssMsg(msg interface{}) error { // Entry point to processing a message for which the current node can be the intended recipient. // Attempts symmetric and asymmetric decryption with stored keys. // Dispatches message to all handlers matching the message topic -func (self *Pss) process(pssmsg *PssMsg) bool { +func (self *Pss) process(pssmsg *PssMsg) error { var err error var recvmsg *whisper.ReceivedMessage var from *PssAddress @@ -363,7 +361,7 @@ func (self *Pss) process(pssmsg *PssMsg) bool { psstopic := Topic(envelope.Topic) if self.allowRaw && psstopic == rawTopic { self.executeHandlers(rawTopic, envelope.Data, nil, false, "") - return true + return nil } if len(envelope.AESNonce) > 0 { // detect symkey msg according to whisperv5/envelope.go:OpenSymmetric @@ -374,18 +372,17 @@ func (self *Pss) process(pssmsg *PssMsg) bool { } recvmsg, keyid, from, err = keyFunc(envelope) if err != nil { - return false + return errors.New("Decryption failed") } if len(pssmsg.To) < addressLength { - if !self.enqueue(pssmsg) { - log.Error("outbox full!") - return false + if err := self.enqueue(pssmsg); err != nil { + return err } } self.executeHandlers(psstopic, recvmsg.Payload, from, asymmetric, keyid) - return true + return nil } @@ -606,13 +603,14 @@ func (self *Pss) cleanKeys() (count int) { // SECTION: Message sending ///////////////////////////////////////////////////////////////////// -func (self *Pss) enqueue(msg *PssMsg) bool { +func (self *Pss) enqueue(msg *PssMsg) error { select { case self.outbox <- msg: - return true + return nil default: } - return true + + return errors.New("outbox full") } // Send a raw message (any encryption is responsibility of calling client) @@ -631,10 +629,7 @@ func (self *Pss) SendRaw(msg []byte, address PssAddress) error { }, } self.addFwdCache(pssmsg) - if !self.enqueue(pssmsg) { - return errors.New("outbox full!") - } - return nil + return self.enqueue(pssmsg) } // Send a message using symmetric encryption @@ -727,10 +722,7 @@ func (self *Pss) send(to []byte, topic Topic, msg []byte, asymmetric bool, key [ Expire: uint32(time.Now().Add(self.msgTTL).Unix()), Payload: envelope, } - if !self.enqueue(pssmsg) { - return errors.New("outbox full!") - } - return nil + return self.enqueue(pssmsg) } // Forwards a pss message to the peer(s) closest to the to recipient address in the PssMsg struct @@ -801,8 +793,8 @@ func (self *Pss) forward(msg *PssMsg) error { if sent == 0 { log.Debug("unable to forward to any peers") time.Sleep(time.Millisecond) - if !self.enqueue(msg) { - return errors.New("outbox full!") + if err := self.enqueue(msg); err != nil { + return err } } diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 9970881bd6..29961ee3ae 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -1192,7 +1192,7 @@ func benchmarkSymkeyBruteforceChangeaddr(b *testing.B) { } b.ResetTimer() for i := 0; i < b.N; i++ { - if !ps.process(pssmsgs[len(pssmsgs)-(i%len(pssmsgs))-1]) { + if err := ps.process(pssmsgs[len(pssmsgs)-(i%len(pssmsgs))-1]); err != nil { b.Fatalf("pss processing failed: %v", err) } } @@ -1274,7 +1274,7 @@ func benchmarkSymkeyBruteforceSameaddr(b *testing.B) { Payload: env, } for i := 0; i < b.N; i++ { - if !ps.process(pssmsg) { + if err := ps.process(pssmsg); err != nil { b.Fatalf("pss processing failed: %v", err) } }