1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
package whisper
import (
"fmt"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rlp"
"gopkg.in/fatih/set.v0"
)
// peer represents a whisper protocol peer connection.
type peer struct {
host *Whisper
peer *p2p.Peer
ws p2p.MsgReadWriter
known *set.Set // Messages already known by the peer to avoid wasting bandwidth
quit chan struct{}
}
// newPeer creates and initializes a new whisper peer connection, returning either
// the newly constructed link or a failure reason.
func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) (*peer, error) {
p := &peer{
host: host,
peer: remote,
ws: rw,
known: set.New(),
quit: make(chan struct{}),
}
if err := p.handshake(); err != nil {
return nil, err
}
return p, nil
}
// start initiates the peer updater, periodically broadcasting the whisper packets
// into the network.
func (self *peer) start() {
go self.update()
self.peer.Debugln("whisper started")
}
// stop terminates the peer updater, stopping message forwarding to it.
func (self *peer) stop() {
close(self.quit)
self.peer.Debugln("whisper stopped")
}
// handshake sends the protocol initiation status message to the remote peer and
// verifies the remote status too.
func (self *peer) handshake() error {
// Send own status message, fetch remote one
if err := p2p.SendItems(self.ws, statusCode, protocolVersion); err != nil {
return err
}
packet, err := self.ws.ReadMsg()
if err != nil {
return err
}
if packet.Code != statusCode {
return fmt.Errorf("peer sent %x before status packet", packet.Code)
}
// Decode the rest of the status packet and verify protocol match
s := rlp.NewStream(packet.Payload)
if _, err := s.List(); err != nil {
return fmt.Errorf("bad status message: %v", err)
}
peerVersion, err := s.Uint()
if err != nil {
return fmt.Errorf("bad status message: %v", err)
}
if peerVersion != protocolVersion {
return fmt.Errorf("protocol version mismatch %d != %d", peerVersion, protocolVersion)
}
return packet.Discard() // ignore anything after protocol version
}
// update executes periodic operations on the peer, including message transmission
// and expiration.
func (self *peer) update() {
// Start the tickers for the updates
expire := time.NewTicker(expirationTicks)
transmit := time.NewTicker(transmissionTicks)
// Loop and transmit until termination is requested
for {
select {
case <-expire.C:
self.expire()
case <-transmit.C:
if err := self.broadcast(); err != nil {
self.peer.Infoln("broadcast failed:", err)
return
}
case <-self.quit:
return
}
}
}
// mark marks an envelope known to the peer so that it won't be sent back.
func (self *peer) mark(envelope *Envelope) {
self.known.Add(envelope.Hash())
}
// marked checks if an envelope is already known to the remote peer.
func (self *peer) marked(envelope *Envelope) bool {
return self.known.Has(envelope.Hash())
}
// expire iterates over all the known envelopes in the host and removes all
// expired (unknown) ones from the known list.
func (self *peer) expire() {
// Assemble the list of available envelopes
available := set.NewNonTS()
for _, envelope := range self.host.envelopes() {
available.Add(envelope.Hash())
}
// Cross reference availability with known status
unmark := make(map[common.Hash]struct{})
self.known.Each(func(v interface{}) bool {
if !available.Has(v.(common.Hash)) {
unmark[v.(common.Hash)] = struct{}{}
}
return true
})
// Dump all known but unavailable
for hash, _ := range unmark {
self.known.Remove(hash)
}
}
// broadcast iterates over the collection of envelopes and transmits yet unknown
// ones over the network.
func (self *peer) broadcast() error {
// Fetch the envelopes and collect the unknown ones
envelopes := self.host.envelopes()
transmit := make([]*Envelope, 0, len(envelopes))
for _, envelope := range envelopes {
if !self.marked(envelope) {
transmit = append(transmit, envelope)
self.mark(envelope)
}
}
// Transmit the unknown batch (potentially empty)
if err := p2p.Send(self.ws, messagesCode, transmit); err != nil {
return err
}
self.peer.DebugDetailln("broadcasted", len(transmit), "message(s)")
return nil
}
|
