Merge remote-tracking branch 'origin/develop' into etcm-275/async-processing

Author: KonradStaniec

.buildkite/pipeline.yml

@@ -5,4 +5,5 @@ steps:
       nix eval --json '(import ./.buildkite { pipeline = ./.buildkite/pipeline.nix; })' \
       | buildkite-agent pipeline upload --no-interpolation
     agents:
-      queue: project42
+      queue: project42
+    timeout_in_minutes: 60

src/main/scala/io/iohk/ethereum/healthcheck/Healthcheck.scala

@@ -1,7 +1,6 @@
 package io.iohk.ethereum.healthcheck
 
-import scala.concurrent.{ExecutionContext, Future}
-import scala.util.{Failure, Success}
+import monix.eval.Task
 
 /**
   * Represents a health check, runs it and interprets the outcome.
@@ -18,20 +17,18 @@ import scala.util.{Failure, Success}
   */
 case class Healthcheck[Error, Result](
     description: String,
-    f: () => Future[Either[Error, Result]],
+    f: Task[Either[Error, Result]],
     mapResultToError: Result => Option[String] = (_: Result) => None,
     mapErrorToError: Error => Option[String] = (error: Error) => Some(String.valueOf(error)),
     mapExceptionToError: Throwable => Option[String] = (t: Throwable) => Some(String.valueOf(t))
 ) {
 
-  def apply()(implicit ec: ExecutionContext): Future[HealthcheckResult] = {
-    f().transform {
-      case Success(Left(error)) =>
-        Success(HealthcheckResult(description, mapErrorToError(error)))
-      case Success(Right(result)) =>
-        Success(HealthcheckResult(description, mapResultToError(result)))
-      case Failure(t) =>
-        Success(HealthcheckResult(description, mapExceptionToError(t)))
-    }
+  def apply(): Task[HealthcheckResult] = {
+    f.map {
+      case Left(error) =>
+        HealthcheckResult(description, mapErrorToError(error))
+      case Right(result) =>
+        HealthcheckResult(description, mapResultToError(result))
+    }.onErrorHandle(t => HealthcheckResult(description, mapExceptionToError(t)))
   }
 }

src/main/scala/io/iohk/ethereum/jsonrpc/AkkaTaskOps.scala

@@ -0,0 +1,20 @@
+package io.iohk.ethereum.jsonrpc
+
+import akka.actor.{Actor, ActorRef}
+import akka.pattern.ask
+import akka.util.Timeout
+import monix.eval.Task
+
+import scala.reflect.ClassTag
+
+object AkkaTaskOps {
+  implicit class TaskActorOps(val to: ActorRef) extends AnyVal {
+
+    def askFor[A](
+        message: Any
+    )(implicit timeout: Timeout, classTag: ClassTag[A], sender: ActorRef = Actor.noSender): Task[A] =
+      Task
+        .deferFuture((to ? message).mapTo[A])
+        .timeout(timeout.duration)
+  }
+}

src/main/scala/io/iohk/ethereum/jsonrpc/CheckpointingService.scala

@@ -6,9 +6,7 @@ import io.iohk.ethereum.blockchain.sync.regular.RegularSync.NewCheckpoint
 import io.iohk.ethereum.crypto.ECDSASignature
 import io.iohk.ethereum.domain.Blockchain
 import io.iohk.ethereum.utils.Logger
-import monix.execution.Scheduler.Implicits.global
-
-import scala.concurrent.Future
+import monix.eval.Task
 
 class CheckpointingService(
     blockchain: Blockchain,
@@ -21,12 +19,12 @@ class CheckpointingService(
     lazy val bestBlockNum = blockchain.getBestBlockNumber()
     lazy val blockToReturnNum = bestBlockNum - bestBlockNum % req.checkpointingInterval
 
-    Future {
+    Task {
       blockchain.getBlockByNumber(blockToReturnNum)
     }.flatMap {
       case Some(b) =>
         val resp = GetLatestBlockResponse(b.hash, b.number)
-        Future.successful(Right(resp))
+        Task.now(Right(resp))
 
       case None =>
         log.error(
@@ -37,7 +35,7 @@ class CheckpointingService(
     }
   }
 
-  def pushCheckpoint(req: PushCheckpointRequest): ServiceResponse[PushCheckpointResponse] = Future {
+  def pushCheckpoint(req: PushCheckpointRequest): ServiceResponse[PushCheckpointResponse] = Task {
     syncController ! NewCheckpoint(req.hash, req.signatures)
     Right(PushCheckpointResponse())
   }

src/main/scala/io/iohk/ethereum/jsonrpc/DebugService.scala

@@ -1,49 +1,44 @@
 package io.iohk.ethereum.jsonrpc
 
 import akka.actor.ActorRef
-import akka.pattern._
 import akka.util.Timeout
+import io.iohk.ethereum.jsonrpc.AkkaTaskOps._
 import io.iohk.ethereum.jsonrpc.DebugService.{ListPeersInfoRequest, ListPeersInfoResponse}
 import io.iohk.ethereum.network.EtcPeerManagerActor.{PeerInfo, PeerInfoResponse}
 import io.iohk.ethereum.network.PeerManagerActor.Peers
 import io.iohk.ethereum.network.{EtcPeerManagerActor, Peer, PeerActor, PeerId, PeerManagerActor}
+import monix.eval.Task
 
-import scala.concurrent.ExecutionContext.Implicits.global
-import scala.concurrent.Future
 import scala.concurrent.duration._
 
 object DebugService {
-
   case class ListPeersInfoRequest()
   case class ListPeersInfoResponse(peers: List[PeerInfo])
-
 }
 
 class DebugService(peerManager: ActorRef, etcPeerManager: ActorRef) {
 
   def listPeersInfo(getPeersInfoRequest: ListPeersInfoRequest): ServiceResponse[ListPeersInfoResponse] = {
-    val result = for {
+    for {
       ids <- getPeerIds
-      peers <- Future.traverse(ids)(getPeerInfo)
-    } yield ListPeersInfoResponse(peers.flatten)
-
-    result.map(Right(_))
+      peers <- Task.traverse(ids)(getPeerInfo)
+    } yield Right(ListPeersInfoResponse(peers.flatten))
   }
 
-  private def getPeerIds: Future[List[PeerId]] = {
+  private def getPeerIds: Task[List[PeerId]] = {
     implicit val timeout: Timeout = Timeout(5.seconds)
 
-    (peerManager ? PeerManagerActor.GetPeers)
-      .mapTo[Peers]
-      .recover { case _ => Peers(Map.empty[Peer, PeerActor.Status]) }
+    peerManager
+      .askFor[Peers](PeerManagerActor.GetPeers)
+      .onErrorRecover { case _ => Peers(Map.empty[Peer, PeerActor.Status]) }
       .map(_.peers.keySet.map(_.id).toList)
   }
 
-  private def getPeerInfo(peer: PeerId): Future[Option[PeerInfo]] = {
+  private def getPeerInfo(peer: PeerId): Task[Option[PeerInfo]] = {
     implicit val timeout: Timeout = Timeout(5.seconds)
 
-    (etcPeerManager ? EtcPeerManagerActor.PeerInfoRequest(peer))
-      .mapTo[PeerInfoResponse]
-      .collect { case PeerInfoResponse(info) => info }
+    etcPeerManager
+      .askFor[PeerInfoResponse](EtcPeerManagerActor.PeerInfoRequest(peer))
+      .map(resp => resp.peerInfo)
   }
 }