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.

648 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
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. for _, reactor := range sw.reactors {
  191. reactor.Stop()
  192. }
  193. }
  194. // addPeer checks the given peer's validity, performs a handshake, and adds the
  195. // peer to the switch and to all registered reactors.
  196. // NOTE: This performs a blocking handshake before the peer is added.
  197. // NOTE: If error is returned, caller is responsible for calling peer.CloseConn()
  198. func (sw *Switch) addPeer(peer *peer) error {
  199. if err := sw.FilterConnByAddr(peer.Addr()); err != nil {
  200. return err
  201. }
  202. if err := sw.FilterConnByPubKey(peer.PubKey()); err != nil {
  203. return err
  204. }
  205. if err := peer.HandshakeTimeout(sw.nodeInfo, time.Duration(sw.peerConfig.HandshakeTimeout*time.Second)); err != nil {
  206. return err
  207. }
  208. // Avoid self
  209. if sw.nodeInfo.PubKey.Equals(peer.PubKey().Wrap()) {
  210. return errors.New("Ignoring connection from self")
  211. }
  212. // Check version, chain id
  213. if err := sw.nodeInfo.CompatibleWith(peer.NodeInfo()); err != nil {
  214. return err
  215. }
  216. // Check for duplicate peer
  217. if sw.peers.Has(peer.Key()) {
  218. return ErrSwitchDuplicatePeer
  219. }
  220. // Start peer
  221. if sw.IsRunning() {
  222. sw.startInitPeer(peer)
  223. }
  224. // Add the peer to .peers.
  225. // We start it first so that a peer in the list is safe to Stop.
  226. // It should not err since we already checked peers.Has().
  227. if err := sw.peers.Add(peer); err != nil {
  228. return err
  229. }
  230. sw.Logger.Info("Added peer", "peer", peer)
  231. return nil
  232. }
  233. // FilterConnByAddr returns an error if connecting to the given address is forbidden.
  234. func (sw *Switch) FilterConnByAddr(addr net.Addr) error {
  235. if sw.filterConnByAddr != nil {
  236. return sw.filterConnByAddr(addr)
  237. }
  238. return nil
  239. }
  240. // FilterConnByPubKey returns an error if connecting to the given public key is forbidden.
  241. func (sw *Switch) FilterConnByPubKey(pubkey crypto.PubKeyEd25519) error {
  242. if sw.filterConnByPubKey != nil {
  243. return sw.filterConnByPubKey(pubkey)
  244. }
  245. return nil
  246. }
  247. // SetAddrFilter sets the function for filtering connections by address.
  248. func (sw *Switch) SetAddrFilter(f func(net.Addr) error) {
  249. sw.filterConnByAddr = f
  250. }
  251. // SetPubKeyFilter sets the function for filtering connections by public key.
  252. func (sw *Switch) SetPubKeyFilter(f func(crypto.PubKeyEd25519) error) {
  253. sw.filterConnByPubKey = f
  254. }
  255. func (sw *Switch) startInitPeer(peer *peer) {
  256. err := peer.Start() // spawn send/recv routines
  257. if err != nil {
  258. // Should never happen
  259. sw.Logger.Error("Error starting peer", "peer", peer, "err", err)
  260. }
  261. for _, reactor := range sw.reactors {
  262. reactor.AddPeer(peer)
  263. }
  264. }
  265. // DialSeeds dials a list of seeds asynchronously in random order.
  266. func (sw *Switch) DialSeeds(addrBook *AddrBook, seeds []string) error {
  267. netAddrs, errs := NewNetAddressStrings(seeds)
  268. for _, err := range errs {
  269. sw.Logger.Error("Error in seed's address", "err", err)
  270. }
  271. if addrBook != nil {
  272. // add seeds to `addrBook`
  273. ourAddrS := sw.nodeInfo.ListenAddr
  274. ourAddr, _ := NewNetAddressString(ourAddrS)
  275. for _, netAddr := range netAddrs {
  276. // do not add ourselves
  277. if netAddr.Equals(ourAddr) {
  278. continue
  279. }
  280. addrBook.AddAddress(netAddr, ourAddr)
  281. }
  282. addrBook.Save()
  283. }
  284. // permute the list, dial them in random order.
  285. perm := sw.rng.Perm(len(netAddrs))
  286. for i := 0; i < len(perm); i++ {
  287. go func(i int) {
  288. sw.randomSleep(0)
  289. j := perm[i]
  290. sw.dialSeed(netAddrs[j])
  291. }(i)
  292. }
  293. return nil
  294. }
  295. // sleep for interval plus some random amount of ms on [0, dialRandomizerIntervalMilliseconds]
  296. func (sw *Switch) randomSleep(interval time.Duration) {
  297. r := time.Duration(sw.rng.Int63n(dialRandomizerIntervalMilliseconds)) * time.Millisecond
  298. time.Sleep(r + interval)
  299. }
  300. func (sw *Switch) dialSeed(addr *NetAddress) {
  301. peer, err := sw.DialPeerWithAddress(addr, true)
  302. if err != nil {
  303. sw.Logger.Error("Error dialing seed", "err", err)
  304. } else {
  305. sw.Logger.Info("Connected to seed", "peer", peer)
  306. }
  307. }
  308. // DialPeerWithAddress dials the given peer and runs sw.addPeer if it connects successfully.
  309. // If `persistent == true`, the switch will always try to reconnect to this peer if the connection ever fails.
  310. func (sw *Switch) DialPeerWithAddress(addr *NetAddress, persistent bool) (Peer, error) {
  311. sw.dialing.Set(addr.IP.String(), addr)
  312. defer sw.dialing.Delete(addr.IP.String())
  313. sw.Logger.Info("Dialing peer", "address", addr)
  314. peer, err := newOutboundPeer(addr, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig)
  315. if err != nil {
  316. sw.Logger.Error("Failed to dial peer", "address", addr, "err", err)
  317. return nil, err
  318. }
  319. peer.SetLogger(sw.Logger.With("peer", addr))
  320. if persistent {
  321. peer.makePersistent()
  322. }
  323. err = sw.addPeer(peer)
  324. if err != nil {
  325. sw.Logger.Error("Failed to add peer", "address", addr, "err", err)
  326. peer.CloseConn()
  327. return nil, err
  328. }
  329. sw.Logger.Info("Dialed and added peer", "address", addr, "peer", peer)
  330. return peer, nil
  331. }
  332. // IsDialing returns true if the switch is currently dialing the given address.
  333. func (sw *Switch) IsDialing(addr *NetAddress) bool {
  334. return sw.dialing.Has(addr.IP.String())
  335. }
  336. // Broadcast runs a go routine for each attempted send, which will block
  337. // trying to send for defaultSendTimeoutSeconds. Returns a channel
  338. // which receives success values for each attempted send (false if times out).
  339. // NOTE: Broadcast uses goroutines, so order of broadcast may not be preserved.
  340. // TODO: Something more intelligent.
  341. func (sw *Switch) Broadcast(chID byte, msg interface{}) chan bool {
  342. successChan := make(chan bool, len(sw.peers.List()))
  343. sw.Logger.Debug("Broadcast", "channel", chID, "msg", msg)
  344. for _, peer := range sw.peers.List() {
  345. go func(peer Peer) {
  346. success := peer.Send(chID, msg)
  347. successChan <- success
  348. }(peer)
  349. }
  350. return successChan
  351. }
  352. // NumPeers returns the count of outbound/inbound and outbound-dialing peers.
  353. func (sw *Switch) NumPeers() (outbound, inbound, dialing int) {
  354. peers := sw.peers.List()
  355. for _, peer := range peers {
  356. if peer.IsOutbound() {
  357. outbound++
  358. } else {
  359. inbound++
  360. }
  361. }
  362. dialing = sw.dialing.Size()
  363. return
  364. }
  365. // Peers returns the set of peers that are connected to the switch.
  366. func (sw *Switch) Peers() IPeerSet {
  367. return sw.peers
  368. }
  369. // StopPeerForError disconnects from a peer due to external error.
  370. // If the peer is persistent, it will attempt to reconnect.
  371. // TODO: make record depending on reason.
  372. func (sw *Switch) StopPeerForError(peer Peer, reason interface{}) {
  373. sw.Logger.Error("Stopping peer for error", "peer", peer, "err", reason)
  374. sw.stopAndRemovePeer(peer, reason)
  375. if peer.IsPersistent() {
  376. go sw.reconnectToPeer(peer)
  377. }
  378. }
  379. // reconnectToPeer tries to reconnect to the peer, first repeatedly
  380. // with a fixed interval, then with exponential backoff.
  381. // If no success after all that, it stops trying, and leaves it
  382. // to the PEX/Addrbook to find the peer again
  383. func (sw *Switch) reconnectToPeer(peer Peer) {
  384. addr, _ := NewNetAddressString(peer.NodeInfo().RemoteAddr)
  385. start := time.Now()
  386. sw.Logger.Info("Reconnecting to peer", "peer", peer)
  387. for i := 0; i < reconnectAttempts; i++ {
  388. if !sw.IsRunning() {
  389. return
  390. }
  391. peer, err := sw.DialPeerWithAddress(addr, true)
  392. if err != nil {
  393. sw.Logger.Info("Error reconnecting to peer. Trying again", "tries", i, "err", err, "peer", peer)
  394. // sleep a set amount
  395. sw.randomSleep(reconnectInterval)
  396. continue
  397. } else {
  398. sw.Logger.Info("Reconnected to peer", "peer", peer)
  399. return
  400. }
  401. }
  402. sw.Logger.Error("Failed to reconnect to peer. Beginning exponential backoff",
  403. "peer", peer, "elapsed", time.Since(start))
  404. for i := 0; i < reconnectBackOffAttempts; i++ {
  405. if !sw.IsRunning() {
  406. return
  407. }
  408. // sleep an exponentially increasing amount
  409. sleepIntervalSeconds := math.Pow(reconnectBackOffBaseSeconds, float64(i))
  410. sw.randomSleep(time.Duration(sleepIntervalSeconds) * time.Second)
  411. peer, err := sw.DialPeerWithAddress(addr, true)
  412. if err != nil {
  413. sw.Logger.Info("Error reconnecting to peer. Trying again", "tries", i, "err", err, "peer", peer)
  414. continue
  415. } else {
  416. sw.Logger.Info("Reconnected to peer", "peer", peer)
  417. return
  418. }
  419. }
  420. sw.Logger.Error("Failed to reconnect to peer. Giving up", "peer", peer, "elapsed", time.Since(start))
  421. }
  422. // StopPeerGracefully disconnects from a peer gracefully.
  423. // TODO: handle graceful disconnects.
  424. func (sw *Switch) StopPeerGracefully(peer Peer) {
  425. sw.Logger.Info("Stopping peer gracefully")
  426. sw.stopAndRemovePeer(peer, nil)
  427. }
  428. func (sw *Switch) stopAndRemovePeer(peer Peer, reason interface{}) {
  429. sw.peers.Remove(peer)
  430. peer.Stop()
  431. for _, reactor := range sw.reactors {
  432. reactor.RemovePeer(peer, reason)
  433. }
  434. }
  435. func (sw *Switch) listenerRoutine(l Listener) {
  436. for {
  437. inConn, ok := <-l.Connections()
  438. if !ok {
  439. break
  440. }
  441. // ignore connection if we already have enough
  442. maxPeers := sw.config.MaxNumPeers
  443. if maxPeers <= sw.peers.Size() {
  444. sw.Logger.Info("Ignoring inbound connection: already have enough peers", "address", inConn.RemoteAddr().String(), "numPeers", sw.peers.Size(), "max", maxPeers)
  445. continue
  446. }
  447. // New inbound connection!
  448. err := sw.addPeerWithConnectionAndConfig(inConn, sw.peerConfig)
  449. if err != nil {
  450. sw.Logger.Info("Ignoring inbound connection: error while adding peer", "address", inConn.RemoteAddr().String(), "err", err)
  451. continue
  452. }
  453. // NOTE: We don't yet have the listening port of the
  454. // remote (if they have a listener at all).
  455. // The peerHandshake will handle that.
  456. }
  457. // cleanup
  458. }
  459. //------------------------------------------------------------------
  460. // Connects switches via arbitrary net.Conn. Used for testing.
  461. // MakeConnectedSwitches returns n switches, connected according to the connect func.
  462. // If connect==Connect2Switches, the switches will be fully connected.
  463. // initSwitch defines how the i'th switch should be initialized (ie. with what reactors).
  464. // NOTE: panics if any switch fails to start.
  465. func MakeConnectedSwitches(cfg *cfg.P2PConfig, n int, initSwitch func(int, *Switch) *Switch, connect func([]*Switch, int, int)) []*Switch {
  466. switches := make([]*Switch, n)
  467. for i := 0; i < n; i++ {
  468. switches[i] = makeSwitch(cfg, i, "testing", "123.123.123", initSwitch)
  469. }
  470. if err := StartSwitches(switches); err != nil {
  471. panic(err)
  472. }
  473. for i := 0; i < n; i++ {
  474. for j := i + 1; j < n; j++ {
  475. connect(switches, i, j)
  476. }
  477. }
  478. return switches
  479. }
  480. // Connect2Switches will connect switches i and j via net.Pipe().
  481. // Blocks until a connection is established.
  482. // NOTE: caller ensures i and j are within bounds.
  483. func Connect2Switches(switches []*Switch, i, j int) {
  484. switchI := switches[i]
  485. switchJ := switches[j]
  486. c1, c2 := netPipe()
  487. doneCh := make(chan struct{})
  488. go func() {
  489. err := switchI.addPeerWithConnection(c1)
  490. if err != nil {
  491. panic(err)
  492. }
  493. doneCh <- struct{}{}
  494. }()
  495. go func() {
  496. err := switchJ.addPeerWithConnection(c2)
  497. if err != nil {
  498. panic(err)
  499. }
  500. doneCh <- struct{}{}
  501. }()
  502. <-doneCh
  503. <-doneCh
  504. }
  505. // StartSwitches calls sw.Start() for each given switch.
  506. // It returns the first encountered error.
  507. func StartSwitches(switches []*Switch) error {
  508. for _, s := range switches {
  509. err := s.Start() // start switch and reactors
  510. if err != nil {
  511. return err
  512. }
  513. }
  514. return nil
  515. }
  516. func makeSwitch(cfg *cfg.P2PConfig, i int, network, version string, initSwitch func(int, *Switch) *Switch) *Switch {
  517. privKey := crypto.GenPrivKeyEd25519()
  518. // new switch, add reactors
  519. // TODO: let the config be passed in?
  520. s := initSwitch(i, NewSwitch(cfg))
  521. s.SetNodeInfo(&NodeInfo{
  522. PubKey: privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
  523. Moniker: cmn.Fmt("switch%d", i),
  524. Network: network,
  525. Version: version,
  526. RemoteAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
  527. ListenAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
  528. })
  529. s.SetNodePrivKey(privKey)
  530. return s
  531. }
  532. func (sw *Switch) addPeerWithConnection(conn net.Conn) error {
  533. peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig)
  534. if err != nil {
  535. if err := conn.Close(); err != nil {
  536. sw.Logger.Error("Error closing connection", "err", err)
  537. }
  538. return err
  539. }
  540. peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
  541. if err = sw.addPeer(peer); err != nil {
  542. peer.CloseConn()
  543. return err
  544. }
  545. return nil
  546. }
  547. func (sw *Switch) addPeerWithConnectionAndConfig(conn net.Conn, config *PeerConfig) error {
  548. peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, config)
  549. if err != nil {
  550. if err := conn.Close(); err != nil {
  551. sw.Logger.Error("Error closing connection", "err", err)
  552. }
  553. return err
  554. }
  555. peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
  556. if err = sw.addPeer(peer); err != nil {
  557. peer.CloseConn()
  558. return err
  559. }
  560. return nil
  561. }