You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
137 lines
2.8 KiB
137 lines
2.8 KiB
package p2p
|
|
|
|
import (
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
|
|
. "github.com/tendermint/go-common"
|
|
cfg "github.com/tendermint/go-config"
|
|
"github.com/tendermint/go-wire"
|
|
)
|
|
|
|
type Peer struct {
|
|
BaseService
|
|
|
|
outbound bool
|
|
mconn *MConnection
|
|
|
|
*NodeInfo
|
|
Key string
|
|
Data *CMap // User data.
|
|
}
|
|
|
|
// NOTE: blocking
|
|
// Before creating a peer with newPeer(), perform a handshake on connection.
|
|
func peerHandshake(conn net.Conn, ourNodeInfo *NodeInfo) (*NodeInfo, error) {
|
|
var peerNodeInfo = new(NodeInfo)
|
|
var err1 error
|
|
var err2 error
|
|
Parallel(
|
|
func() {
|
|
var n int
|
|
wire.WriteBinary(ourNodeInfo, conn, &n, &err1)
|
|
},
|
|
func() {
|
|
var n int
|
|
wire.ReadBinary(peerNodeInfo, conn, maxNodeInfoSize, &n, &err2)
|
|
log.Notice("Peer handshake", "peerNodeInfo", peerNodeInfo)
|
|
})
|
|
if err1 != nil {
|
|
return nil, err1
|
|
}
|
|
if err2 != nil {
|
|
return nil, err2
|
|
}
|
|
peerNodeInfo.RemoteAddr = conn.RemoteAddr().String()
|
|
return peerNodeInfo, nil
|
|
}
|
|
|
|
// NOTE: call peerHandshake on conn before calling newPeer().
|
|
func newPeer(config cfg.Config, conn net.Conn, peerNodeInfo *NodeInfo, outbound bool, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{})) *Peer {
|
|
var p *Peer
|
|
onReceive := func(chID byte, msgBytes []byte) {
|
|
reactor := reactorsByCh[chID]
|
|
if reactor == nil {
|
|
PanicSanity(Fmt("Unknown channel %X", chID))
|
|
}
|
|
reactor.Receive(chID, p, msgBytes)
|
|
}
|
|
onError := func(r interface{}) {
|
|
p.Stop()
|
|
onPeerError(p, r)
|
|
}
|
|
mconn := NewMConnection(config, conn, chDescs, onReceive, onError)
|
|
p = &Peer{
|
|
outbound: outbound,
|
|
mconn: mconn,
|
|
NodeInfo: peerNodeInfo,
|
|
Key: peerNodeInfo.PubKey.KeyString(),
|
|
Data: NewCMap(),
|
|
}
|
|
p.BaseService = *NewBaseService(log, "Peer", p)
|
|
return p
|
|
}
|
|
|
|
func (p *Peer) OnStart() error {
|
|
p.BaseService.OnStart()
|
|
_, err := p.mconn.Start()
|
|
return err
|
|
}
|
|
|
|
func (p *Peer) OnStop() {
|
|
p.BaseService.OnStop()
|
|
p.mconn.Stop()
|
|
}
|
|
|
|
func (p *Peer) Connection() *MConnection {
|
|
return p.mconn
|
|
}
|
|
|
|
func (p *Peer) IsOutbound() bool {
|
|
return p.outbound
|
|
}
|
|
|
|
func (p *Peer) Send(chID byte, msg interface{}) bool {
|
|
if !p.IsRunning() {
|
|
return false
|
|
}
|
|
return p.mconn.Send(chID, msg)
|
|
}
|
|
|
|
func (p *Peer) TrySend(chID byte, msg interface{}) bool {
|
|
if !p.IsRunning() {
|
|
return false
|
|
}
|
|
return p.mconn.TrySend(chID, msg)
|
|
}
|
|
|
|
func (p *Peer) CanSend(chID byte) bool {
|
|
if !p.IsRunning() {
|
|
return false
|
|
}
|
|
return p.mconn.CanSend(chID)
|
|
}
|
|
|
|
func (p *Peer) WriteTo(w io.Writer) (n int64, err error) {
|
|
var n_ int
|
|
wire.WriteString(p.Key, w, &n_, &err)
|
|
n += int64(n_)
|
|
return
|
|
}
|
|
|
|
func (p *Peer) String() string {
|
|
if p.outbound {
|
|
return fmt.Sprintf("Peer{%v %v out}", p.mconn, p.Key[:12])
|
|
} else {
|
|
return fmt.Sprintf("Peer{%v %v in}", p.mconn, p.Key[:12])
|
|
}
|
|
}
|
|
|
|
func (p *Peer) Equals(other *Peer) bool {
|
|
return p.Key == other.Key
|
|
}
|
|
|
|
func (p *Peer) Get(key string) interface{} {
|
|
return p.Data.Get(key)
|
|
}
|