Feature/futures

build.sbt

@@ -65,6 +65,13 @@ lazy val gatlingDependencies = Seq(
   "io.gatling" % "gatling-test-framework" % "3.9.5" % "it,test" exclude("com.typesafe.scala-logging", "scala-logging_2.13"),
 )
 
+lazy val akkaDependencies = Seq(
+  ("com.typesafe.akka" %% "akka-http" % "10.5.0").cross(CrossVersion.for3Use2_13),
+  ("com.typesafe.akka" %% "akka-actor" % "2.8.0").cross(CrossVersion.for3Use2_13),
+  ("com.typesafe.akka" %% "akka-actor-typed" % "2.8.0").cross(CrossVersion.for3Use2_13),
+  ("com.typesafe.akka" %% "akka-stream" % "2.8.0").cross(CrossVersion.for3Use2_13)
+)
+
 /* =====================================================================================================================
  * Project Settings
  * ===================================================================================================================== */
@@ -98,6 +105,7 @@ lazy val core = project
     libraryDependencies ++= http4sDependencies,
     libraryDependencies ++= gatlingDependencies,
     libraryDependencies ++= integrationTestDependencies,
+    libraryDependencies ++= akkaDependencies,
     dockerExposedPorts ++= Seq(9090),
     Defaults.itSettings,
     IntegrationTest / fork := true,
@@ -114,6 +122,7 @@ lazy val persistence = project
     libraryDependencies ++= http4sDependencies,
     libraryDependencies ++= gatlingDependencies,
     libraryDependencies ++= integrationTestDependencies,
+    libraryDependencies ++= akkaDependencies,
     libraryDependencies ++= Seq(
       "org.scala-lang.modules" %% "scala-xml" % "2.1.0", // XML
       "com.lihaoyi" %% "upickle" % "3.1.0", // JSON

core/src/main/scala/lib/CoreRestClient.scala

@@ -1,5 +1,13 @@
 package lib
 
+import akka.actor.typed.ActorSystem
+import akka.actor.typed.scaladsl.Behaviors
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.model.*
+import akka.http.scaladsl.model.HttpMethods.*
+import akka.http.scaladsl.server.Directives.*
+import akka.http.scaladsl.server.{ExceptionHandler, Route}
+import akka.http.scaladsl.unmarshalling.Unmarshal
 import com.typesafe.config.ConfigFactory
 import com.typesafe.scalalogging.StrictLogging
 import di.CoreModule
@@ -9,53 +17,72 @@ import lib.field.FieldInterface
 import lib.json.HexJson
 import requests.{Requester, Response, get, post}
 
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.duration.Duration
+import concurrent.duration.DurationInt
+import scala.concurrent.{Await, Future}
+import scala.language.postfixOps
 import scala.util.{Failure, Success, Try}
 
 case class CoreRestClient() extends ControllerInterface[Player] with StrictLogging:
   private lazy val config = ConfigFactory.load()
+  private val maxWaitSeconds: Duration = config.getInt("db.maxWaitSeconds") seconds
+
   private val fallBackUrl = "http://0.0.0.0:8080"
   private val coreUrl =
     Try(s"http://${config.getString("http.core.host")}:${config.getString("http.core.port")}") match
       case Success(value) => value
       case Failure(exception) => logger.error(s"${exception.getMessage} - Using fallback url: $fallBackUrl"); fallBackUrl
+  private val system: ActorSystem[Any] = ActorSystem(Behaviors.empty, "my-system")
+  given ActorSystem[Any] = system
 
   var hexField: FieldInterface[Player] =
     HexJson.decode(exportField) match
       case Success(value) => value
       case Failure(_) => null
 
-  override def gameStatus: GameStatus = GameStatus.valueOf(
-    fetch(get, s"$coreUrl/status") match
+  override def gameStatus: GameStatus =
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.GET, uri = s"$coreUrl/status"))
+    GameStatus.valueOf(getResponseAsString(response) match
       case "" => "ERROR"
-      case x => x
-  )
+      case x => x)
 
   override def fillAll(c: Player): Unit =
-    validate(HexJson.decode(fetch(post, s"$coreUrl/fillAll/$c")))
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.POST, uri = s"$coreUrl/fillAll/$c"))
+    Try(validateResponse(response))
 
   override def save(): Try[Unit] =
