Migrate from Spray to Akka-Http (#11)

* add -feature flag to scala compile to show feature warnings * Remove cors support. This is no longer needed since the management UI server communicates with the marketplace instead management UI website itself * migrate spray routing to akka routing * run standalone application instead of spray servlet; fix issue with authorization header -> use akka Authorization object instead of strings * reformat routes code * remove spray libraries; fix http request for getting key from producer * use pathmatcher Remaining instead of Segment to work with tokens/ids containing slashes * first product test runs * all product tests run * consumer test work again; remove some unused options; deleting a consumer is only possible with the correct token in the auth header * merchant tests up and running * offer tests work; disabled some tests because its currently not possible to properly test signature validation (cannot add offers in tests) * instructions for testing * remove unused dependencies
hpi-epic · Oct 15, 2018 · 82b7559 · 82b7559
1 parent e756b5f
commit 82b7559
Show file tree

Hide file tree

Showing 22 changed files with 649 additions and 708 deletions.
diff --git a/Dockerfile b/Dockerfile
@@ -4,14 +4,9 @@ ENV APP_HOME /marketplace
 RUN mkdir $APP_HOME
 WORKDIR $APP_HOME
 
-ADD build.sbt $APP_HOME
-ADD project/build.properties $APP_HOME/project/
-ADD project/plugins.sbt $APP_HOME/project/
-
-RUN sbt update
-
 ADD . $APP_HOME
 
+RUN sbt update
 RUN sbt compile
 
 CMD ["./start-marketplace.sh"]
diff --git a/README.md b/README.md
@@ -20,19 +20,37 @@ The marketplace is written in Scala and managed with sbt. Ensure to have a sbt i
 
 ## Setup
 
-First, create a new database for the marketplace to connect to on `localhost:5432` or change the corresponding settings: `CREATE DATABASE marketplace;`
+First, create a new Postgres database for the marketplace to connect to on `localhost:5432` or change the corresponding settings: `CREATE DATABASE marketplace;`
 * Database Name: `marketplace`
 * Database User: `postgres`
 * Database User password: ` ` (none)
 
-After cloning the repository, run `sbt update` and `sbt compile` to download all required dependencies and compile the sorce code. Once you're done, you might start the server by either running `sbt ~tomcat:start` or `sbt tomcat:start`. The difference between the two commands is, that the first one with the ~ reloads all source code files (incl. compiling) when changes are made to any of these, while the second doesn't reflect changes made to files after the server has been started.
+After cloning the repository, run `sbt update` and `sbt compile` to download all required dependencies and compile the sorce code.
+Once you're done, you might start the server by either running `sbt run`.
 
 In the default configuration, you can access the server with `http://localhost:8080/` in your browser or by using your preferred REST GUI. All available routes are documented in our global API definition, available here: https://hpi-epic.github.io/masterproject-pricewars.
 
 ## Configuration
 
 All pre-defined config settings are located in `/src/main/resources/application.conf`. Where appropiate, values from the environment are used (for the Docker setup) and a specialized config file `/src/main/resources/application.deployment.conf` is present for our VM deployment.
 
+## Testing
+
+As a prerequisite to run tests the three services `Kafka`, `Redis`, and `Postgres` must run.
+
+You can use the configuration from the Price Wars repository and run Kafka and Redis with the command `docker-compose up kafka redis`.
+Kafka could be stubbed out for tests but this is not implemented yet.
+
+We recommend the following commands to run a local Postgres instance as Docker container used for testing:
+```
+docker run -d -p 5432:5432 --name marketplace_test_db -e POSTGRES_PASSWORD="" -d postgres:9.6.5
+docker exec -it marketplace_test_db bash
+psql -U postgres
+CREATE DATABASE marketplace_test;
+```
+
+If these services are running you can run marketplace tests with `sbt test`.
+
 ## Concept
 
 The marketplace acts as the main platform to trade products offered by different merchants, to enforce time limits and is the main source to feed Kafka with various logs. Therefore (and due to the authorization required for several actions), it is the main source of trust within a simulation.

diff --git a/build.sbt b/build.sbt
@@ -4,43 +4,38 @@ version := "1.0"
 
 scalaVersion := "2.11.8"
 
-scalacOptions := Seq("-unchecked", "-deprecation", "-encoding", "utf8")
+scalacOptions := Seq("-unchecked", "-deprecation", "-encoding", "utf8", "-feature")
 
 resolvers += Resolver.bintrayRepo("cakesolutions", "maven")
 
 libraryDependencies ++= {
   val akkaV = "2.5.16"
+  val akkaHttpV = "10.1.5"
   val sprayV = "1.3.3"
   val specs2V = "3.8.6"
   val scalikejdbcV = "3.3.0"
-  val logbackV = "1.1.8"
   val kafkaV = "1.1.1"
-  val scalaxmlV = "1.0.6"
   val commonsCodecV = "1.10"
   val redisClientV = "3.8"
   val scalarxV = "0.3.2"
 
   Seq(
-    "io.spray"                %%  "spray-servlet"             % sprayV,
-    "io.spray"                %%  "spray-routing"             % sprayV,
-    "io.spray"                %%  "spray-json"                % sprayV,
-    "io.spray"                %%  "spray-can"                 % sprayV,
-    "io.spray"                %%  "spray-testkit"             % sprayV % "test" exclude("org.specs2", "specs2_2.11"),
-    "com.typesafe.akka"       %%  "akka-actor"                % akkaV,
-    "com.typesafe.akka"       %%  "akka-testkit"              % akkaV % "test",
-    "com.typesafe.akka"       %%  "akka-slf4j"                % akkaV,
-    "org.specs2"              %%  "specs2-core"               % specs2V % "test",
-    "org.scalikejdbc"         %%  "scalikejdbc"               % scalikejdbcV,
-    "org.scalikejdbc"         %%  "scalikejdbc-config"        % scalikejdbcV,
-    "ch.qos.logback"          %   "logback-classic"           % logbackV,
-    "net.cakesolutions"       %% "scala-kafka-client-akka"    % kafkaV,
-    "org.scala-lang.modules"  %%  "scala-xml"                 % scalaxmlV,
-    "commons-codec"           %  "commons-codec"              % commonsCodecV,
-    "net.debasishg"           %% "redisclient"                % redisClientV,
-    "com.lihaoyi"             %% "scalarx"                    % scalarxV
+    "io.spray"                %% "spray-json"                % sprayV,
+    "com.typesafe.akka"       %% "akka-actor"                % akkaV,
+    "com.typesafe.akka"       %% "akka-testkit"              % akkaV % "test",
+    "com.typesafe.akka"       %% "akka-slf4j"                % akkaV,
+    "com.typesafe.akka"       %% "akka-stream"               % akkaV,
+    "com.typesafe.akka"       %% "akka-http"                 % akkaHttpV,
+    "com.typesafe.akka"       %% "akka-http-spray-json"      % akkaHttpV,
+    "com.typesafe.akka"       %% "akka-http-testkit"         % akkaHttpV % "test",
+    "org.specs2"              %% "specs2-core"               % specs2V % "test",
+    "org.scalikejdbc"         %% "scalikejdbc"               % scalikejdbcV,
+    "org.scalikejdbc"         %% "scalikejdbc-config"        % scalikejdbcV,
+    "net.cakesolutions"       %% "scala-kafka-client-akka"   % kafkaV,
+    "commons-codec"           %  "commons-codec"             % commonsCodecV,
+    "net.debasishg"           %% "redisclient"               % redisClientV,
+    "com.lihaoyi"             %% "scalarx"                   % scalarxV
   )
 }
 
 parallelExecution in Test := false
-
-enablePlugins(TomcatPlugin)
diff --git a/src/main/scala/de/hpi/epic/pricewars/CORSSupport.scala b/src/main/scala/de/hpi/epic/pricewars/CORSSupport.scala
diff --git a/src/main/scala/de/hpi/epic/pricewars/Server.scala b/src/main/scala/de/hpi/epic/pricewars/Server.scala
@@ -1,11 +1,17 @@
 package de.hpi.epic.pricewars
 
-import akka.actor.{ActorSystem, Props}
-import de.hpi.epic.pricewars.services.{DatabaseStore, MarketplaceServiceActor}
-import spray.servlet.WebBoot
-
-class Server extends WebBoot {
-  implicit val system = ActorSystem("marketplace")
-  val serviceActor = system.actorOf(Props[MarketplaceServiceActor])
-  DatabaseStore.setup()
+import akka.actor.ActorSystem
+import akka.http.scaladsl.Http
+import akka.stream.ActorMaterializer
+import de.hpi.epic.pricewars.services.{DatabaseStore, MarketplaceService}
+
+object Server {
+  def main(args: Array[String]) {
+
+    implicit val system: ActorSystem = ActorSystem("marketplace")
+    implicit val materializer: ActorMaterializer = ActorMaterializer()
+
+    DatabaseStore.setup()
+    Http().bindAndHandle(MarketplaceService.route, "0.0.0.0", port = 8080)
+  }
 }
diff --git a/src/main/scala/de/hpi/epic/pricewars/connectors/MerchantConnector.scala b/src/main/scala/de/hpi/epic/pricewars/connectors/MerchantConnector.scala
@@ -1,33 +1,32 @@
 package de.hpi.epic.pricewars.connectors
 
 import akka.actor.ActorSystem
-import akka.io.IO
-import akka.pattern.ask
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.model._
 import akka.util.Timeout
 import com.typesafe.config.{Config, ConfigFactory}
 import de.hpi.epic.pricewars.data.{Merchant, Offer}
 import de.hpi.epic.pricewars.services.DatabaseStore
-import spray.can.Http
-import spray.http.HttpMethods._
-import spray.http._
 
 import scala.concurrent.duration._
 
 object MerchantConnector {
 
   implicit val system: ActorSystem = ActorSystem()
   implicit val timeout: Timeout = Timeout(15.seconds)
+
   import system.dispatcher // implicit execution context
 
   val config: Config = ConfigFactory.load
   val remove_merchant = config.getBoolean("remove_merchant_on_notification_error")
 
   def notifyMerchant(merchant: Merchant, offer_id: Long, amount: Int, price: BigDecimal, offer: Offer) = {
     val json = s"""{"offer_id": $offer_id, "uid": ${offer.uid}, "product_id": ${offer.product_id}, "quality": ${offer.quality}, "amount_sold": $amount, "price_sold": $price, "price": ${offer.price}, "merchant_id": "${merchant.merchant_id.get}", "amount": ${offer.amount}}"""
-    val request = (IO(Http) ? HttpRequest(POST,
-      Uri(merchant.api_endpoint_url + "/sold"),
-      entity = HttpEntity(MediaTypes.`application/json`, json)
-    )).mapTo[HttpResponse]
+    val request = Http().singleRequest(HttpRequest(
+      HttpMethods.POST,
+      merchant.api_endpoint_url + "/sold",
+      entity = HttpEntity(MediaTypes.`application/json`, json)))
+
     def errorHandler(): Unit = {
       if (remove_merchant) {
         println("merchant not responding, killing: " + merchant.algorithm_name)
@@ -36,6 +35,7 @@ object MerchantConnector {
         println("merchant not responding, not killing: " + merchant.algorithm_name)
       }
     }
+
     request.onFailure {
       case t: Throwable =>
         println(t.getMessage)
@@ -44,11 +44,12 @@ object MerchantConnector {
       case _ =>
         errorHandler()
     }
-    request.onSuccess{ case HttpResponse(status, _, _, _) => {
+    request.onSuccess { case HttpResponse(status, _, _, _) => {
       if (status == StatusCodes.PreconditionRequired) {
         println("merchant requested to be deleted: " + merchant.algorithm_name)
         DatabaseStore.deleteMerchant(merchant.merchant_id.get, delete_with_token = false)
       }
-    }}
+    }
+    }
   }
 }
diff --git a/src/main/scala/de/hpi/epic/pricewars/connectors/ProducerConnector.scala b/src/main/scala/de/hpi/epic/pricewars/connectors/ProducerConnector.scala
@@ -2,46 +2,43 @@ package de.hpi.epic.pricewars.connectors
 
 import javax.crypto.Cipher
 import javax.crypto.spec.SecretKeySpec
-
 import akka.actor.ActorSystem
-import akka.io.IO
-import akka.pattern.ask
-import akka.util.Timeout
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.model.{HttpRequest, HttpResponse, StatusCodes}
+import akka.http.scaladsl.unmarshalling.Unmarshal
+import akka.stream.ActorMaterializer
 import com.typesafe.config.{Config, ConfigFactory}
 import org.apache.commons.codec.binary.Base64
-import spray.can.Http
-import spray.http.HttpMethods._
-import spray.http._
 import spray.json._
 
-import scala.concurrent.Await
+import scala.concurrent.{Await, ExecutionContextExecutor}
 import scala.concurrent.duration._
-
 import de.hpi.epic.pricewars.data.Signature
 
+
 object ProducerConnector {
   val config: Config = ConfigFactory.load
 
   val producer_url: String = config.getString("producer_url")
   var producer_key: Option[String] = None
 
   implicit val system: ActorSystem = ActorSystem()
-  implicit val timeout: Timeout = Timeout(15.seconds)
+  // needed for onComplete
+  implicit val executionContext: ExecutionContextExecutor = system.dispatcher
+  // used for Unmarshal
+  implicit val materializer: ActorMaterializer = ActorMaterializer()
 
   def getProducerKey(url: String = producer_url): Option[String] = {
-    val request = (IO(Http) ? HttpRequest(GET, Uri(url + "/decryption_key"))).mapTo[HttpResponse]
-    try {
-      Await.result(request, Duration.Inf) match {
-        case HttpResponse(status, entity, headers, protocol) =>
-          val response_json = entity.asString.parseJson
-          response_json.asJsObject.getFields("decryption_key") match {
-            case Seq(JsString(decryption_key)) =>
-              println("Producer Key updated!: " + decryption_key)
-              Some(decryption_key)
+    val responseFuture = Http().singleRequest(HttpRequest(uri = url + "/decryption_key"))
+    Await.result(responseFuture, 1.second) match {
+      case response @ HttpResponse(StatusCodes.OK, _, _, _) =>
+        Await.result(Unmarshal(response.entity).to[String].map { jsonString =>
+          jsonString.parseJson.asJsObject.getFields("decryption_key") match {
+            case Seq(JsString(decryption_key)) => Some(decryption_key)
+            case _ => None
           }
-      }
-    } catch {
-      case e: Exception => None
+        }, 1.second)
+      case _ => None
     }
   }