mirror of https://github.com/poanetwork/quorum.git
183 lines
5.1 KiB
Go
183 lines
5.1 KiB
Go
// Copyright 2016 The go-ethereum Authors
|
|
// This file is part of the go-ethereum library.
|
|
//
|
|
// The go-ethereum library is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU Lesser General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// The go-ethereum library is distributed in the hope that it will be useful,
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
// GNU Lesser General Public License for more details.
|
|
//
|
|
// You should have received a copy of the GNU Lesser General Public License
|
|
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
package whisperv5
|
|
|
|
import (
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/common"
|
|
"github.com/ethereum/go-ethereum/logger"
|
|
"github.com/ethereum/go-ethereum/logger/glog"
|
|
"github.com/ethereum/go-ethereum/p2p"
|
|
"github.com/ethereum/go-ethereum/rlp"
|
|
set "gopkg.in/fatih/set.v0"
|
|
)
|
|
|
|
// peer represents a whisper protocol peer connection.
|
|
type Peer struct {
|
|
host *Whisper
|
|
peer *p2p.Peer
|
|
ws p2p.MsgReadWriter
|
|
trusted bool
|
|
|
|
known *set.Set // Messages already known by the peer to avoid wasting bandwidth
|
|
|
|
quit chan struct{}
|
|
}
|
|
|
|
// newPeer creates a new whisper peer object, but does not run the handshake itself.
|
|
func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer {
|
|
return &Peer{
|
|
host: host,
|
|
peer: remote,
|
|
ws: rw,
|
|
trusted: false,
|
|
known: set.New(),
|
|
quit: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
// start initiates the peer updater, periodically broadcasting the whisper packets
|
|
// into the network.
|
|
func (p *Peer) start() {
|
|
go p.update()
|
|
glog.V(logger.Debug).Infof("%v: whisper started", p.peer)
|
|
}
|
|
|
|
// stop terminates the peer updater, stopping message forwarding to it.
|
|
func (p *Peer) stop() {
|
|
close(p.quit)
|
|
glog.V(logger.Debug).Infof("%v: whisper stopped", p.peer)
|
|
}
|
|
|
|
// handshake sends the protocol initiation status message to the remote peer and
|
|
// verifies the remote status too.
|
|
func (p *Peer) handshake() error {
|
|
// Send the handshake status message asynchronously
|
|
errc := make(chan error, 1)
|
|
go func() {
|
|
errc <- p2p.Send(p.ws, statusCode, ProtocolVersion)
|
|
}()
|
|
// Fetch the remote status packet and verify protocol match
|
|
packet, err := p.ws.ReadMsg()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if packet.Code != statusCode {
|
|
return fmt.Errorf("peer sent %x before status packet", packet.Code)
|
|
}
|
|
s := rlp.NewStream(packet.Payload, uint64(packet.Size))
|
|
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)
|
|
}
|
|
// Wait until out own status is consumed too
|
|
if err := <-errc; err != nil {
|
|
return fmt.Errorf("failed to send status packet: %v", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// update executes periodic operations on the peer, including message transmission
|
|
// and expiration.
|
|
func (p *Peer) update() {
|
|
// Start the tickers for the updates
|
|
expire := time.NewTicker(expirationCycle)
|
|
transmit := time.NewTicker(transmissionCycle)
|
|
|
|
// Loop and transmit until termination is requested
|
|
for {
|
|
select {
|
|
case <-expire.C:
|
|
p.expire()
|
|
|
|
case <-transmit.C:
|
|
if err := p.broadcast(); err != nil {
|
|
glog.V(logger.Info).Infof("%v: broadcast failed: %v", p.peer, err)
|
|
return
|
|
}
|
|
|
|
case <-p.quit:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// mark marks an envelope known to the peer so that it won't be sent back.
|
|
func (peer *Peer) mark(envelope *Envelope) {
|
|
peer.known.Add(envelope.Hash())
|
|
}
|
|
|
|
// marked checks if an envelope is already known to the remote peer.
|
|
func (peer *Peer) marked(envelope *Envelope) bool {
|
|
return peer.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 (peer *Peer) expire() {
|
|
// Assemble the list of available envelopes
|
|
available := set.NewNonTS()
|
|
for _, envelope := range peer.host.Envelopes() {
|
|
available.Add(envelope.Hash())
|
|
}
|
|
// Cross reference availability with known status
|
|
unmark := make(map[common.Hash]struct{})
|
|
peer.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 {
|
|
peer.known.Remove(hash)
|
|
}
|
|
}
|
|
|
|
// broadcast iterates over the collection of envelopes and transmits yet unknown
|
|
// ones over the network.
|
|
func (p *Peer) broadcast() error {
|
|
// Fetch the envelopes and collect the unknown ones
|
|
envelopes := p.host.Envelopes()
|
|
transmit := make([]*Envelope, 0, len(envelopes))
|
|
for _, envelope := range envelopes {
|
|
if !p.marked(envelope) {
|
|
transmit = append(transmit, envelope)
|
|
p.mark(envelope)
|
|
}
|
|
}
|
|
if len(transmit) == 0 {
|
|
return nil
|
|
}
|
|
// Transmit the unknown batch (potentially empty)
|
|
if err := p2p.Send(p.ws, messagesCode, transmit); err != nil {
|
|
return err
|
|
}
|
|
glog.V(logger.Detail).Infoln(p.peer, "broadcasted", len(transmit), "message(s)")
|
|
return nil
|
|
}
|
|
|
|
func (p *Peer) ID() []byte {
|
|
id := p.peer.ID()
|
|
return id[:]
|
|
}
|