Merge pull request #113 from bitrockteam/hotfix/combine-stream-comple…

…tion-result Combine stream completion
bitrockteam · Jun 1, 2020 · 513216e · 513216e
2 parents 51731d9 + 8271611
commit 513216e
Show file tree

Hide file tree

Showing 5 changed files with 16 additions and 11 deletions.
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
@@ -10,7 +10,7 @@ object Dependencies {
 
   object Versions {
     lazy val Scala               = "2.12.10"
-    lazy val Akka                = "2.6.3"
+    lazy val Akka                = "2.6.4"
     lazy val AkkaHttp            = "10.1.11"
     lazy val Alpakka             = "2.0.2"
     lazy val ConfluentPlatform   = "5.4.0"

diff --git a/project/build.properties b/project/build.properties
@@ -1 +1 @@
-sbt.version=1.3.8
+sbt.version=1.3.9
diff --git a/project/plugins.sbt b/project/plugins.sbt
@@ -1,5 +1,5 @@
-addSbtPlugin("com.typesafe.sbt"          % "sbt-native-packager" % "1.6.1")
+addSbtPlugin("com.typesafe.sbt"          % "sbt-native-packager" % "1.6.2")
 addSbtPlugin("com.github.gseitz"         % "sbt-release"         % "1.0.13")
 addSbtPlugin("org.scalameta"             % "sbt-scalafmt"        % "2.3.2")
-addSbtPlugin("ch.epfl.scala"             % "sbt-scalafix"        % "0.9.11")
+addSbtPlugin("ch.epfl.scala"             % "sbt-scalafix"        % "0.9.13")
 addSbtPlugin("io.github.davidgregory084" % "sbt-tpolecat"        % "0.1.11")
diff --git a/src/main/scala/it/bitrock/dvs/producer/aviationedge/Main.scala b/src/main/scala/it/bitrock/dvs/producer/aviationedge/Main.scala
@@ -20,11 +20,11 @@ object Main extends App with LazyLogging {
 
   bindingFuture.map(serverBinding => logger.info(s"Exposing to ${serverBinding.localAddress}"))
 
-  val (cancellableFlight, flightCompletion, _, _)     = runAviationEdgeStream[FlightStream.type]()
-  val (cancellableAirplane, airplaneCompletion, _, _) = runAviationEdgeStream[AirplaneStream.type]()
-  val (cancellableAirport, airportCompletion, _, _)   = runAviationEdgeStream[AirportStream.type]()
-  val (cancellableAirline, airlineCompletion, _, _)   = runAviationEdgeStream[AirlineStream.type]()
-  val (cancellableCity, cityCompletion, _, _)         = runAviationEdgeStream[CityStream.type]()
+  val (cancellableFlight, flightCompletion)           = runAviationEdgeStream[FlightStream.type]()
+  val (cancellableAirplane, airplaneCompletion)       = runAviationEdgeStream[AirplaneStream.type]()
+  val (cancellableAirport, airportCompletion)         = runAviationEdgeStream[AirportStream.type]()
+  val (cancellableAirline, airlineCompletion)         = runAviationEdgeStream[AirlineStream.type]()
+  val (cancellableCity, cityCompletion)               = runAviationEdgeStream[CityStream.type]()
   val (cancellableFlightState, flightStateCompletion) = runOpenSkyStream[FlightStateStream.type]()
 
   val streamsCompletion =

diff --git a/src/main/scala/it/bitrock/dvs/producer/aviationedge/services/MainFunctions.scala b/src/main/scala/it/bitrock/dvs/producer/aviationedge/services/MainFunctions.scala
@@ -39,7 +39,7 @@ object MainFunctions {
   def runAviationEdgeStream[A: ApiProviderStreamContext]()(
       implicit system: ActorSystem,
       ec: ExecutionContext
-  ): (Cancellable, Future[Done], Future[Done], Future[Done]) = {
+  ): (Cancellable, Future[Done]) = {
     val config = AviationStreamContext[A].config(apiProviderConfig)
 
     val tickSource = new TickSource(config.pollingStart, config.pollingInterval, aviationConfig.tickSource).source
@@ -54,7 +54,7 @@ object MainFunctions {
 
     val jsonSource = tickSource.via(aviationFlow).via(monitoringGraph(monitoringSink)).mapConcat(identity)
 
-    mainGraph(jsonSource, rawSink, errorSink, invalidFlightSink).run()
+    mainGraph(jsonSource, rawSink, errorSink, invalidFlightSink).mapMaterializedValue(combineCompletionValue).run()
   }
 
   def runOpenSkyStream[A: ApiProviderStreamContext]()(
@@ -73,4 +73,9 @@ object MainFunctions {
 
     collectRightMessagesGraph(jsonSource, rawSink).run()
   }
+
+  private def combineCompletionValue[T](value: (Cancellable, Future[T], Future[T], Future[T]))(implicit ec: ExecutionContext) =
+    value match {
+      case (cancel, one, two, three) => (cancel, Future.firstCompletedOf(List(one, two, three)))
+    }
 }