complete handshake with peers (#3446)

fix

node-test/src/test/scala/org/bitcoins/node/NeutrinoNodeTest.scala

@@ -31,13 +31,22 @@ class NeutrinoNodeTest extends NodeTestWithCachedBitcoindPair {
 
   behavior of "NeutrinoNode"
 
-  it must "be able to connect and then disconnect from all peers" in {
+  it must "be able to connect,initialize and then disconnect from all peers" in {
     nodeConnectedWithBitcoind: NeutrinoNodeConnectedWithBitcoinds =>
       //checking all peers are connected
       val node = nodeConnectedWithBitcoind.node
       val connFs = node.peers.indices.map(node.isConnected)
       val connF = Future.sequence(connFs).map(_.forall(_ == true))
-      val assertion1F = connF.map(assert(_))
+      val connAssertion = connF.map(assert(_))
+
+      //checking all peers are initialized
+      val isInitializedFs = node.peers.indices.map(node.isInitialized)
+      val isInitializedF = for {
+        _ <- connAssertion
+        f <- Future.sequence(isInitializedFs).map(_.forall(_ == true))
+      } yield f
+
+      val initAssertion = isInitializedF.map(assert(_))
 
       //checking all peers can be disconnected
       def isAllDisconnectedF: Future[Boolean] = {
@@ -46,7 +55,7 @@ class NeutrinoNodeTest extends NodeTestWithCachedBitcoindPair {
         res
       }
       val disconnF = for {
-        _ <- assertion1F
+        _ <- initAssertion
         _ <- node.stop()
         f <- isAllDisconnectedF
       } yield f

node/src/main/scala/org/bitcoins/node/Node.scala

@@ -126,18 +126,22 @@ trait Node extends NodeApi with ChainQueryApi with P2PLogger {
     val chainApiF = startConfsF.flatMap(_ => chainApiFromDb())
 
     val startNodeF = {
-      peerMsgSenders.foreach(_.connect())
-      val isInitializedF = for {
-        _ <- AsyncUtil.retryUntilSatisfiedF(() => isInitialized(0),
-                                            maxTries = 200,
-                                            interval = 250.millis)
-      } yield ()
+      val isInitializedFs = peerMsgSenders.indices.map { idx =>
+        peerMsgSenders(idx).connect()
+        val isInitializedF = for {
+          _ <- AsyncUtil.retryUntilSatisfiedF(() => isInitialized(idx),
+                                              maxTries = 200,
+                                              interval = 250.millis)
+        } yield ()
+        isInitializedF.failed.foreach(err =>
+          logger.error(
+            s"Failed to connect with peer=${peers(idx)} with err=$err"))
+        isInitializedF.map { _ =>
+          logger.info(s"Our peer=${peers(idx)} has been initialized")
+        }
+      }
 
-      isInitializedF.failed.foreach(err =>
-        logger.error(s"Failed to connect with peer=${peers(0)} with err=$err"))
-
-      isInitializedF.map { _ =>
-        logger.info(s"Our peer=${peers(0)} has been initialized")
+      Future.sequence(isInitializedFs).map { _ =>
         logger.info(s"Our node has been full started. It took=${System
           .currentTimeMillis() - start}ms")
         this