[ETCM-355] RLPxConnectionHandler.processMessage doesn't use Try

Author: lukasz-golebiewski

src/main/scala/io/iohk/ethereum/network/p2p/Message.scala

@@ -23,8 +23,7 @@ trait MessageSerializable extends Message {
 
 @FunctionalInterface
 trait MessageDecoder extends Logger { self =>
-
-  type DecodingError = Throwable // TODO: Replace Throwable with an ADT when feasible
+  import MessageDecoder._
 
   def fromBytes(`type`: Int, payload: Array[Byte]): Either[DecodingError, Message]
 
@@ -41,3 +40,7 @@ trait MessageDecoder extends Logger { self =>
       }
   }
 }
+
+object MessageDecoder {
+  type DecodingError = Throwable // TODO: Replace Throwable with an ADT when feasible
+}

src/main/scala/io/iohk/ethereum/network/p2p/MessageDecoders.scala

@@ -19,6 +19,7 @@ import io.iohk.ethereum.network.p2p.messages.WireProtocol.Ping._
 import io.iohk.ethereum.network.p2p.messages.WireProtocol.Pong._
 import io.iohk.ethereum.network.p2p.messages.WireProtocol._
 import scala.util.Try
+import MessageDecoder._
 
 object NetworkMessageDecoder extends MessageDecoder {
 

src/main/scala/io/iohk/ethereum/network/rlpx/MessageCodec.scala

@@ -13,6 +13,7 @@ import org.xerial.snappy.Snappy
 import io.iohk.ethereum.network.handshaker.EtcHelloExchangeState
 import io.iohk.ethereum.network.p2p.Message
 import io.iohk.ethereum.network.p2p.MessageDecoder
+import io.iohk.ethereum.network.p2p.MessageDecoder.DecodingError
 import io.iohk.ethereum.network.p2p.MessageSerializable
 import io.iohk.ethereum.network.p2p.messages.WireProtocol.Hello
 
@@ -32,12 +33,12 @@ class MessageCodec(
   val contextIdCounter = new AtomicInteger
 
   // TODO: ETCM-402 - messageDecoder should use negotiated protocol version
-  def readMessages(data: ByteString): Seq[Try[Message]] = {
+  def readMessages(data: ByteString): Seq[Either[DecodingError, Message]] = {
     val frames = frameCodec.readFrames(data)
     readFrames(frames)
   }
 
-  def readFrames(frames: Seq[Frame]): Seq[Try[Message]] =
+  def readFrames(frames: Seq[Frame]): Seq[Either[DecodingError, Message]] =
     frames.map { frame =>
       val frameData = frame.payload.toArray
       val payloadTry =
@@ -47,8 +48,8 @@ class MessageCodec(
           Success(frameData)
         }
 
-      payloadTry.map { payload =>
-        messageDecoder.fromBytesUnsafe(frame.`type`, payload)
+      payloadTry.toEither.flatMap { payload =>
+        messageDecoder.fromBytes(frame.`type`, payload)
       }
     }
 

src/main/scala/io/iohk/ethereum/network/rlpx/RLPxConnectionHandler.scala

@@ -19,6 +19,7 @@ import org.bouncycastle.util.encoders.Hex
 
 import io.iohk.ethereum.network.p2p.EthereumMessageDecoder
 import io.iohk.ethereum.network.p2p.Message
+import io.iohk.ethereum.network.p2p.MessageDecoder._
 import io.iohk.ethereum.network.p2p.MessageSerializable
 import io.iohk.ethereum.network.p2p.NetworkMessageDecoder
 import io.iohk.ethereum.network.p2p.messages.Capability
@@ -255,11 +256,11 @@ class RLPxConnectionHandler(
         messagesSoFar.foreach(processMessage)
       }
 
-    def processMessage(messageTry: Try[Message]): Unit = messageTry match {
-      case Success(message) =>
+    def processMessage(messageTry: Either[DecodingError, Message]): Unit = messageTry match {
+      case Right(message) =>
         context.parent ! MessageReceived(message)
 
-      case Failure(ex) =>
+      case Left(ex) =>
         log.info("Cannot decode message from {}, because of {}", peerId, ex.getMessage)
         // break connection in case of failed decoding, to avoid attack which would send us garbage
         context.stop(self)
@@ -450,8 +451,10 @@ object RLPxConnectionHandler {
     private def extractHello(frame: Frame): Option[Hello] = {
       val frameData = frame.payload.toArray
       if (frame.`type` == Hello.code) {
-        val m = NetworkMessageDecoder.fromBytesUnsafe(frame.`type`, frameData)
-        Some(m.asInstanceOf[Hello])
+        NetworkMessageDecoder.fromBytes(frame.`type`, frameData) match {
+          case Left(err) => throw err // TODO: rethink throwing here
+          case Right(msg) => Some(msg.asInstanceOf[Hello])
+        }
       } else {
         None
       }

src/test/scala/io/iohk/ethereum/network/p2p/MessageCodecSpec.scala

@@ -25,7 +25,7 @@ class MessageCodecSpec extends AnyFlatSpec with Matchers {
 
     // remote peer did not receive local status so it treats all remote messages as uncompressed
     assert(remoteReadNotCompressedStatus.size == 1)
-    assert(remoteReadNotCompressedStatus.head.get == status)
+    assert(remoteReadNotCompressedStatus.head == Right(status))
   }
 
   it should "compress messages when remote side advertises p2p version larger or equal 5" in new TestSetup {
@@ -41,7 +41,7 @@ class MessageCodecSpec extends AnyFlatSpec with Matchers {
     // remote peer did not receive local status so it treats all remote messages as uncompressed,
     // but local peer compress messages after V5 Hello message
     assert(remoteReadNotCompressedStatus.size == 1)
-    assert(remoteReadNotCompressedStatus.head.isFailure)
+    assert(remoteReadNotCompressedStatus.head.isLeft)
   }
 
   it should "compress messages when both sides advertises p2p version larger or equal 5" in new TestSetup {
@@ -56,7 +56,7 @@ class MessageCodecSpec extends AnyFlatSpec with Matchers {
 
     // both peers exchanged v5 hellos, so they should send compressed messages
     assert(remoteReadNextMessageAfterHello.size == 1)
-    assert(remoteReadNextMessageAfterHello.head.get == status)
+    assert(remoteReadNextMessageAfterHello.head == Right(status))
   }
 
   it should "compress and decompress first message after hello when receiving 2 frames" in new TestSetup {
@@ -72,8 +72,8 @@ class MessageCodecSpec extends AnyFlatSpec with Matchers {
 
     // both peers exchanged v5 hellos, so they should send compressed messages
     assert(remoteReadBothMessages.size == 2)
-    assert(remoteReadBothMessages.head.get == helloV5)
-    assert(remoteReadBothMessages.last.get == status)
+    assert(remoteReadBothMessages.head == Right(helloV5))
+    assert(remoteReadBothMessages.last == Right(status))
   }
 
   trait TestSetup extends SecureChannelSetup {