aboutsummaryrefslogtreecommitdiffstats
path: root/whisper/whisperv6/peer.go
diff options
context:
space:
mode:
Diffstat (limited to 'whisper/whisperv6/peer.go')
-rw-r--r--whisper/whisperv6/peer.go83
1 files changed, 42 insertions, 41 deletions
diff --git a/whisper/whisperv6/peer.go b/whisper/whisperv6/peer.go
index 08071c0f7..4f9a7c378 100644
--- a/whisper/whisperv6/peer.go
+++ b/whisper/whisperv6/peer.go
@@ -28,7 +28,7 @@ import (
set "gopkg.in/fatih/set.v0"
)
-// peer represents a whisper protocol peer connection.
+// Peer represents a whisper protocol peer connection.
type Peer struct {
host *Whisper
peer *p2p.Peer
@@ -58,48 +58,48 @@ func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer {
// start initiates the peer updater, periodically broadcasting the whisper packets
// into the network.
-func (p *Peer) start() {
- go p.update()
- log.Trace("start", "peer", p.ID())
+func (peer *Peer) start() {
+ go peer.update()
+ log.Trace("start", "peer", peer.ID())
}
// stop terminates the peer updater, stopping message forwarding to it.
-func (p *Peer) stop() {
- close(p.quit)
- log.Trace("stop", "peer", p.ID())
+func (peer *Peer) stop() {
+ close(peer.quit)
+ log.Trace("stop", "peer", peer.ID())
}
// handshake sends the protocol initiation status message to the remote peer and
// verifies the remote status too.
-func (p *Peer) handshake() error {
+func (peer *Peer) handshake() error {
// Send the handshake status message asynchronously
errc := make(chan error, 1)
go func() {
- pow := p.host.MinPow()
+ pow := peer.host.MinPow()
powConverted := math.Float64bits(pow)
- bloom := p.host.BloomFilter()
- errc <- p2p.SendItems(p.ws, statusCode, ProtocolVersion, powConverted, bloom)
+ bloom := peer.host.BloomFilter()
+ errc <- p2p.SendItems(peer.ws, statusCode, ProtocolVersion, powConverted, bloom)
}()
// Fetch the remote status packet and verify protocol match
- packet, err := p.ws.ReadMsg()
+ packet, err := peer.ws.ReadMsg()
if err != nil {
return err
}
if packet.Code != statusCode {
- return fmt.Errorf("peer [%x] sent packet %x before status packet", p.ID(), packet.Code)
+ return fmt.Errorf("peer [%x] sent packet %x before status packet", peer.ID(), packet.Code)
}
s := rlp.NewStream(packet.Payload, uint64(packet.Size))
_, err = s.List()
if err != nil {
- return fmt.Errorf("peer [%x] sent bad status message: %v", p.ID(), err)
+ return fmt.Errorf("peer [%x] sent bad status message: %v", peer.ID(), err)
}
peerVersion, err := s.Uint()
if err != nil {
- return fmt.Errorf("peer [%x] sent bad status message (unable to decode version): %v", p.ID(), err)
+ return fmt.Errorf("peer [%x] sent bad status message (unable to decode version): %v", peer.ID(), err)
}
if peerVersion != ProtocolVersion {
- return fmt.Errorf("peer [%x]: protocol version mismatch %d != %d", p.ID(), peerVersion, ProtocolVersion)
+ return fmt.Errorf("peer [%x]: protocol version mismatch %d != %d", peer.ID(), peerVersion, ProtocolVersion)
}
// only version is mandatory, subsequent parameters are optional
@@ -107,34 +107,34 @@ func (p *Peer) handshake() error {
if err == nil {
pow := math.Float64frombits(powRaw)
if math.IsInf(pow, 0) || math.IsNaN(pow) || pow < 0.0 {
- return fmt.Errorf("peer [%x] sent bad status message: invalid pow", p.ID())
+ return fmt.Errorf("peer [%x] sent bad status message: invalid pow", peer.ID())
}
- p.powRequirement = pow
+ peer.powRequirement = pow
var bloom []byte
err = s.Decode(&bloom)
if err == nil {
sz := len(bloom)
if sz != bloomFilterSize && sz != 0 {
- return fmt.Errorf("peer [%x] sent bad status message: wrong bloom filter size %d", p.ID(), sz)
+ return fmt.Errorf("peer [%x] sent bad status message: wrong bloom filter size %d", peer.ID(), sz)
}
if isFullNode(bloom) {
- p.bloomFilter = nil
+ peer.bloomFilter = nil
} else {
- p.bloomFilter = bloom
+ peer.bloomFilter = bloom
}
}
}
if err := <-errc; err != nil {
- return fmt.Errorf("peer [%x] failed to send status packet: %v", p.ID(), err)
+ return fmt.Errorf("peer [%x] failed to send status packet: %v", peer.ID(), err)
}
return nil
}
// update executes periodic operations on the peer, including message transmission
// and expiration.
-func (p *Peer) update() {
+func (peer *Peer) update() {
// Start the tickers for the updates
expire := time.NewTicker(expirationCycle)
transmit := time.NewTicker(transmissionCycle)
@@ -143,15 +143,15 @@ func (p *Peer) update() {
for {
select {
case <-expire.C:
- p.expire()
+ peer.expire()
case <-transmit.C:
- if err := p.broadcast(); err != nil {
- log.Trace("broadcast failed", "reason", err, "peer", p.ID())
+ if err := peer.broadcast(); err != nil {
+ log.Trace("broadcast failed", "reason", err, "peer", peer.ID())
return
}
- case <-p.quit:
+ case <-peer.quit:
return
}
}
@@ -185,24 +185,24 @@ func (peer *Peer) expire() {
// broadcast iterates over the collection of envelopes and transmits yet unknown
// ones over the network.
-func (p *Peer) broadcast() error {
- envelopes := p.host.Envelopes()
+func (peer *Peer) broadcast() error {
+ envelopes := peer.host.Envelopes()
bundle := make([]*Envelope, 0, len(envelopes))
for _, envelope := range envelopes {
- if !p.marked(envelope) && envelope.PoW() >= p.powRequirement && p.bloomMatch(envelope) {
+ if !peer.marked(envelope) && envelope.PoW() >= peer.powRequirement && peer.bloomMatch(envelope) {
bundle = append(bundle, envelope)
}
}
if len(bundle) > 0 {
// transmit the batch of envelopes
- if err := p2p.Send(p.ws, messagesCode, bundle); err != nil {
+ if err := p2p.Send(peer.ws, messagesCode, bundle); err != nil {
return err
}
// mark envelopes only if they were successfully sent
for _, e := range bundle {
- p.mark(e)
+ peer.mark(e)
}
log.Trace("broadcast", "num. messages", len(bundle))
@@ -210,25 +210,26 @@ func (p *Peer) broadcast() error {
return nil
}
-func (p *Peer) ID() []byte {
- id := p.peer.ID()
+// ID returns a peer's id
+func (peer *Peer) ID() []byte {
+ id := peer.peer.ID()
return id[:]
}
-func (p *Peer) notifyAboutPowRequirementChange(pow float64) error {
+func (peer *Peer) notifyAboutPowRequirementChange(pow float64) error {
i := math.Float64bits(pow)
- return p2p.Send(p.ws, powRequirementCode, i)
+ return p2p.Send(peer.ws, powRequirementCode, i)
}
-func (p *Peer) notifyAboutBloomFilterChange(bloom []byte) error {
- return p2p.Send(p.ws, bloomFilterExCode, bloom)
+func (peer *Peer) notifyAboutBloomFilterChange(bloom []byte) error {
+ return p2p.Send(peer.ws, bloomFilterExCode, bloom)
}
-func (p *Peer) bloomMatch(env *Envelope) bool {
- if p.bloomFilter == nil {
+func (peer *Peer) bloomMatch(env *Envelope) bool {
+ if peer.bloomFilter == nil {
// no filter - full node, accepts all envelops
return true
}
- return bloomFilterMatch(p.bloomFilter, env.Bloom())
+ return bloomFilterMatch(peer.bloomFilter, env.Bloom())
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-01-03 03:13:58 +0800