Bidirectional streaming: comingBackMode

Author: franciscodr

client/src/main/scala/client/ClientApp.scala

@@ -3,6 +3,7 @@ package com.fortyseven.client
 import cats.effect.{Effect, IO, Timer}
 import com.fortyseven.commons._
 import com.fortyseven.commons.config.ServiceConfig
+import com.fortyseven.protocol.implicits._
 import fs2.{Stream, StreamApp}
 import io.chrisdavenport.log4cats.Logger
 import monix.execution.Scheduler
@@ -13,13 +14,15 @@ class ClientProgram[F[_]: Effect: Logger] extends AppBoot[F] {
 
   implicit val TM: Timer[F] = Timer.derive[F](Effect[F], IO.timer(S))
 
-  override def appStream(config: ServiceConfig): fs2.Stream[F, StreamApp.ExitCode] =
+  override def appStream(config: ServiceConfig): fs2.Stream[F, StreamApp.ExitCode] = {
     for {
       serviceApi <- SmartHomeServiceApi.createInstance(config.host.value, config.port.value)
       _          <- Stream.eval(serviceApi.isEmpty)
       summary    <- serviceApi.getTemperature
       _          <- Stream.eval(Logger[F].info(s"The average temperature is: ${summary.averageTemperature}"))
-    } yield StreamApp.ExitCode.Success
+      response   <- serviceApi.comingBackMode(LocationsGenerator.get[F])
+    } yield response.actions
+  }.to(LogSink[F].showLines).drain.as(StreamApp.ExitCode.Success)
 }
 
 object ClientApp extends ClientProgram[IO]

client/src/main/scala/client/SmartHomeServiceApi.scala

@@ -18,6 +18,8 @@ trait SmartHomeServiceApi[F[_]] {
   def isEmpty: F[Boolean]
 
   def getTemperature: Stream[F, TemperaturesSummary]
+
+  def comingBackMode(locations: Stream[F, Location]): Stream[F, ComingBackModeResponse]
 }
 
 object SmartHomeServiceApi {
@@ -39,6 +41,12 @@ object SmartHomeServiceApi {
         _           <- Stream.eval(L.info(s"* Received new temperature: 👍 --> $temperature"))
       } yield temperature
     }.fold(TemperaturesSummary.empty)((summary, temperature) => summary.append(temperature))
+
+    def comingBackMode(locations: Stream[F, Location]): Stream[F, ComingBackModeResponse] =
+      for {
+        clientRPC <- Stream.eval(clientRPCF)
+        response  <- clientRPC.comingBackMode(locations)
+      } yield response
   }
 
   def createInstance[F[_]: Effect](

protocol/src/main/scala/protocol/Messages.scala

@@ -17,3 +17,6 @@ final case class IsEmptyRequest()
 
 @message
 final case class IsEmptyResponse(result: Boolean)
+
+@message
+final case class ComingBackModeResponse(actions: List[SmartHomeAction])

protocol/src/main/scala/protocol/SmartHomeService.scala

@@ -8,4 +8,6 @@ import fs2.Stream
   def isEmpty(request: IsEmptyRequest): F[IsEmptyResponse]
 
   def getTemperature(empty: Empty.type): Stream[F, Temperature]
+
+  def comingBackMode(request: Stream[F, Location]): Stream[F, ComingBackModeResponse]
 }

server/src/main/scala/server/SmartHomeServiceHandler.scala

@@ -7,7 +7,7 @@ import freestyle.rpc.protocol.Empty
 import fs2.Stream
 import io.chrisdavenport.log4cats.Logger
 
-class SmartHomeServiceHandler[F[_]: Async: Logger: Timer: TemperatureReader]
+class SmartHomeServiceHandler[F[_]: Async: Logger: Timer: TemperatureReader: SmartHomeSupervisor]
     extends SmartHomeService[F] {
   val serviceName = "SmartHomeService"
 
@@ -19,4 +19,19 @@ class SmartHomeServiceHandler[F[_]: Async: Logger: Timer: TemperatureReader]
       _            <- Stream.eval(Logger[F].info(s"$serviceName - getTemperature Request"))
       temperatures <- TemperatureReader[F].sendSamples.take(20)
     } yield temperatures
+
+  override def comingBackMode(request: Stream[F, Location]): Stream[F, ComingBackModeResponse] =
+    for {
+      _        <- Stream.eval(Logger[F].info(s"$serviceName - Enabling Coming Back Home mode"))
+      location <- request
+      _ <- Stream.eval(
+        if (location.distanceToDestination > 0.0d)
+          Logger[F]
+            .info(s"$serviceName - Distance to destination: ${location.distanceToDestination} mi")
+        else
+          Logger[F]
+            .info(s"$serviceName - You have reached your destination 🏡"))
+      response <- Stream.eval(
+        SmartHomeSupervisor[F].performAction(location).map(ComingBackModeResponse))
+    } yield response
 }