Server streaming: getTemperature

Author: franciscodr

README.md

@@ -560,13 +560,13 @@ trait SmartHomeServiceApi[F[_]] {
 Whose interpretation could be:
 
 ```scala
-def getTemperature: Stream[F, TemperaturesSummary] = for {
-  client <- Stream.eval(clientF)
-  response <- client
-    .getTemperature(Empty)
-    .flatMap(t => Stream.eval(L.info(s"* Received new temperature: 👍  --> $t")).as(t))
-    .fold(TemperaturesSummary.empty)((summary, temperature) => summary.append(temperature))
-} yield response
+def getTemperature: Stream[F, TemperaturesSummary] = {
+  for {
+    client      <- Stream.eval(clientF)
+    temperature <- client.getTemperature(Empty)
+    _           <- Stream.eval(L.info(s"* Received new temperature: 👍 --> $temperature"))
+  } yield temperature
+}.fold(TemperaturesSummary.empty)((summary, temperature) => summary.append(temperature))
 ```
 
 Basically, we are logging the incoming values and at the end we calculate the average of those values.

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

@@ -17,6 +17,8 @@ class ClientProgram[F[_]: Effect: Logger] extends AppBoot[F] {
     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
 }
 

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

@@ -5,6 +5,8 @@ import cats.syntax.applicative._
 import cats.syntax.flatMap._
 import cats.syntax.functor._
 import com.fortyseven.protocol._
+import freestyle.rpc.protocol.Empty
+import fs2.Stream
 import io.chrisdavenport.log4cats.Logger
 import io.grpc.{CallOptions, ManagedChannel}
 import monix.execution.Scheduler
@@ -14,6 +16,8 @@ import scala.concurrent.duration._
 trait SmartHomeServiceApi[F[_]] {
 
   def isEmpty: F[Boolean]
+
+  def getTemperature: Stream[F, TemperaturesSummary]
 }
 
 object SmartHomeServiceApi {
@@ -27,6 +31,14 @@ object SmartHomeServiceApi {
         result    <- clientRPC.isEmpty(IsEmptyRequest())
         _         <- L.info(s"Result: $result")
       } yield result.result
+
+    def getTemperature: Stream[F, TemperaturesSummary] = {
+      for {
+        clientRPC   <- Stream.eval(clientRPCF)
+        temperature <- clientRPC.getTemperature(Empty)
+        _           <- Stream.eval(L.info(s"* Received new temperature: 👍 --> $temperature"))
+      } yield temperature
+    }.fold(TemperaturesSummary.empty)((summary, temperature) => summary.append(temperature))
   }
 
   def createInstance[F[_]: Effect](

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

@@ -1,9 +1,11 @@
 package com.fortyseven.protocol
 
-import freestyle.rpc.protocol.{service, Protobuf}
+import freestyle.rpc.protocol._
+import fs2.Stream
 
 @service(Protobuf) trait SmartHomeService[F[_]] {
 
   def isEmpty(request: IsEmptyRequest): F[IsEmptyResponse]
 
+  def getTemperature(empty: Empty.type): Stream[F, Temperature]
 }

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

@@ -1,14 +1,22 @@
 package com.fortyseven.server
 
-import cats.effect.Sync
+import cats.effect.{Async, Timer}
 import cats.syntax.functor._
-import com.fortyseven.protocol.{IsEmptyRequest, IsEmptyResponse, SmartHomeService}
+import com.fortyseven.protocol._
+import freestyle.rpc.protocol.Empty
+import fs2.Stream
 import io.chrisdavenport.log4cats.Logger
 
-class SmartHomeServiceHandler[F[_]: Sync: Logger] extends SmartHomeService[F] {
+class SmartHomeServiceHandler[F[_]: Async: Logger: Timer: TemperatureReader]
+    extends SmartHomeService[F] {
   val serviceName = "SmartHomeService"
 
   override def isEmpty(request: IsEmptyRequest): F[IsEmptyResponse] =
     Logger[F].info(s"$serviceName - Request: $request").as(IsEmptyResponse(true))
 
+  override def getTemperature(empty: Empty.type): Stream[F, Temperature] =
+    for {
+      _            <- Stream.eval(Logger[F].info(s"$serviceName - getTemperature Request"))
+      temperatures <- TemperatureReader[F].sendSamples.take(20)
+    } yield temperatures
 }