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.

649 lines
19 KiB

9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
8 years ago
9 years ago
8 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
8 years ago
9 years ago
7 years ago
9 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
9 years ago
8 years ago
9 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
9 years ago
8 years ago
7 years ago
7 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
9 years ago
8 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
8 years ago
9 years ago
7 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
7 years ago
8 years ago
9 years ago
7 years ago
9 years ago
8 years ago
9 years ago
9 years ago
9 years ago
9 years ago
7 years ago
7 years ago
7 years ago
7 years ago
8 years ago
7 years ago
8 years ago
8 years ago
8 years ago
  1. package p2p
  2. import (
  3. "fmt"
  4. "math"
  5. "math/rand"
  6. "net"
  7. "time"
  8. "github.com/pkg/errors"
  9. crypto "github.com/tendermint/go-crypto"
  10. cfg "github.com/tendermint/tendermint/config"
  11. cmn "github.com/tendermint/tmlibs/common"
  12. )
  13. const (
  14. // wait a random amount of time from this interval
  15. // before dialing seeds or reconnecting to help prevent DoS
  16. dialRandomizerIntervalMilliseconds = 3000
  17. // repeatedly try to reconnect for a few minutes
  18. // ie. 5 * 20 = 100s
  19. reconnectAttempts = 20
  20. reconnectInterval = 5 * time.Second
  21. // then move into exponential backoff mode for ~1day
  22. // ie. 3**10 = 16hrs
  23. reconnectBackOffAttempts = 10
  24. reconnectBackOffBaseSeconds = 3
  25. )
  26. type Reactor interface {
  27. cmn.Service // Start, Stop
  28. SetSwitch(*Switch)
  29. GetChannels() []*ChannelDescriptor
  30. AddPeer(peer Peer)
  31. RemovePeer(peer Peer, reason interface{})
  32. Receive(chID byte, peer Peer, msgBytes []byte) // CONTRACT: msgBytes are not nil
  33. }
  34. //--------------------------------------
  35. type BaseReactor struct {
  36. cmn.BaseService // Provides Start, Stop, .Quit
  37. Switch *Switch
  38. }
  39. func NewBaseReactor(name string, impl Reactor) *BaseReactor {
  40. return &BaseReactor{
  41. BaseService: *cmn.NewBaseService(nil, name, impl),
  42. Switch: nil,
  43. }
  44. }
  45. func (br *BaseReactor) SetSwitch(sw *Switch) {
  46. br.Switch = sw
  47. }
  48. func (_ *BaseReactor) GetChannels() []*ChannelDescriptor { return nil }
  49. func (_ *BaseReactor) AddPeer(peer Peer) {}
  50. func (_ *BaseReactor) RemovePeer(peer Peer, reason interface{}) {}
  51. func (_ *BaseReactor) Receive(chID byte, peer Peer, msgBytes []byte) {}
  52. //-----------------------------------------------------------------------------
  53. /*
  54. The `Switch` handles peer connections and exposes an API to receive incoming messages
  55. on `Reactors`. Each `Reactor` is responsible for handling incoming messages of one
  56. or more `Channels`. So while sending outgoing messages is typically performed on the peer,
  57. incoming messages are received on the reactor.
  58. */
  59. type Switch struct {
  60. cmn.BaseService
  61. config *cfg.P2PConfig
  62. peerConfig *PeerConfig
  63. listeners []Listener
  64. reactors map[string]Reactor
  65. chDescs []*ChannelDescriptor
  66. reactorsByCh map[byte]Reactor
  67. peers *PeerSet
  68. dialing *cmn.CMap
  69. nodeInfo *NodeInfo // our node info
  70. nodePrivKey crypto.PrivKeyEd25519 // our node privkey
  71. filterConnByAddr func(net.Addr) error
  72. filterConnByPubKey func(crypto.PubKeyEd25519) error
  73. rng *rand.Rand // seed for randomizing dial times and orders
  74. }
  75. var (
  76. ErrSwitchDuplicatePeer = errors.New("Duplicate peer")
  77. )
  78. func NewSwitch(config *cfg.P2PConfig) *Switch {
  79. sw := &Switch{
  80. config: config,
  81. peerConfig: DefaultPeerConfig(),
  82. reactors: make(map[string]Reactor),
  83. chDescs: make([]*ChannelDescriptor, 0),
  84. reactorsByCh: make(map[byte]Reactor),
  85. peers: NewPeerSet(),
  86. dialing: cmn.NewCMap(),
  87. nodeInfo: nil,
  88. }
  89. // Ensure we have a completely undeterministic PRNG. cmd.RandInt64() draws
  90. // from a seed that's initialized with OS entropy on process start.
  91. sw.rng = rand.New(rand.NewSource(cmn.RandInt64()))
  92. // TODO: collapse the peerConfig into the config ?
  93. sw.peerConfig.MConfig.flushThrottle = time.Duration(config.FlushThrottleTimeout) * time.Millisecond
  94. sw.peerConfig.MConfig.SendRate = config.SendRate
  95. sw.peerConfig.MConfig.RecvRate = config.RecvRate
  96. sw.peerConfig.MConfig.maxMsgPacketPayloadSize = config.MaxMsgPacketPayloadSize
  97. sw.BaseService = *cmn.NewBaseService(nil, "P2P Switch", sw)
  98. return sw
  99. }
  100. // AddReactor adds the given reactor to the switch.
  101. // NOTE: Not goroutine safe.
  102. func (sw *Switch) AddReactor(name string, reactor Reactor) Reactor {
  103. // Validate the reactor.
  104. // No two reactors can share the same channel.
  105. reactorChannels := reactor.GetChannels()
  106. for _, chDesc := range reactorChannels {
  107. chID := chDesc.ID
  108. if sw.reactorsByCh[chID] != nil {
  109. cmn.PanicSanity(fmt.Sprintf("Channel %X has multiple reactors %v & %v", chID, sw.reactorsByCh[chID], reactor))
  110. }
  111. sw.chDescs = append(sw.chDescs, chDesc)
  112. sw.reactorsByCh[chID] = reactor
  113. }
  114. sw.reactors[name] = reactor
  115. reactor.SetSwitch(sw)
  116. return reactor
  117. }
  118. // Reactors returns a map of reactors registered on the switch.
  119. // NOTE: Not goroutine safe.
  120. func (sw *Switch) Reactors() map[string]Reactor {
  121. return sw.reactors
  122. }
  123. // Reactor returns the reactor with the given name.
  124. // NOTE: Not goroutine safe.
  125. func (sw *Switch) Reactor(name string) Reactor {
  126. return sw.reactors[name]
  127. }
  128. // AddListener adds the given listener to the switch for listening to incoming peer connections.
  129. // NOTE: Not goroutine safe.
  130. func (sw *Switch) AddListener(l Listener) {
  131. sw.listeners = append(sw.listeners, l)
  132. }
  133. // Listeners returns the list of listeners the switch listens on.
  134. // NOTE: Not goroutine safe.
  135. func (sw *Switch) Listeners() []Listener {
  136. return sw.listeners
  137. }
  138. // IsListening returns true if the switch has at least one listener.
  139. // NOTE: Not goroutine safe.
  140. func (sw *Switch) IsListening() bool {
  141. return len(sw.listeners) > 0
  142. }
  143. // SetNodeInfo sets the switch's NodeInfo for checking compatibility and handshaking with other nodes.
  144. // NOTE: Not goroutine safe.
  145. func (sw *Switch) SetNodeInfo(nodeInfo *NodeInfo) {
  146. sw.nodeInfo = nodeInfo
  147. }
  148. // NodeInfo returns the switch's NodeInfo.
  149. // NOTE: Not goroutine safe.
  150. func (sw *Switch) NodeInfo() *NodeInfo {
  151. return sw.nodeInfo
  152. }
  153. // SetNodePrivKey sets the switch's private key for authenticated encryption.
  154. // NOTE: Overwrites sw.nodeInfo.PubKey.
  155. // NOTE: Not goroutine safe.
  156. func (sw *Switch) SetNodePrivKey(nodePrivKey crypto.PrivKeyEd25519) {
  157. sw.nodePrivKey = nodePrivKey
  158. if sw.nodeInfo != nil {
  159. sw.nodeInfo.PubKey = nodePrivKey.PubKey().Unwrap().(crypto.PubKeyEd25519)
  160. }
  161. }
  162. // OnStart implements BaseService. It starts all the reactors, peers, and listeners.
  163. func (sw *Switch) OnStart() error {
  164. // Start reactors
  165. for _, reactor := range sw.reactors {
  166. err := reactor.Start()
  167. if err != nil {
  168. return errors.Wrapf(err, "failed to start %v", reactor)
  169. }
  170. }
  171. // Start listeners
  172. for _, listener := range sw.listeners {
  173. go sw.listenerRoutine(listener)
  174. }
  175. return nil
  176. }
  177. // OnStop implements BaseService. It stops all listeners, peers, and reactors.
  178. func (sw *Switch) OnStop() {
  179. // Stop listeners
  180. for _, listener := range sw.listeners {
  181. listener.Stop()
  182. }
  183. sw.listeners = nil
  184. // Stop peers
  185. for _, peer := range sw.peers.List() {
  186. peer.Stop()
  187. sw.peers.Remove(peer)
  188. }
  189. // Stop reactors
  190. sw.Logger.Debug("Switch: Stopping reactors")
  191. for _, reactor := range sw.reactors {
  192. reactor.Stop()
  193. }
  194. }
  195. // addPeer checks the given peer's validity, performs a handshake, and adds the
  196. // peer to the switch and to all registered reactors.
  197. // NOTE: This performs a blocking handshake before the peer is added.
  198. // NOTE: If error is returned, caller is responsible for calling peer.CloseConn()
  199. func (sw *Switch) addPeer(peer *peer) error {
  200. if err := sw.FilterConnByAddr(peer.Addr()); err != nil {
  201. return err
  202. }
  203. if err := sw.FilterConnByPubKey(peer.PubKey()); err != nil {
  204. return err
  205. }
  206. if err := peer.HandshakeTimeout(sw.nodeInfo, time.Duration(sw.peerConfig.HandshakeTimeout*time.Second)); err != nil {
  207. return err
  208. }
  209. // Avoid self
  210. if sw.nodeInfo.PubKey.Equals(peer.PubKey().Wrap()) {
  211. return errors.New("Ignoring connection from self")
  212. }
  213. // Check version, chain id
  214. if err := sw.nodeInfo.CompatibleWith(peer.NodeInfo()); err != nil {
  215. return err
  216. }
  217. // Check for duplicate peer
  218. if sw.peers.Has(peer.Key()) {
  219. return ErrSwitchDuplicatePeer
  220. }
  221. // Start peer
  222. if sw.IsRunning() {
  223. sw.startInitPeer(peer)
  224. }
  225. // Add the peer to .peers.
  226. // We start it first so that a peer in the list is safe to Stop.
  227. // It should not err since we already checked peers.Has().
  228. if err := sw.peers.Add(peer); err != nil {
  229. return err
  230. }
  231. sw.Logger.Info("Added peer", "peer", peer)
  232. return nil
  233. }
  234. // FilterConnByAddr returns an error if connecting to the given address is forbidden.
  235. func (sw *Switch) FilterConnByAddr(addr net.Addr) error {
  236. if sw.filterConnByAddr != nil {
  237. return sw.filterConnByAddr(addr)
  238. }
  239. return nil
  240. }
  241. // FilterConnByPubKey returns an error if connecting to the given public key is forbidden.
  242. func (sw *Switch) FilterConnByPubKey(pubkey crypto.PubKeyEd25519) error {
  243. if sw.filterConnByPubKey != nil {
  244. return sw.filterConnByPubKey(pubkey)
  245. }
  246. return nil
  247. }
  248. // SetAddrFilter sets the function for filtering connections by address.
  249. func (sw *Switch) SetAddrFilter(f func(net.Addr) error) {
  250. sw.filterConnByAddr = f
  251. }
  252. // SetPubKeyFilter sets the function for filtering connections by public key.
  253. func (sw *Switch) SetPubKeyFilter(f func(crypto.PubKeyEd25519) error) {
  254. sw.filterConnByPubKey = f
  255. }
  256. func (sw *Switch) startInitPeer(peer *peer) {
  257. err := peer.Start() // spawn send/recv routines
  258. if err != nil {
  259. // Should never happen
  260. sw.Logger.Error("Error starting peer", "peer", peer, "err", err)
  261. }
  262. for _, reactor := range sw.reactors {
  263. reactor.AddPeer(peer)
  264. }
  265. }
  266. // DialSeeds dials a list of seeds asynchronously in random order.
  267. func (sw *Switch) DialSeeds(addrBook *AddrBook, seeds []string) error {
  268. netAddrs, errs := NewNetAddressStrings(seeds)
  269. for _, err := range errs {
  270. sw.Logger.Error("Error in seed's address", "err", err)
  271. }
  272. if addrBook != nil {
  273. // add seeds to `addrBook`
  274. ourAddrS := sw.nodeInfo.ListenAddr
  275. ourAddr, _ := NewNetAddressString(ourAddrS)
  276. for _, netAddr := range netAddrs {
  277. // do not add ourselves
  278. if netAddr.Equals(ourAddr) {
  279. continue
  280. }
  281. addrBook.AddAddress(netAddr, ourAddr)
  282. }
  283. addrBook.Save()
  284. }
  285. // permute the list, dial them in random order.
  286. perm := sw.rng.Perm(len(netAddrs))
  287. for i := 0; i < len(perm); i++ {
  288. go func(i int) {
  289. sw.randomSleep(0)
  290. j := perm[i]
  291. sw.dialSeed(netAddrs[j])
  292. }(i)
  293. }
  294. return nil
  295. }
  296. // sleep for interval plus some random amount of ms on [0, dialRandomizerIntervalMilliseconds]
  297. func (sw *Switch) randomSleep(interval time.Duration) {
  298. r := time.Duration(sw.rng.Int63n(dialRandomizerIntervalMilliseconds)) * time.Millisecond
  299. time.Sleep(r + interval)
  300. }
  301. func (sw *Switch) dialSeed(addr *NetAddress) {
  302. peer, err := sw.DialPeerWithAddress(addr, true)
  303. if err != nil {
  304. sw.Logger.Error("Error dialing seed", "err", err)
  305. } else {
  306. sw.Logger.Info("Connected to seed", "peer", peer)
  307. }
  308. }
  309. // DialPeerWithAddress dials the given peer and runs sw.addPeer if it connects successfully.
  310. // If `persistent == true`, the switch will always try to reconnect to this peer if the connection ever fails.
  311. func (sw *Switch) DialPeerWithAddress(addr *NetAddress, persistent bool) (Peer, error) {
  312. sw.dialing.Set(addr.IP.String(), addr)
  313. defer sw.dialing.Delete(addr.IP.String())
  314. sw.Logger.Info("Dialing peer", "address", addr)
  315. peer, err := newOutboundPeer(addr, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig)
  316. if err != nil {
  317. sw.Logger.Error("Failed to dial peer", "address", addr, "err", err)
  318. return nil, err
  319. }
  320. peer.SetLogger(sw.Logger.With("peer", addr))
  321. if persistent {
  322. peer.makePersistent()
  323. }
  324. err = sw.addPeer(peer)
  325. if err != nil {
  326. sw.Logger.Error("Failed to add peer", "address", addr, "err", err)
  327. peer.CloseConn()
  328. return nil, err
  329. }
  330. sw.Logger.Info("Dialed and added peer", "address", addr, "peer", peer)
  331. return peer, nil
  332. }
  333. // IsDialing returns true if the switch is currently dialing the given address.
  334. func (sw *Switch) IsDialing(addr *NetAddress) bool {
  335. return sw.dialing.Has(addr.IP.String())
  336. }
  337. // Broadcast runs a go routine for each attempted send, which will block
  338. // trying to send for defaultSendTimeoutSeconds. Returns a channel
  339. // which receives success values for each attempted send (false if times out).
  340. // NOTE: Broadcast uses goroutines, so order of broadcast may not be preserved.
  341. // TODO: Something more intelligent.
  342. func (sw *Switch) Broadcast(chID byte, msg interface{}) chan bool {
  343. successChan := make(chan bool, len(sw.peers.List()))
  344. sw.Logger.Debug("Broadcast", "channel", chID, "msg", msg)
  345. for _, peer := range sw.peers.List() {
  346. go func(peer Peer) {
  347. success := peer.Send(chID, msg)
  348. successChan <- success
  349. }(peer)
  350. }
  351. return successChan
  352. }
  353. // NumPeers returns the count of outbound/inbound and outbound-dialing peers.
  354. func (sw *Switch) NumPeers() (outbound, inbound, dialing int) {
  355. peers := sw.peers.List()
  356. for _, peer := range peers {
  357. if peer.IsOutbound() {
  358. outbound++
  359. } else {
  360. inbound++
  361. }
  362. }
  363. dialing = sw.dialing.Size()
  364. return
  365. }
  366. // Peers returns the set of peers that are connected to the switch.
  367. func (sw *Switch) Peers() IPeerSet {
  368. return sw.peers
  369. }
  370. // StopPeerForError disconnects from a peer due to external error.
  371. // If the peer is persistent, it will attempt to reconnect.
  372. // TODO: make record depending on reason.
  373. func (sw *Switch) StopPeerForError(peer Peer, reason interface{}) {
  374. sw.Logger.Error("Stopping peer for error", "peer", peer, "err", reason)
  375. sw.stopAndRemovePeer(peer, reason)
  376. if peer.IsPersistent() {
  377. go sw.reconnectToPeer(peer)
  378. }
  379. }
  380. // reconnectToPeer tries to reconnect to the peer, first repeatedly
  381. // with a fixed interval, then with exponential backoff.
  382. // If no success after all that, it stops trying, and leaves it
  383. // to the PEX/Addrbook to find the peer again
  384. func (sw *Switch) reconnectToPeer(peer Peer) {
  385. addr, _ := NewNetAddressString(peer.NodeInfo().RemoteAddr)
  386. start := time.Now()
  387. sw.Logger.Info("Reconnecting to peer", "peer", peer)
  388. for i := 0; i < reconnectAttempts; i++ {
  389. if !sw.IsRunning() {
  390. return
  391. }
  392. peer, err := sw.DialPeerWithAddress(addr, true)
  393. if err != nil {
  394. sw.Logger.Info("Error reconnecting to peer. Trying again", "tries", i, "err", err, "peer", peer)
  395. // sleep a set amount
  396. sw.randomSleep(reconnectInterval)
  397. continue
  398. } else {
  399. sw.Logger.Info("Reconnected to peer", "peer", peer)
  400. return
  401. }
  402. }
  403. sw.Logger.Error("Failed to reconnect to peer. Beginning exponential backoff",
  404. "peer", peer, "elapsed", time.Since(start))
  405. for i := 0; i < reconnectBackOffAttempts; i++ {
  406. if !sw.IsRunning() {
  407. return
  408. }
  409. // sleep an exponentially increasing amount
  410. sleepIntervalSeconds := math.Pow(reconnectBackOffBaseSeconds, float64(i))
  411. sw.randomSleep(time.Duration(sleepIntervalSeconds) * time.Second)
  412. peer, err := sw.DialPeerWithAddress(addr, true)
  413. if err != nil {
  414. sw.Logger.Info("Error reconnecting to peer. Trying again", "tries", i, "err", err, "peer", peer)
  415. continue
  416. } else {
  417. sw.Logger.Info("Reconnected to peer", "peer", peer)
  418. return
  419. }
  420. }
  421. sw.Logger.Error("Failed to reconnect to peer. Giving up", "peer", peer, "elapsed", time.Since(start))
  422. }
  423. // StopPeerGracefully disconnects from a peer gracefully.
  424. // TODO: handle graceful disconnects.
  425. func (sw *Switch) StopPeerGracefully(peer Peer) {
  426. sw.Logger.Info("Stopping peer gracefully")
  427. sw.stopAndRemovePeer(peer, nil)
  428. }
  429. func (sw *Switch) stopAndRemovePeer(peer Peer, reason interface{}) {
  430. sw.peers.Remove(peer)
  431. peer.Stop()
  432. for _, reactor := range sw.reactors {
  433. reactor.RemovePeer(peer, reason)
  434. }
  435. }
  436. func (sw *Switch) listenerRoutine(l Listener) {
  437. for {
  438. inConn, ok := <-l.Connections()
  439. if !ok {
  440. break
  441. }
  442. // ignore connection if we already have enough
  443. maxPeers := sw.config.MaxNumPeers
  444. if maxPeers <= sw.peers.Size() {
  445. sw.Logger.Info("Ignoring inbound connection: already have enough peers", "address", inConn.RemoteAddr().String(), "numPeers", sw.peers.Size(), "max", maxPeers)
  446. continue
  447. }
  448. // New inbound connection!
  449. err := sw.addPeerWithConnectionAndConfig(inConn, sw.peerConfig)
  450. if err != nil {
  451. sw.Logger.Info("Ignoring inbound connection: error while adding peer", "address", inConn.RemoteAddr().String(), "err", err)
  452. continue
  453. }
  454. // NOTE: We don't yet have the listening port of the
  455. // remote (if they have a listener at all).
  456. // The peerHandshake will handle that.
  457. }
  458. // cleanup
  459. }
  460. //------------------------------------------------------------------
  461. // Connects switches via arbitrary net.Conn. Used for testing.
  462. // MakeConnectedSwitches returns n switches, connected according to the connect func.
  463. // If connect==Connect2Switches, the switches will be fully connected.
  464. // initSwitch defines how the i'th switch should be initialized (ie. with what reactors).
  465. // NOTE: panics if any switch fails to start.
  466. func MakeConnectedSwitches(cfg *cfg.P2PConfig, n int, initSwitch func(int, *Switch) *Switch, connect func([]*Switch, int, int)) []*Switch {
  467. switches := make([]*Switch, n)
  468. for i := 0; i < n; i++ {
  469. switches[i] = makeSwitch(cfg, i, "testing", "123.123.123", initSwitch)
  470. }
  471. if err := StartSwitches(switches); err != nil {
  472. panic(err)
  473. }
  474. for i := 0; i < n; i++ {
  475. for j := i + 1; j < n; j++ {
  476. connect(switches, i, j)
  477. }
  478. }
  479. return switches
  480. }
  481. // Connect2Switches will connect switches i and j via net.Pipe().
  482. // Blocks until a connection is established.
  483. // NOTE: caller ensures i and j are within bounds.
  484. func Connect2Switches(switches []*Switch, i, j int) {
  485. switchI := switches[i]
  486. switchJ := switches[j]
  487. c1, c2 := netPipe()
  488. doneCh := make(chan struct{})
  489. go func() {
  490. err := switchI.addPeerWithConnection(c1)
  491. if err != nil {
  492. panic(err)
  493. }
  494. doneCh <- struct{}{}
  495. }()
  496. go func() {
  497. err := switchJ.addPeerWithConnection(c2)
  498. if err != nil {
  499. panic(err)
  500. }
  501. doneCh <- struct{}{}
  502. }()
  503. <-doneCh
  504. <-doneCh
  505. }
  506. // StartSwitches calls sw.Start() for each given switch.
  507. // It returns the first encountered error.
  508. func StartSwitches(switches []*Switch) error {
  509. for _, s := range switches {
  510. err := s.Start() // start switch and reactors
  511. if err != nil {
  512. return err
  513. }
  514. }
  515. return nil
  516. }
  517. func makeSwitch(cfg *cfg.P2PConfig, i int, network, version string, initSwitch func(int, *Switch) *Switch) *Switch {
  518. privKey := crypto.GenPrivKeyEd25519()
  519. // new switch, add reactors
  520. // TODO: let the config be passed in?
  521. s := initSwitch(i, NewSwitch(cfg))
  522. s.SetNodeInfo(&NodeInfo{
  523. PubKey: privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
  524. Moniker: cmn.Fmt("switch%d", i),
  525. Network: network,
  526. Version: version,
  527. RemoteAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
  528. ListenAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
  529. })
  530. s.SetNodePrivKey(privKey)
  531. return s
  532. }
  533. func (sw *Switch) addPeerWithConnection(conn net.Conn) error {
  534. peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig)
  535. if err != nil {
  536. if err := conn.Close(); err != nil {
  537. sw.Logger.Error("Error closing connection", "err", err)
  538. }
  539. return err
  540. }
  541. peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
  542. if err = sw.addPeer(peer); err != nil {
  543. peer.CloseConn()
  544. return err
  545. }
  546. return nil
  547. }
  548. func (sw *Switch) addPeerWithConnectionAndConfig(conn net.Conn, config *PeerConfig) error {
  549. peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, config)
  550. if err != nil {
  551. if err := conn.Close(); err != nil {
  552. sw.Logger.Error("Error closing connection", "err", err)
  553. }
  554. return err
  555. }
  556. peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
  557. if err = sw.addPeer(peer); err != nil {
  558. peer.CloseConn()
  559. return err
  560. }
  561. return nil
  562. }