9 "github.com/pkg/errors"
10 log "github.com/sirupsen/logrus"
11 crypto "github.com/tendermint/go-crypto"
12 wire "github.com/tendermint/go-wire"
13 cmn "github.com/tendermint/tmlibs/common"
15 cfg "github.com/bytom/config"
18 // Peer could be marked as persistent, in which case you can use
19 // Redial function to reconnect. Note that inbound peers can't be
20 // made persistent. They should be made persistent on the other end.
22 // Before using a peer, you will need to perform a handshake on connection.
28 conn net.Conn // source connection
29 mconn *MConnection // multiplex connection
36 Data *cmn.CMap // User data.
39 // PeerConfig is a Peer configuration.
40 type PeerConfig struct {
41 AuthEnc bool `mapstructure:"auth_enc"` // authenticated encryption
43 // times are in seconds
44 HandshakeTimeout time.Duration `mapstructure:"handshake_timeout"`
45 DialTimeout time.Duration `mapstructure:"dial_timeout"`
47 MConfig *MConnConfig `mapstructure:"connection"`
49 Fuzz bool `mapstructure:"fuzz"` // fuzz connection (for testing)
50 FuzzConfig *FuzzConnConfig `mapstructure:"fuzz_config"`
53 // DefaultPeerConfig returns the default config.
54 func DefaultPeerConfig(config *cfg.P2PConfig) *PeerConfig {
57 HandshakeTimeout: time.Duration(config.HandshakeTimeout), // * time.Second,
58 DialTimeout: time.Duration(config.DialTimeout), // * time.Second,
59 MConfig: DefaultMConnConfig(),
61 FuzzConfig: DefaultFuzzConnConfig(),
65 func newOutboundPeer(addr *NetAddress, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{}), ourNodePrivKey crypto.PrivKeyEd25519, config *cfg.P2PConfig) (*Peer, error) {
66 return newOutboundPeerWithConfig(addr, reactorsByCh, chDescs, onPeerError, ourNodePrivKey, DefaultPeerConfig(config))
69 func newOutboundPeerWithConfig(addr *NetAddress, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{}), ourNodePrivKey crypto.PrivKeyEd25519, config *PeerConfig) (*Peer, error) {
70 conn, err := dial(addr, config)
72 return nil, errors.Wrap(err, "Error creating peer")
75 peer, err := newPeerFromConnAndConfig(conn, true, reactorsByCh, chDescs, onPeerError, ourNodePrivKey, config)
83 func newInboundPeer(conn net.Conn, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{}), ourNodePrivKey crypto.PrivKeyEd25519, config *cfg.P2PConfig) (*Peer, error) {
84 return newInboundPeerWithConfig(conn, reactorsByCh, chDescs, onPeerError, ourNodePrivKey, DefaultPeerConfig(config))
87 func newInboundPeerWithConfig(conn net.Conn, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{}), ourNodePrivKey crypto.PrivKeyEd25519, config *PeerConfig) (*Peer, error) {
88 return newPeerFromConnAndConfig(conn, false, reactorsByCh, chDescs, onPeerError, ourNodePrivKey, config)
91 func newPeerFromConnAndConfig(rawConn net.Conn, outbound bool, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{}), ourNodePrivKey crypto.PrivKeyEd25519, config *PeerConfig) (*Peer, error) {
96 // so we have time to do peer handshakes and get set up
97 conn = FuzzConnAfterFromConfig(conn, 10*time.Second, config.FuzzConfig)
100 // Encrypt connection
102 conn.SetDeadline(time.Now().Add(config.HandshakeTimeout * time.Second))
105 conn, err = MakeSecretConnection(conn, ourNodePrivKey)
107 return nil, errors.Wrap(err, "Error creating peer")
111 // Key and NodeInfo are set after Handshake
119 p.mconn = createMConnection(conn, p, reactorsByCh, chDescs, onPeerError, config.MConfig)
121 p.BaseService = *cmn.NewBaseService(nil, "Peer", p)
126 // CloseConn should be used when the peer was created, but never started.
127 func (p *Peer) CloseConn() {
131 // makePersistent marks the peer as persistent.
132 func (p *Peer) makePersistent() {
134 panic("inbound peers can't be made persistent")
140 // IsPersistent returns true if the peer is persitent, false otherwise.
141 func (p *Peer) IsPersistent() bool {
145 // HandshakeTimeout performs a handshake between a given node and the peer.
147 func (p *Peer) HandshakeTimeout(ourNodeInfo *NodeInfo, timeout time.Duration) error {
148 // Set deadline for handshake so we don't block forever on conn.ReadFull
149 p.conn.SetDeadline(time.Now().Add(timeout))
151 var peerNodeInfo = new(NodeInfo)
157 wire.WriteBinary(ourNodeInfo, p.conn, &n, &err1)
161 wire.ReadBinary(peerNodeInfo, p.conn, maxNodeInfoSize, &n, &err2)
162 log.WithField("peerNodeInfo", peerNodeInfo).Info("Peer handshake")
165 return errors.Wrap(err1, "Error during handshake/write")
168 return errors.Wrap(err2, "Error during handshake/read")
171 if p.config.AuthEnc {
172 // Check that the professed PubKey matches the sconn's.
173 if !peerNodeInfo.PubKey.Equals(p.PubKey().Wrap()) {
174 return fmt.Errorf("Ignoring connection with unmatching pubkey: %v vs %v",
175 peerNodeInfo.PubKey, p.PubKey())
180 p.conn.SetDeadline(time.Time{})
182 peerNodeInfo.RemoteAddr = p.Addr().String()
184 p.NodeInfo = peerNodeInfo
185 p.Key = peerNodeInfo.PubKey.KeyString()
190 // Addr returns peer's remote network address.
191 func (p *Peer) Addr() net.Addr {
192 return p.conn.RemoteAddr()
195 // PubKey returns peer's public key.
196 func (p *Peer) PubKey() crypto.PubKeyEd25519 {
197 if p.config.AuthEnc {
198 return p.conn.(*SecretConnection).RemotePubKey()
200 if p.NodeInfo == nil {
201 panic("Attempt to get peer's PubKey before calling Handshake")
206 // OnStart implements BaseService.
207 func (p *Peer) OnStart() error {
208 p.BaseService.OnStart()
209 _, err := p.mconn.Start()
213 // OnStop implements BaseService.
214 func (p *Peer) OnStop() {
215 p.BaseService.OnStop()
219 // Connection returns underlying MConnection.
220 func (p *Peer) Connection() *MConnection {
224 // IsOutbound returns true if the connection is outbound, false otherwise.
225 func (p *Peer) IsOutbound() bool {
229 // Send msg to the channel identified by chID byte. Returns false if the send
230 // queue is full after timeout, specified by MConnection.
231 func (p *Peer) Send(chID byte, msg interface{}) bool {
233 // see Switch#Broadcast, where we fetch the list of peers and loop over
234 // them - while we're looping, one peer may be removed and stopped.
237 return p.mconn.Send(chID, msg)
240 // TrySend msg to the channel identified by chID byte. Immediately returns
241 // false if the send queue is full.
242 func (p *Peer) TrySend(chID byte, msg interface{}) bool {
246 return p.mconn.TrySend(chID, msg)
249 // CanSend returns true if the send queue is not full, false otherwise.
250 func (p *Peer) CanSend(chID byte) bool {
254 return p.mconn.CanSend(chID)
257 // WriteTo writes the peer's public key to w.
258 func (p *Peer) WriteTo(w io.Writer) (n int64, err error) {
260 wire.WriteString(p.Key, w, &n_, &err)
265 // String representation.
266 func (p *Peer) String() string {
268 return fmt.Sprintf("Peer{%v %v out}", p.mconn, p.Key[:12])
271 return fmt.Sprintf("Peer{%v %v in}", p.mconn, p.Key[:12])
274 // Equals reports whenever 2 peers are actually represent the same node.
275 func (p *Peer) Equals(other *Peer) bool {
276 return p.Key == other.Key
279 // Get the data for a given key.
280 func (p *Peer) Get(key string) interface{} {
281 return p.Data.Get(key)
284 func dial(addr *NetAddress, config *PeerConfig) (net.Conn, error) {
285 conn, err := addr.DialTimeout(config.DialTimeout * time.Second)
292 func createMConnection(conn net.Conn, p *Peer, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{}), config *MConnConfig) *MConnection {
293 onReceive := func(chID byte, msgBytes []byte) {
294 reactor := reactorsByCh[chID]
296 if chID == PexChannel {
299 cmn.PanicSanity(cmn.Fmt("Unknown channel %X", chID))
302 reactor.Receive(chID, p, msgBytes)
305 onError := func(r interface{}) {
309 return NewMConnectionWithConfig(conn, chDescs, onReceive, onError, config)