6 - RPC Client Streaming

Update README.md take n (#5) - Checking the last steps
Update README.md (#6) - small adjustments
Adds travis file and minor explanation about RFM (#7)

Author: juanpedromoreno

.travis.yml

@@ -0,0 +1,27 @@
+language: scala
+
+scala:
+- 2.12.4
+
+jdk:
+- oraclejdk8
+
+before_cache:
+- du -h -d 1 $HOME/.ivy2/
+- du -h -d 2 $HOME/.sbt/
+- find $HOME/.sbt -name "*.lock" -type f -delete
+- find $HOME/.ivy2/cache -name "ivydata-*.properties" -type f -delete
+
+cache:
+  directories:
+  - $HOME/.sbt/cache
+  - $HOME/.sbt/1.0
+  - $HOME/.sbt/boot/
+  - $HOME/.sbt/boot/scala*
+  - $HOME/.sbt/launchers
+  - $HOME/.ivy2/cache
+  - $HOME/.ivy2
+
+script:
+- sbt ++$TRAVIS_SCALA_VERSION "app/clean" "app/compile"
+- sbt ++$TRAVIS_SCALA_VERSION "server/clean" "server/compile"

README.md

@@ -18,10 +18,15 @@ object AppRFMClient extends Implicits {
     implicit val rfmClient: RFMAnalysisService.Client[IO] =
       RFMAnalysisService.client[IO](channel)
 
-    val (segments: IO[SegmentList], stream: Observable[UserEvent]) =
-      (rfmClient.segments(Empty), rfmClient.userEvents(Empty))
+    val (segments: IO[SegmentList], stream: Observable[UserEvent], ack: IO[Ack]) =
+      (
+        rfmClient.segments(Empty),
+        rfmClient.userEvents(Empty),
+        rfmClient.orderStream(ordersStreamObs)
+      )
 
     println(s"Segments: \n${segments.unsafeRunSync().list.mkString("\n")}\n")
+    println(s"Client Streaming: \n${ack.unsafeRunSync()}\n")
     Await.ready(
       stream
         .map { u =>
@@ -33,4 +38,25 @@ object AppRFMClient extends Implicits {
       Duration.Inf)
   }
 
+  private[this] def ordersStreamObs: Observable[Order] = {
+    val orderList: List[Order] = (1 to 1000).map { customerId =>
+      import com.fortysevendeg.scalacheck.datetime.GenDateTime
+      import org.joda.time.{DateTime, Period}
+      import org.scalacheck._
+      import com.fortysevendeg.scalacheck.datetime.instances.joda.jodaForPeriod
+
+      (for {
+        date    <- GenDateTime.genDateTimeWithinRange(DateTime.parse("2017-12-01"), Period.days(22))
+        orderId <- Gen.uuid
+        total   <- Gen.choose[Int](5, 200)
+      } yield
+        Order(
+          customerId,
+          CustomerData(date.toString, orderId.toString, total)
+        )).sample.get
+    }.toList
+
+    Observable.fromIterable(orderList)
+  }
+
 }

app/src/main/scala/AppRFMClient.scala

@@ -5,9 +5,11 @@ lazy val commonSettings: Seq[Def.Setting[_]] = Seq(
   addCompilerPlugin("org.scalameta" % "paradise" % "3.0.0-M10" cross CrossVersion.full),
   libraryDependencies ++= Seq(
     "io.frees" %% "frees-core" % freesV,
-    "io.frees"  %% "frees-async-cats-effect" % freesV,
+    "io.frees" %% "frees-async-cats-effect" % freesV,
     "io.frees" %% "frees-rpc" % "0.4.1",
-    "org.scalameta" %% "scalameta" % "1.8.0"),
+    "org.scalameta" %% "scalameta" % "1.8.0",
+    "joda-time" % "joda-time" % "2.9.9",
+    "com.github.tototoshi" %% "scala-csv" % "1.3.5"),
   scalacOptions ++= Seq("-Xplugin-require:macroparadise", "-Ywarn-unused-import"),
   scalacOptions in(Compile, console) ~= (_ filterNot (_ contains "paradise")) // macroparadise plugin doesn't work in repl yet.
 )
@@ -29,7 +31,6 @@ lazy val `data-generator` =
     .dependsOn(`functional-microservices`)
     .settings(
       libraryDependencies ++= Seq(
-        "joda-time" % "joda-time" % "2.9.9",
         "io.monix" %% "monix" % "3.0.0-M2",
         "org.scalacheck" %% "scalacheck" % "1.13.4",
         "com.47deg" %% "scalacheck-toolbox-datetime" % "0.2.3"

build.sbt

@@ -5,13 +5,15 @@ import cats.{~>, Applicative}
 import freestyle.rpc.server.handlers.GrpcServerHandler
 import freestyle.rpc.server._
 import freestyle.rpc.server.implicits._
-import scalaexchange.services.protocol.RFMAnalysisService
+import monix.eval.Task
 
+import scalaexchange.services.protocol.RFMAnalysisService
 import scalaexchange.services.runtime.RFMAnalysisServiceHandler
 
 trait Implicits extends scalaexchange.CommonImplicits {
 
-  implicit def rfmAnalisysServiceHandler[F[_]: Applicative]: RFMAnalysisServiceHandler[F] =
+  implicit def rfmAnalisysServiceHandler[F[_]: Applicative](
+      implicit T2F: Task ~> F): RFMAnalysisServiceHandler[F] =
     new RFMAnalysisServiceHandler[F]
 
   val grpcConfigs: List[GrpcConfig] = List(

server/src/main/scala/implicits.scala

@@ -18,6 +18,8 @@ object protocol {
 
   final case class SegmentList(list: List[Segment])
 
+  final case class Ack(result: String)
+
   @service
   trait RFMAnalysisService[F[_]] {
 
@@ -27,6 +29,10 @@ object protocol {
     @stream[ResponseStreaming.type]
     def userEvents(empty: Empty.type): F[Observable[UserEvent]]
 
+    @rpc(Avro)
+    @stream[RequestStreaming.type]
+    def orderStream(orders: Observable[Order]): F[Ack]
+
   }
 
 }

services/src/main/scala/protocol.scala

@@ -2,14 +2,20 @@ package scalaexchange
 package services
 package runtime
 
-import cats.Applicative
+import java.io.File
+
+import cats.{~>, Applicative}
+import com.github.tototoshi.csv._
 import freestyle.rpc.protocol._
+import monix.eval.Task
+import monix.execution.Scheduler
 import monix.reactive.Observable
 
 import scalaexchange.datagenerator.StreamingService
 import scalaexchange.services.protocol._
 
-class RFMAnalysisServiceHandler[F[_]: Applicative] extends RFMAnalysisService[F] {
+class RFMAnalysisServiceHandler[F[_]: Applicative](implicit S: Scheduler, T2F: Task ~> F)
+    extends RFMAnalysisService[F] {
 
   private[this] val segmentList: List[Segment] = List(
     Segment("Champions", 4, 5, 4, 5, 4, 5),
@@ -26,9 +32,27 @@ class RFMAnalysisServiceHandler[F[_]: Applicative] extends RFMAnalysisService[F]
 
   private[this] val streamingService = new StreamingService
 
+  private[this] val outPath: String = "orders.csv"
+
   override def segments(empty: Empty.type): F[protocol.SegmentList] =
     Applicative[F].pure(SegmentList(segmentList))
 
   override def userEvents(empty: Empty.type): F[Observable[UserEvent]] =
     Applicative[F].pure(streamingService.userEventsStream)
+
+  override def orderStream(orders: Observable[Order]): F[Ack] = T2F {
+    val f: File           = new File(outPath)
+    val writer: CSVWriter = CSVWriter.open(f)
+
+    orders
+      .foreachL { order =>
+        writer.writeRow(
+          List(order.data.date, order.data.orderId, order.customerId, order.data.total)
+        )
+      }
+      .map { _ =>
+        writer.close()
+        Ack(" 👍 ")
+      }
+  }
 }

services/src/main/scala/runtime/RFMAnalysisServiceHandler.scala

@@ -6,3 +6,7 @@ case object UnprocessedCheckout extends EventType
 case object Login               extends EventType
 
 case class UserEvent(userId: Int, eventType: EventType, date: String)
+
+case class CustomerData(date: String, orderId: String, total: Int)
+
+case class Order(customerId: Int, data: CustomerData)