Connectivity Improvements

node/src/main/scala/com/wavesplatform/network/BasicMessagesRepo.scala

@@ -1,12 +1,9 @@
 package com.wavesplatform.network
 
-import java.net.{InetAddress, InetSocketAddress}
-import java.util
-import scala.util.Try
 import com.google.common.primitives.{Bytes, Ints}
 import com.wavesplatform.account.PublicKey
-import com.wavesplatform.block.{Block, MicroBlock}
 import com.wavesplatform.block.serialization.MicroBlockSerializer
+import com.wavesplatform.block.{Block, MicroBlock}
 import com.wavesplatform.common.state.ByteStr
 import com.wavesplatform.crypto
 import com.wavesplatform.crypto.*
@@ -19,6 +16,11 @@ import com.wavesplatform.protobuf.snapshot.{BlockSnapshot as PBBlockSnapshot, Mi
 import com.wavesplatform.protobuf.transaction.{PBSignedTransaction, PBTransactions}
 import com.wavesplatform.transaction.{DataTransaction, EthereumTransaction, Transaction, TransactionParsers}
 
+import java.net.{InetAddress, InetSocketAddress}
+import java.util
+import scala.reflect.ClassTag
+import scala.util.Try
+
 object GetPeersSpec extends MessageSpec[GetPeers.type] {
   override val messageCode: Message.MessageCode = 1: Byte
 
@@ -33,7 +35,7 @@ object GetPeersSpec extends MessageSpec[GetPeers.type] {
   override def serializeData(data: GetPeers.type): Array[Byte] = Array()
 }
 
-object PeersSpec extends MessageSpec[KnownPeers] {
+abstract class InetSocketAddressSeqSpec[A <: AnyRef : ClassTag] extends MessageSpec[A] {
   private val AddressLength = 4
   private val PortLength    = 4
   private val DataLength    = 4
@@ -42,36 +44,46 @@ object PeersSpec extends MessageSpec[KnownPeers] {
 
   override val maxLength: Int = DataLength + 1000 * (AddressLength + PortLength)
 
-  override def deserializeData(bytes: Array[Byte]): Try[KnownPeers] = Try {
+  protected def unwrap(v: A): Seq[InetSocketAddress]
+  protected def wrap(addresses: Seq[InetSocketAddress]): A
+
+  override def deserializeData(bytes: Array[Byte]): Try[A] = Try {
     val lengthBytes = util.Arrays.copyOfRange(bytes, 0, DataLength)
     val length      = Ints.fromByteArray(lengthBytes)
 
-    assert(bytes.length == DataLength + (length * (AddressLength + PortLength)), "Data does not match length")
+    val expectedMessageLength = DataLength + (length * (AddressLength + PortLength))
+    assert(
+      bytes.length == expectedMessageLength,
+      s"Invalid KnownPeers message length ${bytes.length}, expecting $expectedMessageLength for $length peers"
+    )
 
-    KnownPeers((0 until length).map { i =>
+    (0 until length).map { i =>
       val position     = lengthBytes.length + (i * (AddressLength + PortLength))
       val addressBytes = util.Arrays.copyOfRange(bytes, position, position + AddressLength)
       val address      = InetAddress.getByAddress(addressBytes)
       val portBytes    = util.Arrays.copyOfRange(bytes, position + AddressLength, position + AddressLength + PortLength)
       new InetSocketAddress(address, Ints.fromByteArray(portBytes))
-    })
-  }
-
-  override def serializeData(peers: KnownPeers): Array[Byte] = {
-    val length      = peers.peers.size
-    val lengthBytes = Ints.toByteArray(length)
+    }
+  }.map(wrap)
 
-    val xs = for {
-      inetAddress <- peers.peers
+  def serializeData(value: A): Array[Byte] = {
+    val peersWithResolvedAddresses = for {
+      inetAddress <- unwrap(value)
       address     <- Option(inetAddress.getAddress)
     } yield (address.getAddress, inetAddress.getPort)
 
-    xs.foldLeft(lengthBytes) { case (bs, (peerAddress, peerPort)) =>
+    peersWithResolvedAddresses.foldLeft(Ints.toByteArray(peersWithResolvedAddresses.size)) { case (bs, (peerAddress, peerPort)) =>
       Bytes.concat(bs, peerAddress, Ints.toByteArray(peerPort))
     }
   }
 }
 
+object PeersSpec extends InetSocketAddressSeqSpec[KnownPeers] {
+  override protected def unwrap(v: KnownPeers): Seq[InetSocketAddress] = v.peers
+
+  override protected def wrap(addresses: Seq[InetSocketAddress]): KnownPeers = KnownPeers(addresses)
+}
+
 trait SignaturesSeqSpec[A <: AnyRef] extends MessageSpec[A] {
 
   private val DataLength = 4
@@ -355,7 +367,7 @@ object HandshakeSpec {
 object BasicMessagesRepo {
   type Spec = MessageSpec[? <: AnyRef]
 
-  val specs: Seq[Spec] = Seq(
+  private val specs: Seq[Spec] = Seq(
     GetPeersSpec,
     PeersSpec,
     GetSignaturesSpec,

node/src/main/scala/com/wavesplatform/network/NetworkServer.scala

@@ -153,7 +153,7 @@ object NetworkServer extends ScorexLogging {
         peerDatabase.touch(remoteAddress)
         thisConnFuture.channel().closeFuture().addListener(f => handleOutgoingChannelClosed(remoteAddress)(f))
       } else if (thisConnFuture.cause() != null) {
-        peerDatabase.suspendAndClose(thisConnFuture.channel())
+        peerDatabase.suspend(remoteAddress)
         outgoingChannels.remove(remoteAddress, thisConnFuture.channel())
         thisConnFuture.cause() match {
           case e: ClosedChannelException =>