Merge pull request #33 from advancedtelematic/feat/pro-2615/emit-tree…

…hub-commit

[PRO-2615] send treehub commit event instead of dependency to core

build.sbt

@@ -31,7 +31,7 @@ lazy val root = (project in file("."))
     val akkaHttpV = "10.0.0"
     val scalaTestV = "3.0.0"
     val slickV = "3.1.1"
-    val sotaV = "0.2.53"
+    val sotaV = "0.2.81"
 
     Seq(
       "com.typesafe.akka" %% "akka-actor" % akkaV,

src/main/resources/application.conf

@@ -79,6 +79,7 @@ device_registry = {
   devicesUri = "/api/v1/devices"
   deviceGroupsUri = "/api/v1/device_groups"
   mydeviceUri = "/api/v1/mydevice"
+  packagesUri = "/api/v1/packages"
 }
 
 auth {

src/main/scala/com/advancedtelematic/treehub/Boot.scala

@@ -6,7 +6,7 @@ import akka.http.scaladsl.model.Uri
 import akka.http.scaladsl.server.{Directives, Route}
 import akka.stream.ActorMaterializer
 import cats.data.Xor
-import com.advancedtelematic.treehub.client.CoreClient
+import com.advancedtelematic.treehub.client._
 import com.advancedtelematic.treehub.http.{TreeHubRoutes, Http => TreeHubHttp}
 import com.advancedtelematic.treehub.object_store.{LocalFsBlobStore, ObjectStore, S3BlobStore, S3Credentials}
 import com.advancedtelematic.treehub.repo_metrics.{StorageUpdate, UsageMetricsRouter}
@@ -44,6 +44,7 @@ trait Settings {
   val deviceRegistryApi = Uri(config.getString("device_registry.devicesUri"))
   val deviceRegistryGroupApi = Uri(config.getString("device_registry.deviceGroupsUri"))
   val deviceRegistryMyApi = Uri(config.getString("device_registry.mydeviceUri"))
+  val deviceRegistryPackagesUri = Uri(config.getString("device_registry.packagesUri"))
 
   lazy val s3Credentials = {
     val accessKey = config.getString("treehub.s3.accessKey")
@@ -67,7 +68,7 @@ object Boot extends BootApp with Directives with Settings with VersionInfo
   log.info(s"Starting $version on http://$host:$port")
 
   val deviceRegistry = new DeviceRegistryClient(
-    deviceRegistryUri, deviceRegistryApi, deviceRegistryGroupApi, deviceRegistryMyApi
+    deviceRegistryUri, deviceRegistryApi, deviceRegistryGroupApi, deviceRegistryMyApi, deviceRegistryPackagesUri
   )
 
   val tokenValidator = TreeHubHttp.tokenValidator
@@ -82,18 +83,19 @@ object Boot extends BootApp with Directives with Settings with VersionInfo
   }
 
   val objectStore = new ObjectStore(storage)
-  val coreClient = new CoreClient(coreUri, packagesApi, treeHubUri.toString())
-
   val msgPublisher = MessageBus.publisher(system, config) match {
     case Xor.Right(mbp) => mbp
     case Xor.Left(err) => throw err
   }
+  val coreHttpClient = new CoreHttpClient(coreUri, packagesApi, treeHubUri)
+  val coreBusClient = new CoreBusClient(msgPublisher, treeHubUri)
 
   val usageHandler = system.actorOf(UsageMetricsRouter(msgPublisher, objectStore), "usage-router")
 
   val routes: Route =
     (versionHeaders(version) & logResponseMetrics(projectName) & TreeHubHttp.transformAtsAuthHeader) {
-      new TreeHubRoutes(tokenValidator, namespaceExtractor, coreClient, deviceNamespace, objectStore, usageHandler).routes
+      new TreeHubRoutes(tokenValidator, namespaceExtractor, coreHttpClient, coreBusClient,
+                        deviceNamespace, objectStore, usageHandler).routes
     }
 
   Http().bindAndHandle(routes, host, port)

src/main/scala/com/advancedtelematic/treehub/client/CoreBusClient.scala

@@ -0,0 +1,30 @@
+/**
+  * Copyright: Copyright (C) 2016, ATS Advanced Telematic Systems GmbH
+  * License: MPL-2.0
+  */
+package com.advancedtelematic.treehub.client
+
+import scala.concurrent.ExecutionContext
+import akka.http.scaladsl.model.Uri
+import org.genivi.sota.messaging.MessageBusPublisher
+import scala.concurrent.{ExecutionContext, Future}
+import com.advancedtelematic.data.DataType.{Commit, Ref, RefName}
+import org.genivi.sota.messaging.Messages.TreehubCommit
+
+
+class CoreBusClient(messageBusPublisher: MessageBusPublisher, treeHubUri: Uri) extends Core {
+
+  private def mkTreehubCommit(ref: Ref, description: String): TreehubCommit = {
+    //TODO: PRO-1802 pass the refname as the description until we can parse the real description out of the commit
+    val formattedRefName = ref.name.get.replaceFirst("^heads/", "").replace("/", "-")
+    val size = -1 // TODO
+    TreehubCommit(ref.namespace, ref.value.get, formattedRefName, description, size, treeHubUri.toString)
+  }
+
+  def publishRef(ref: Ref, description: String)
+                (implicit ec: ExecutionContext): Future[Unit] =
+    messageBusPublisher
+      .publishSafe(mkTreehubCommit(ref, description))
+      .map(_ => ())
+
+}

src/main/scala/com/advancedtelematic/treehub/client/CoreClient.scala → src/main/scala/com/advancedtelematic/treehub/client/CoreHttpClient.scala

@@ -20,8 +20,8 @@ import org.genivi.sota.marshalling.CirceMarshallingSupport
 import scala.concurrent.{ExecutionContext, Future}
 import scala.reflect.ClassTag
 
-class CoreClient(baseUri: Uri, packagesUri: Uri, treeHubUri: String)
-                (implicit system: ActorSystem, mat: ActorMaterializer) extends Core {
+class CoreHttpClient(baseUri: Uri, packagesUri: Uri, treeHubUri: Uri)
+                    (implicit system: ActorSystem, mat: ActorMaterializer) extends Core {
   import HttpMethods._
   import CirceMarshallingSupport._
 
@@ -31,7 +31,7 @@ class CoreClient(baseUri: Uri, packagesUri: Uri, treeHubUri: String)
 
   def publishRef(ref: Ref, description: String)
                 (implicit ec: ExecutionContext): Future[Unit] = {
-    val fileContents = ImageRequest(ref.value, ref.name, description, treeHubUri).asJson.noSpaces
+    val fileContents = ImageRequest(ref.value, ref.name, description, treeHubUri.toString).asJson.noSpaces
     val bodyPart = BodyPart.Strict("file", HttpEntity(fileContents), Map("fileName" -> ref.name.get))
     val formattedRefName = ref.name.get.replaceFirst("^heads/", "").replace("/", "-")
     val uri = baseUri.withPath(packagesUri.path + s"/treehub-$formattedRefName/${ref.value.get}")

src/main/scala/com/advancedtelematic/treehub/http/TreeHubRoutes.scala

@@ -2,30 +2,28 @@ package com.advancedtelematic.treehub.http
 
 import akka.http.scaladsl.server.{Directives, _}
 import akka.stream.Materializer
-import org.genivi.sota.data.Namespace
 import com.advancedtelematic.treehub.VersionInfo
 import com.advancedtelematic.treehub.client.Core
 import com.advancedtelematic.treehub.object_store.ObjectStore
 import com.advancedtelematic.treehub.repo_metrics.UsageMetricsRouter
+import org.genivi.sota.data.Namespace
 import org.genivi.sota.http.{ErrorHandler, HealthResource}
 import org.genivi.sota.rest.SotaRejectionHandler._
-
 import scala.concurrent.ExecutionContext
 import slick.driver.MySQLDriver.api._
 
-
 class TreeHubRoutes(tokenValidator: Directive0,
                     namespaceExtractor: Directive1[Namespace],
-                    coreClient: Core,
+                    coreHttpClient: Core,
+                    coreBusClient: Core,
                     deviceNamespace: Directive1[Namespace],
                     objectStore: ObjectStore,
-                    usageHandler: UsageMetricsRouter.HandlerRef
-                   )
+                    usageHandler: UsageMetricsRouter.HandlerRef)
                    (implicit val db: Database, ec: ExecutionContext, mat: Materializer) extends VersionInfo {
 
   import Directives._
 
-  def allRoutes(nsExtract: Directive1[Namespace]): Route = {
+  def allRoutes(nsExtract: Directive1[Namespace], coreClient: Core): Route = {
     new ConfResource().route ~
     new ObjectResource(nsExtract, objectStore, usageHandler).route ~
     new RefResource(nsExtract, coreClient, objectStore).route
@@ -35,10 +33,13 @@ class TreeHubRoutes(tokenValidator: Directive0,
     handleRejections(rejectionHandler) {
       ErrorHandler.handleErrors {
         (pathPrefix("api" / "v2") & tokenValidator) {
-            allRoutes(namespaceExtractor) ~
-            pathPrefix("mydevice") {
-              allRoutes(deviceNamespace)
-            }
+          allRoutes(namespaceExtractor, coreHttpClient) ~
+          pathPrefix("mydevice") {
+            allRoutes(deviceNamespace, coreHttpClient)
+          }
+        } ~
+        (pathPrefix("api" / "v3") & tokenValidator) {
+          allRoutes(namespaceExtractor, coreBusClient)
         } ~ new HealthResource(db, versionMap).route
       }
     }

src/test/scala/com/advancedtelematic/treehub/client/FakeBusCore.scala

@@ -0,0 +1,20 @@
+/**
+  * Copyright: Copyright (C) 2016, ATS Advanced Telematic Systems GmbH
+  * License: MPL-2.0
+  */
+package com.advancedtelematic.treehub.client
+
+import akka.actor.ActorSystem
+import com.advancedtelematic.data.DataType.{Commit, Ref}
+import org.genivi.sota.messaging.LocalMessageBus
+import scala.concurrent.{ExecutionContext, Future}
+
+class FakeBusCore()
+      (implicit val system: ActorSystem) extends Core {
+
+  val fakeMsgPublisher = LocalMessageBus.publisher(system)
+
+  override def publishRef(ref: Ref, description: String)
+                         (implicit ec: ExecutionContext): Future[Unit] = Future.successful(())
+
+}

src/test/scala/com/advancedtelematic/treehub/http/FakeCore.scala → src/test/scala/com/advancedtelematic/treehub/client/FakeHttpCore.scala

@@ -2,15 +2,14 @@
   * Copyright: Copyright (C) 2016, ATS Advanced Telematic Systems GmbH
   * License: MPL-2.0
   */
-package com.advancedtelematic.treehub.http
+package com.advancedtelematic.treehub.client
 
 import com.advancedtelematic.data.DataType.{Commit, Ref}
-import com.advancedtelematic.treehub.client.Core
 import org.genivi.sota.data.Namespace
 
 import scala.concurrent.{ExecutionContext, Future}
 
-class FakeCore() extends Core {
+class FakeHttpCore() extends Core {
   override def publishRef(ref: Ref, description: String)
                          (implicit ec: ExecutionContext): Future[Unit] = Future.successful(())
 }

src/test/scala/com/advancedtelematic/util/ResourceSpec.scala

@@ -1,24 +1,24 @@
 package com.advancedtelematic.util
 
-import java.nio.file.Files
-
 import akka.actor.{Actor, ActorLogging, Props}
 import akka.http.scaladsl.model.Multipart.FormData.BodyPart
 import akka.http.scaladsl.model.{HttpEntity, MediaTypes, Multipart}
 import akka.http.scaladsl.server.Directives
 import akka.http.scaladsl.testkit.ScalatestRouteTest
 import com.advancedtelematic.common.DigestCalculator
 import com.advancedtelematic.data.DataType.{Commit, ObjectId}
+import com.advancedtelematic.treehub.Settings
+import com.advancedtelematic.treehub.client._
 import com.advancedtelematic.treehub.http._
 import com.advancedtelematic.treehub.object_store.{LocalFsBlobStore, ObjectStore}
 import com.advancedtelematic.treehub.repo_metrics.UsageMetricsRouter.{UpdateBandwidth, UpdateStorage}
 import com.advancedtelematic.util.FakeUsageUpdate.{CurrentBandwith, CurrentStorage}
 import eu.timepit.refined.api.Refined
+import java.nio.file.Files
 import org.genivi.sota.core.DatabaseSpec
 import org.genivi.sota.data.Namespace
 import org.genivi.sota.http.NamespaceDirectives
 import org.scalatest.Suite
-
 import scala.util.Random
 
 object ResourceSpec {
@@ -71,12 +71,14 @@ class FakeUsageUpdate extends Actor with ActorLogging {
   }
 }
 
-trait ResourceSpec extends ScalatestRouteTest with DatabaseSpec {
+trait ResourceSpec extends ScalatestRouteTest with DatabaseSpec with Settings {
   self: Suite =>
 
   def apiUri(path: String): String = "/api/v2/" + path
+  def apiUri(version: Int, path: String): String = s"/api/v$version/" + path
 
-  val testCore = new FakeCore()
+  val testHttpCore = new FakeHttpCore()
+  val testBusCore = new FakeBusCore()
 
   lazy val namespaceExtractor = NamespaceDirectives.defaultNamespaceExtractor.map(_.namespace)
   val objectStore = new ObjectStore(new LocalFsBlobStore(Files.createTempDirectory("treehub").toFile))
@@ -85,7 +87,8 @@ trait ResourceSpec extends ScalatestRouteTest with DatabaseSpec {
 
   lazy val routes = new TreeHubRoutes(Directives.pass,
     namespaceExtractor,
-    testCore,
+    testHttpCore,
+    testBusCore,
     namespaceExtractor,
     objectStore,
     fakeUsageUpdate).routes