-    Try {
-      validate(HexJson.decode(fetch(post, s"$coreUrl/save")))
-      Success(())
-    }
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.POST, uri = s"$coreUrl/save",
+      entity = HttpEntity(HexJson.encode(hexField))))
+    Try(validateResponse(response))
 
   override def load(): Try[Unit] =
-    Try {
-      validate(HexJson.decode(fetch(get, s"$coreUrl/load")))
-      Success(())
-    }
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.GET, uri = s"$coreUrl/load"))
+    Try(validateResponse(response))
 
   override def place(c: Player, x: Int, y: Int): Unit =
-    validate(HexJson.decode(fetch(post, s"$coreUrl/place/$c/$x/$y")))
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.POST, uri = s"$coreUrl/place/$c/$x/$y"))
+    validateResponse(response)
 
   override def undo(): Unit =
-    validate(HexJson.decode(fetch(post, s"$coreUrl/undo")))
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.POST, uri = s"$coreUrl/undo"))
+    validateResponse(response)
 
   override def redo(): Unit =
-    validate(HexJson.decode(fetch(post, s"$coreUrl/redo")))
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.POST, uri = s"$coreUrl/redo"))
+    validateResponse(response)
 
   override def reset(): Unit =
-    validate(HexJson.decode(fetch(post, s"$coreUrl/reset")))
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.POST, uri = s"$coreUrl/reset"))
+    validateResponse(response)
 
   private def validate(res: Try[FieldInterface[Player]]): Unit =
     res match
@@ -64,4 +91,14 @@ case class CoreRestClient() extends ControllerInterface[Player] with StrictLoggi
         notifyObservers()
       case Failure(_) => logger.error("Failed to decode field")
 
-  override def exportField: String = fetch(get, s"$coreUrl/exportField")
+  override def exportField: String =
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.GET, uri = s"$coreUrl/exportField"))
+    getResponseAsString(response)
+
+  private def validateResponse(res: Future[HttpResponse]): Unit =
+    validate(HexJson.decode(getResponseAsString(res)))
+
+  private def getResponseAsString(res: Future[HttpResponse]): String =
+    val responseAsString: Future[String] = Unmarshal(Await.result(res, maxWaitSeconds).entity).to[String]
+    Await.result(responseAsString, maxWaitSeconds)

persistence/src/it/scala/DAOIntegrationSpec.scala

@@ -4,6 +4,7 @@ import com.typesafe.config.ConfigFactory
 import lib.Player
 import lib.database.DAOInterface
 import lib.database.mongoDB.DAOMongo
+import lib.database.mongoDB.DAOMongo.config
 import lib.database.slick.defaultImpl.DAOSlick
 import lib.database.slick.jsonImpl.DAOSlick as DAOSlickJson
 import lib.field.FieldInterface
@@ -14,12 +15,16 @@ import org.scalatest.wordspec.AnyWordSpec
 import org.testcontainers.containers.wait.strategy.Wait
 
 import java.io.File
+import scala.concurrent.Await
+import scala.concurrent.duration.{Duration, DurationInt}
 import scala.jdk.CollectionConverters.*
+import scala.language.postfixOps
 import scala.util.{Failure, Success}
 
 class DAOIntegrationSpec extends AnyWordSpec with TestContainerForAll:
 
   private lazy val config = ConfigFactory.load()
