aboutsummaryrefslogtreecommitdiffstats
path: root/swarm/pss
diff options
context:
space:
mode:
authorAnton Evangelatov <anton.evangelatov@gmail.com>2018-07-13 23:40:28 +0800
committerBalint Gabor <balint.g@gmail.com>2018-07-13 23:40:28 +0800
commit7c9314f231a7ddffbbbc5fec16c65519a0121eeb (patch)
treedbc4021b66ee8968ad747036741fac7e1b972a39 /swarm/pss
parentf7d3678c28c4b92e45a458e4785bd0f1cdc20e34 (diff)
downloaddexon-7c9314f231a7ddffbbbc5fec16c65519a0121eeb.tar.gz
dexon-7c9314f231a7ddffbbbc5fec16c65519a0121eeb.tar.zst
dexon-7c9314f231a7ddffbbbc5fec16c65519a0121eeb.zip
swarm: integrate OpenTracing; propagate ctx to internal APIs (#17169)
* swarm: propagate ctx, enable opentracing * swarm/tracing: log error when tracing is misconfigured
Diffstat (limited to 'swarm/pss')
-rw-r--r--swarm/pss/ping.go5
-rw-r--r--swarm/pss/pss.go7
-rw-r--r--swarm/pss/pss_test.go12
3 files changed, 13 insertions, 11 deletions
diff --git a/swarm/pss/ping.go b/swarm/pss/ping.go
index 2ef072918..ff635f40a 100644
--- a/swarm/pss/ping.go
+++ b/swarm/pss/ping.go
@@ -19,6 +19,7 @@
package pss
import (
+ "context"
"errors"
"time"
@@ -40,7 +41,7 @@ type Ping struct {
InC chan bool // optional, report back to calling code
}
-func (p *Ping) pingHandler(msg interface{}) error {
+func (p *Ping) pingHandler(ctx context.Context, msg interface{}) error {
var pingmsg *PingMsg
var ok bool
if pingmsg, ok = msg.(*PingMsg); !ok {
@@ -80,7 +81,7 @@ func NewPingProtocol(ping *Ping) *p2p.Protocol {
for {
select {
case ispong := <-ping.OutC:
- pp.Send(&PingMsg{
+ pp.Send(context.TODO(), &PingMsg{
Created: time.Now(),
Pong: ispong,
})
diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go
index dd081e93a..5c060b248 100644
--- a/swarm/pss/pss.go
+++ b/swarm/pss/pss.go
@@ -18,6 +18,7 @@ package pss
import (
"bytes"
+ "context"
"crypto/ecdsa"
"crypto/rand"
"errors"
@@ -71,7 +72,7 @@ type senderPeer interface {
Info() *p2p.PeerInfo
ID() discover.NodeID
Address() []byte
- Send(interface{}) error
+ Send(context.Context, interface{}) error
}
// per-key peer related information
@@ -344,7 +345,7 @@ func (p *Pss) getHandlers(topic Topic) map[*Handler]bool {
// Check if address partially matches
// If yes, it CAN be for us, and we process it
// Only passes error to pss protocol handler if payload is not valid pssmsg
-func (p *Pss) handlePssMsg(msg interface{}) error {
+func (p *Pss) handlePssMsg(ctx context.Context, msg interface{}) error {
metrics.GetOrRegisterCounter("pss.handlepssmsg", nil).Inc(1)
pssmsg, ok := msg.(*PssMsg)
@@ -844,7 +845,7 @@ func (p *Pss) forward(msg *PssMsg) error {
p.fwdPoolMu.RUnlock()
// attempt to send the message
- err := pp.Send(msg)
+ err := pp.Send(context.TODO(), msg)
if err != nil {
metrics.GetOrRegisterCounter("pss.pp.send.error", nil).Inc(1)
log.Error(err.Error())
diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go
index c738247f1..41b03db28 100644
--- a/swarm/pss/pss_test.go
+++ b/swarm/pss/pss_test.go
@@ -334,7 +334,7 @@ func TestHandlerConditions(t *testing.T) {
Data: []byte{0x66, 0x6f, 0x6f},
},
}
- if err := ps.handlePssMsg(msg); err != nil {
+ if err := ps.handlePssMsg(context.TODO(), msg); err != nil {
t.Fatal(err.Error())
}
tmr := time.NewTimer(time.Millisecond * 100)
@@ -351,7 +351,7 @@ func TestHandlerConditions(t *testing.T) {
// message should pass and queue due to partial length
msg.To = addr[0:1]
msg.Payload.Data = []byte{0x78, 0x79, 0x80, 0x80, 0x79}
- if err := ps.handlePssMsg(msg); err != nil {
+ if err := ps.handlePssMsg(context.TODO(), msg); err != nil {
t.Fatal(err.Error())
}
tmr.Reset(time.Millisecond * 100)
@@ -374,7 +374,7 @@ func TestHandlerConditions(t *testing.T) {
// full address mismatch should put message in queue
msg.To[0] = 0xff
- if err := ps.handlePssMsg(msg); err != nil {
+ if err := ps.handlePssMsg(context.TODO(), msg); err != nil {
t.Fatal(err.Error())
}
tmr.Reset(time.Millisecond * 10)
@@ -397,7 +397,7 @@ func TestHandlerConditions(t *testing.T) {
// expired message should be dropped
msg.Expire = uint32(time.Now().Add(-time.Second).Unix())
- if err := ps.handlePssMsg(msg); err != nil {
+ if err := ps.handlePssMsg(context.TODO(), msg); err != nil {
t.Fatal(err.Error())
}
tmr.Reset(time.Millisecond * 10)
@@ -417,7 +417,7 @@ func TestHandlerConditions(t *testing.T) {
}{
pssMsg: &PssMsg{},
}
- if err := ps.handlePssMsg(fckedupmsg); err == nil {
+ if err := ps.handlePssMsg(context.TODO(), fckedupmsg); err == nil {
t.Fatalf("expected error from processMsg but error nil")
}
@@ -427,7 +427,7 @@ func TestHandlerConditions(t *testing.T) {
ps.outbox <- msg
}
msg.Payload.Data = []byte{0x62, 0x61, 0x72}
- err = ps.handlePssMsg(msg)
+ err = ps.handlePssMsg(context.TODO(), msg)
if err == nil {
t.Fatal("expected error when mailbox full, but was nil")
}