Skip to content

Commit 3a97c8e

Browse files
franciscodrfranciscodr
committed
Data generator for temperatures
1 parent 7f6b8d5 commit 3a97c8e

File tree

4 files changed

+261
-0
lines changed

4 files changed

+261
-0
lines changed

‎README.md

+184
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,11 @@
1717
- [Server](#server-1)
1818
- [Client](#client-1)
1919
- [Result](#result)
20+
- [Server-streaming RPC service: `GetTemperature`](#server-streaming-rpc-service-gettemperature)
21+
- [Protocol](#protocol-3)
22+
- [Server](#server-2)
23+
- [Client](#client-2)
24+
- [Result](#result-1)
2025

2126
<!-- END doctoc generated TOC please keep comment here to allow auto update -->
2227

@@ -457,6 +462,185 @@ And the server log the request as expected:
457462
INFO - SmartHomeService - Request: IsEmptyRequest()
458463
```
459464

465+
466+
## Server-streaming RPC service: `GetTemperature`
467+
468+
Following the established plan, the next step is building the service that returns a stream of temperature values, to let clients subscribe to collect real-time info.
469+
470+
### Protocol
471+
472+
As usual we should add this operation in the protocol.
473+
474+
**_Messages.scala_**
475+
476+
Adding new models:
477+
478+
```scala
479+
case class TemperatureUnit(value: String) extends AnyVal
480+
case class Temperature(value: Double, unit: TemperatureUnit)
481+
```
482+
483+
**_SmartHomeService.scala_**
484+
485+
And the `getTemperature` operation:
486+
487+
```scala
488+
@service(Protobuf) trait SmartHomeService[F[_]] {
489+
490+
def isEmpty(request: IsEmptyRequest): F[IsEmptyResponse]
491+
492+
def getTemperature(empty: Empty.type): Stream[F, Temperature]
493+
}
494+
```
495+
496+
### Server
497+
498+
If we want to emit a stream of `Temperature` values, we would be well advised to develop a producer of `Temperature` in the server side. For instance:
499+
500+
```scala
501+
trait TemperatureReader[F[_]] {
502+
def sendSamples: Stream[F, Temperature]
503+
}
504+
505+
object TemperatureReader {
506+
implicit def instance[F[_]: Sync: Logger: Timer]: TemperatureReader[F] =
507+
new TemperatureReader[F] {
508+
val seed = Temperature(77d, TemperatureUnit("Fahrenheit"))
509+
510+
def readTemperature(current: Temperature): F[Temperature] =
511+
Timer[F]
512+
.sleep(1.second)
513+
.flatMap(_ =>
514+
Sync[F].delay {
515+
val increment: Double = Random.nextDouble() / 2d
516+
val signal = if (Random.nextBoolean()) 1 else -1
517+
val currentValue = current.value
518+
519+
current.copy(
520+
value = BigDecimal(currentValue + (signal * increment))
521+
.setScale(2, RoundingMode.HALF_UP)
522+
.doubleValue)
523+
})
524+
525+
override def sendSamples: Stream[F, Temperature] =
526+
Stream.iterateEval(seed) { t =>
527+
Logger[F].info(s"* New Temperature 👍 --> $t").flatMap(_ => readTemperature(t))
528+
}
529+
}
530+
531+
def apply[F[_]](implicit ev: TemperatureReader[F]): TemperatureReader[F] = ev
532+
}
533+
```
534+
535+
And this can be returned as response of the new service, in the interpreter.
536+
537+
```scala
538+
override def getTemperature(empty: Empty.type): Stream[F, Temperature] = for {
539+
_ <- Stream.eval(Logger[F].info(s"$serviceName - getTemperature Request"))
540+
temperatures <- TemperatureReader[F].sendSamples.take(20)
541+
} yield temperatures
542+
```
543+
544+
### Client
545+
546+
We have nothing less than adapt the client to consume the new service when it starting up. To this, a couple of changes are needed:
547+
548+
Firstly we should enrich the algebra
549+
550+
```scala
551+
trait SmartHomeServiceApi[F[_]] {
552+
553+
def isEmpty(): F[Boolean]
554+
555+
def getTemperature(): Stream[F, Temperature]
556+
557+
}
558+
```
559+
560+
Whose interpretation could be:
561+
562+
```scala
563+
def getTemperature: Stream[F, TemperaturesSummary] = for {
564+
client <- Stream.eval(clientF)
565+
response <- client
566+
.getTemperature(Empty)
567+
.flatMap(t => Stream.eval(L.info(s"* Received new temperature: 👍 --> $t")).as(t))
568+
.fold(TemperaturesSummary.empty)((summary, temperature) => summary.append(temperature))
569+
} yield response
570+
```
571+
572+
Basically, we are logging the incoming values and at the end we calculate the average of those values.
573+
574+
Now, the client app calls to both services: `isEmpty` and `getTemperature`.
575+
And finally, to call it:
576+
577+
```scala
578+
for {
579+
serviceApi <- SmartHomeServiceApi.createInstance(config.host.value, config.port.value)
580+
_ <- Stream.eval(serviceApi.isEmpty)
581+
summary <- serviceApi.getTemperature
582+
_ <- Stream.eval(Logger[F].info(s"The average temperature is: ${summary.averageTemperature}"))
583+
} yield StreamApp.ExitCode.Success
584+
```
585+
586+
### Result
587+
588+
When we run the client now with `sbt runClient` we get:
589+
590+
```bash
591+
INFO - Created new RPC client for (localhost,19683)
592+
INFO - Result: IsEmptyResponse(true)
593+
INFO - * Received new temperature: 👍 --> Temperature(77.0,TemperatureUnit(Fahrenheit))
594+
INFO - * Received new temperature: 👍 --> Temperature(77.25,TemperatureUnit(Fahrenheit))
595+
INFO - * Received new temperature: 👍 --> Temperature(77.58,TemperatureUnit(Fahrenheit))
596+
INFO - * Received new temperature: 👍 --> Temperature(78.02,TemperatureUnit(Fahrenheit))
597+
INFO - * Received new temperature: 👍 --> Temperature(77.67,TemperatureUnit(Fahrenheit))
598+
INFO - * Received new temperature: 👍 --> Temperature(77.5,TemperatureUnit(Fahrenheit))
599+
INFO - * Received new temperature: 👍 --> Temperature(77.58,TemperatureUnit(Fahrenheit))
600+
INFO - * Received new temperature: 👍 --> Temperature(77.15,TemperatureUnit(Fahrenheit))
601+
INFO - * Received new temperature: 👍 --> Temperature(76.66,TemperatureUnit(Fahrenheit))
602+
INFO - * Received new temperature: 👍 --> Temperature(76.45,TemperatureUnit(Fahrenheit))
603+
INFO - * Received new temperature: 👍 --> Temperature(76.77,TemperatureUnit(Fahrenheit))
604+
INFO - * Received new temperature: 👍 --> Temperature(76.74,TemperatureUnit(Fahrenheit))
605+
INFO - * Received new temperature: 👍 --> Temperature(76.41,TemperatureUnit(Fahrenheit))
606+
INFO - * Received new temperature: 👍 --> Temperature(76.59,TemperatureUnit(Fahrenheit))
607+
INFO - * Received new temperature: 👍 --> Temperature(76.77,TemperatureUnit(Fahrenheit))
608+
INFO - * Received new temperature: 👍 --> Temperature(76.49,TemperatureUnit(Fahrenheit))
609+
INFO - * Received new temperature: 👍 --> Temperature(76.04,TemperatureUnit(Fahrenheit))
610+
INFO - * Received new temperature: 👍 --> Temperature(76.42,TemperatureUnit(Fahrenheit))
611+
INFO - * Received new temperature: 👍 --> Temperature(75.95,TemperatureUnit(Fahrenheit))
612+
INFO - * Received new temperature: 👍 --> Temperature(75.97,TemperatureUnit(Fahrenheit))
613+
INFO - The average temperature is: Temperature(76.85,TemperatureUnit(Fahrenheit))
614+
INFO - Removed 1 RPC clients from cache.
615+
```
616+
617+
And the server log the request as expected:
618+
619+
```bash
620+
INFO - ServiceName(seedServer) - Starting app.server at Host(localhost):Port(19683)
621+
INFO - SmartHomeService - Request: IsEmptyRequest()
622+
INFO - SmartHomeService - getTemperature Request
623+
INFO - * New Temperature 👍 --> Temperature(77.0,TemperatureUnit(Fahrenheit))
624+
INFO - * New Temperature 👍 --> Temperature(77.25,TemperatureUnit(Fahrenheit))
625+
INFO - * New Temperature 👍 --> Temperature(77.58,TemperatureUnit(Fahrenheit))
626+
INFO - * New Temperature 👍 --> Temperature(78.02,TemperatureUnit(Fahrenheit))
627+
INFO - * New Temperature 👍 --> Temperature(77.67,TemperatureUnit(Fahrenheit))
628+
INFO - * New Temperature 👍 --> Temperature(77.5,TemperatureUnit(Fahrenheit))
629+
INFO - * New Temperature 👍 --> Temperature(77.58,TemperatureUnit(Fahrenheit))
630+
INFO - * New Temperature 👍 --> Temperature(77.15,TemperatureUnit(Fahrenheit))
631+
INFO - * New Temperature 👍 --> Temperature(76.66,TemperatureUnit(Fahrenheit))
632+
INFO - * New Temperature 👍 --> Temperature(76.45,TemperatureUnit(Fahrenheit))
633+
INFO - * New Temperature 👍 --> Temperature(76.77,TemperatureUnit(Fahrenheit))
634+
INFO - * New Temperature 👍 --> Temperature(76.74,TemperatureUnit(Fahrenheit))
635+
INFO - * New Temperature 👍 --> Temperature(76.41,TemperatureUnit(Fahrenheit))
636+
INFO - * New Temperature 👍 --> Temperature(76.59,TemperatureUnit(Fahrenheit))
637+
INFO - * New Temperature 👍 --> Temperature(76.77,TemperatureUnit(Fahrenheit))
638+
INFO - * New Temperature 👍 --> Temperature(76.49,TemperatureUnit(Fahrenheit))
639+
INFO - * New Temperature 👍 --> Temperature(76.04,TemperatureUnit(Fahrenheit))
640+
INFO - * New Temperature 👍 --> Temperature(76.42,TemperatureUnit(Fahrenheit))
641+
INFO - * New Temperature 👍 --> Temperature(75.95,TemperatureUnit(Fahrenheit))
642+
```
643+
460644
<!-- DOCTOC SKIP -->
461645
# Copyright
462646

‎client/src/main/scala/client/TemperaturesSummary.scala

+29
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
package com.fortyseven.client
2+
3+
import com.fortyseven.protocol.{Temperature, TemperatureUnit}
4+
5+
import scala.math.BigDecimal.RoundingMode
6+
7+
case class TemperaturesSummary(
8+
temperatures: Vector[Temperature],
9+
averageTemperature: Temperature
10+
) {
11+
def append(newTemperature: Temperature): TemperaturesSummary = {
12+
val temp = this.temperatures :+ newTemperature
13+
this.copy(
14+
temp,
15+
Temperature(
16+
BigDecimal(
17+
this.averageTemperature.value + (newTemperature.value - this.averageTemperature.value) / (this.temperatures.length + 1))
18+
.setScale(2, RoundingMode.HALF_UP)
19+
.toDouble,
20+
this.averageTemperature.unit
21+
)
22+
)
23+
}
24+
}
25+
26+
object TemperaturesSummary {
27+
val empty: TemperaturesSummary =
28+
TemperaturesSummary(Vector.empty, Temperature(0d, TemperatureUnit("Fahrenheit")))
29+
}

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

+4
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,10 @@ package com.fortyseven.protocol
22

33
import freestyle.rpc.protocol.message
44

5+
case class TemperatureUnit(value: String) extends AnyVal
6+
7+
case class Temperature(value: Double, unit: TemperatureUnit)
8+
59
@message
610
final case class IsEmptyRequest()
711

‎server/src/main/scala/server/TemperatureReader.scala

+44
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
package com.fortyseven.server
2+
3+
import cats.effect._
4+
import cats.syntax.flatMap._
5+
import com.fortyseven.protocol.{Temperature, TemperatureUnit}
6+
import fs2.Stream
7+
import io.chrisdavenport.log4cats.Logger
8+
9+
import scala.concurrent.duration._
10+
import scala.math.BigDecimal.RoundingMode
11+
import scala.util.Random
12+
13+
trait TemperatureReader[F[_]] {
14+
def sendSamples: Stream[F, Temperature]
15+
}
16+
17+
object TemperatureReader {
18+
implicit def instance[F[_]: Sync: Logger: Timer]: TemperatureReader[F] =
19+
new TemperatureReader[F] {
20+
val seed = Temperature(77d, TemperatureUnit("Fahrenheit"))
21+
22+
def readTemperature(current: Temperature): F[Temperature] =
23+
Timer[F]
24+
.sleep(1.second)
25+
.flatMap(_ =>
26+
Sync[F].delay {
27+
val increment: Double = Random.nextDouble() / 2d
28+
val signal = if (Random.nextBoolean()) 1 else -1
29+
val currentValue = current.value
30+
31+
current.copy(
32+
value = BigDecimal(currentValue + (signal * increment))
33+
.setScale(2, RoundingMode.HALF_UP)
34+
.doubleValue)
35+
})
36+
37+
override def sendSamples: Stream[F, Temperature] =
38+
Stream.iterateEval(seed) { t =>
39+
Logger[F].info(s"* New Temperature 👍 --> $t").flatMap(_ => readTemperature(t))
40+
}
41+
}
42+
43+
def apply[F[_]](implicit ev: TemperatureReader[F]): TemperatureReader[F] = ev
44+
}

0 commit comments

Comments
 (0)