+  private val maxWaitSeconds: Duration = config.getInt("db.maxWaitSeconds") seconds
 
   override val containerDef: ContainerDef = DockerComposeContainer.Def(
     new File("db-integration-test.yml"),
@@ -32,27 +37,27 @@ class DAOIntegrationSpec extends AnyWordSpec with TestContainerForAll:
   "The Slick DAO" when {
     "nothing is saved" should {
       "not be able to load a field" in {
-        DAOSlick.load() shouldBe a[Failure[_]]
+        Await.result(DAOSlick.load(), maxWaitSeconds) shouldBe a[Failure[_]]
       }
     }
     "something is saved" should {
       "be able to save a field" in {
         val mockField = Field()(using new Matrix(5, 5))
-        DAOSlick.save(mockField) shouldBe a[Success[_]]
+        Await.result(DAOSlick.save(mockField), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to load a field" in {
-        DAOSlick.load() shouldBe a[Success[_]]
+        Await.result(DAOSlick.load(), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to delete a field" in {
-        DAOSlick.delete(Some(1)) shouldBe a[Success[_]]
+        Await.result(DAOSlick.delete(Some(1)), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to update a field" in {
         val mockField = Field()(using new Matrix(5, 5))
         val updatedField = mockField.place(Player.fromChar('X'), 0, 0)
-        DAOSlick.update(0, updatedField) shouldBe a[Success[_]]
+        Await.result(DAOSlick.update(0, updatedField), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to load updated field" in {
-        val field = DAOSlick.load(Some(0)).get
+        val field = Await.result(DAOSlick.load(Some(0)), maxWaitSeconds).get
         field shouldBe a[FieldInterface[Player]]
         field.matrix.cell(0, 0) shouldBe Player.fromChar('X')
       }
@@ -61,27 +66,27 @@ class DAOIntegrationSpec extends AnyWordSpec with TestContainerForAll:
   "The Slick DAO with JSON" when {
     "nothing is saved" should {
       "not be able to load a field" in {
-        DAOSlickJson.load() shouldBe a[Failure[_]]
+        Await.result(DAOSlickJson.load(), maxWaitSeconds) shouldBe a[Failure[_]]
       }
     }
     "something is saved" should {
       "be able to save a field" in {
         val mockField = Field()(using new Matrix(5, 5))
-        DAOSlickJson.save(mockField) shouldBe a[Success[_]]
+        Await.result(DAOSlickJson.save(mockField), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to load a field" in {
-        DAOSlickJson.load() shouldBe a[Success[_]]
+        Await.result(DAOSlickJson.load(), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to delete a field" in {
-        DAOSlickJson.delete(Some(1)) shouldBe a[Success[_]]
+        Await.result(DAOSlickJson.delete(Some(1)), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to update a field" in {
         val mockField = Field()(using new Matrix(5, 5))
         val updatedField = mockField.place(Player.fromChar('X'), 0, 0)
-        DAOSlickJson.update(0, updatedField) shouldBe a[Success[_]]
+        Await.result(DAOSlickJson.update(0, updatedField), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to load updated field" in {
-        val field = DAOSlickJson.load(Some(0)).get
+        val field = Await.result(DAOSlickJson.load(Some(0)), maxWaitSeconds).get
         field shouldBe a[FieldInterface[Player]]
         field.matrix.cell(0, 0) shouldBe Player.fromChar('X')
       }
@@ -90,27 +95,27 @@ class DAOIntegrationSpec extends AnyWordSpec with TestContainerForAll:
   "The MongoDB DAO" when {
     "nothing is saved" should {
       "not be able to load a field" in {
-        DAOMongo.load() shouldBe a[Failure[_]]
+        Await.result(DAOMongo.load(), maxWaitSeconds) shouldBe a[Failure[_]]
       }
     }
     "something is saved" should {
       "be able to save a field" in {
         val mockField = Field()(using new Matrix(5, 5))
-        DAOMongo.save(mockField) shouldBe a[Success[_]]
+        Await.result(DAOMongo.save(mockField), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to load a field" in {
-        DAOMongo.load() shouldBe a[Success[_]]
+        Await.result(DAOMongo.load(), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to delete a field" in {
-        DAOMongo.delete(Some(1)) shouldBe a[Success[_]]
+        Await.result(DAOMongo.delete(Some(1)), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to update a field" in {
         val mockField = Field()(using new Matrix(5, 5))
         val updatedField = mockField.place(Player.fromChar('X'), 0, 0)
-        DAOMongo.update(0, updatedField) shouldBe a[Success[_]]
+        Await.result(DAOMongo.update(0, updatedField), maxWaitSeconds) shouldBe a[Success[_]]
       }
       "be able to load updated field" in {
-        val field = DAOMongo.load(Some(0)).get
+        val field = Await.result(DAOMongo.load(Some(0)), scala.concurrent.duration.Duration.Inf).get
         field shouldBe a[FieldInterface[Player]]
         field.matrix.cell(0, 0) shouldBe Player.fromChar('X')
       }

persistence/src/it/scala/gatling/GatlingBaseSpec.scala

@@ -9,8 +9,8 @@ import io.gatling.core.structure.ScenarioBuilder
 import io.gatling.http.Predef.*
 import io.gatling.http.protocol.HttpProtocolBuilder
 import io.gatling.jdbc.Predef.*
-import org.testcontainers.containers.wait.strategy.Wait
 import lib.json.HexJson
+import org.testcontainers.containers.wait.strategy.Wait
 
 import java.io.File
 import scala.concurrent.duration.*

persistence/src/main/scala/lib/FileIORestClient.scala

@@ -1,32 +1,65 @@
 package lib
 
+import akka.actor.typed.ActorSystem
+import akka.actor.typed.scaladsl.Behaviors
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.model.*
+import akka.http.scaladsl.model.HttpMethods.*
+import akka.http.scaladsl.server.Directives.*
+import akka.http.scaladsl.server.{ExceptionHandler, Route}
+import akka.http.scaladsl.unmarshalling.Unmarshal
 import com.typesafe.config.ConfigFactory
 import com.typesafe.scalalogging.StrictLogging
 import lib.Http.fetch
 import lib.field.FieldInterface
 import lib.json.HexJson
 import requests.{get, post}
 
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.duration.Duration
+import concurrent.duration.DurationInt
+import scala.concurrent.{Await, Future}
+import scala.language.postfixOps
 import scala.util.{Failure, Success, Try}
 
-case class FileIORestClient() extends FileIOInterface[Player] with StrictLogging {
-
+case class FileIORestClient() extends FileIOInterface[Player] with StrictLogging:
   private lazy val config = ConfigFactory.load()
+  private val maxWaitSeconds: Duration = config.getInt("db.maxWaitSeconds") seconds
+
   private val persistenceUrl =
     Try(s"http://${config.getString("http.persistence.host")}:${config.getString("http.persistence.port")}") match
       case Success(value) => value
       case Failure(exception) => logger.error(exception.getMessage); "http://0.0.0.0:8081"
-
+  private val system: ActorSystem[Any] = ActorSystem(Behaviors.empty, "my-system")
+  given ActorSystem[Any] = system
+
   override def load: Try[FieldInterface[Player]] =
-    HexJson.decode(fetch(get, s"$persistenceUrl/load"))
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.GET, uri = s"$persistenceUrl/load"))
+    HexJson.decode(getResponseAsString(response))
 
   override def save(field: FieldInterface[Player]): Try[Unit] =
-    Try {
-      fetch(post, s"$persistenceUrl/save", HexJson.encode(field))
-    } match {
-      case Failure(exception) => throw exception
-      case Success(_) => Success(())
-    }
-
-  override def exportGame(field: FieldInterface[Player], xCount: Int, oCount: Int, turn: Int): String = fetch(get, s"$persistenceUrl/load")
-}
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.POST, uri = s"$persistenceUrl/save",
+      entity = HexJson.encode(field)))
+    processResponse(response)
+
+  override def exportGame(field: FieldInterface[Player], xCount: Int, oCount: Int, turn: Int): String =
+    val response: Future[HttpResponse] =
+      Http().singleRequest(HttpRequest(method = HttpMethods.GET, uri = s"$persistenceUrl/exportField"))
+    getResponseAsString(response)
+
+  private def processResponse(res: Future[HttpResponse]): Try[Unit] =
+    Await.result(res, maxWaitSeconds) match
+      case HttpResponse(StatusCodes.OK, _, entity, _) =>
+        Success(())
+      case HttpResponse(code, _, entity, _) =>
+        logger.error(s"Request failed with code $code and entity $entity")
+        Failure(new Exception(s"Request failed with code $code and entity $entity"))
+
+  private def decodeResponse(res: Future[HttpResponse]): Try[FieldInterface[Player]] =
+    HexJson.decode(getResponseAsString(res))
+
+  private def getResponseAsString(res: Future[HttpResponse]): String =
+    val responseAsString: Future[String] = Unmarshal(Await.result(res, maxWaitSeconds).entity).to[String]
+    Await.result(responseAsString, maxWaitSeconds)