class: center, middle, inverse # Fake it till you make it with FS2 James Collier @JRCCollier  --- ## Playback Session Telemetry events will be emitted from devices. ```scala import java.time.Instant import java.util.UUID sealed trait PlaybackEvent { def sessionId: UUID } final case class Started(timestamp: Instant, sessionId: UUID) extends PlaybackEvent final case class Paused(timestamp: Instant, sessionId: UUID) extends PlaybackEvent final case class Resumed(timestamp: Instant, sessionId: UUID) extends PlaybackEvent final case class Ended(timestamp: Instant, sessionId: UUID) extends PlaybackEvent ``` --- ## 1000ft view  --- class: inverse ## The catch 😱 it's going to be a while till devices are instrumented! -- The show must go on! -- Need to develop while application teams are instrumenting. --- ## Fake it till we instrument it We have the ~~technology~~ schema, we can build the events. --  --- ## Let's just use scalacheck! ```scala import org.scalacheck.ScalacheckShapeless._ import org.scalacheck.{Arbitrary, Gen} private implicit val arbInstant: Arbitrary[Instant] = Arbitrary(Gen.calendar.map(_.toInstant)) val arbPlaybackEvt = implicitly[Arbitrary[PlaybackEvent]] ``` Now let's generate some events... -- ```scala List.fill(10)(arbPlaybackEvt.arbitrary.sample).flatten // res0: List[PlaybackEvent] = List( // Ended(0000-12-30T05:00:00Z, f7496dc1-2d02-4c40-84a7-4ad374293d84), // Ended(+292269052-08-23T02:23:20.516Z, 54112d79-bf5a-44de-89cb-18ab6fd855e5), // Paused(1579-11-08T22:01:34.366Z, 9a18624f-92d4-4187-895b-dd6250e32e10), // Started(2687-02-25T17:27:35.937Z, 80931d28-871c-4c2c-bf87-7ae3fd2fba3f), // Started(0004-11-01T20:17:29.116Z, 45629f39-4678-491c-a7c5-d1367cf04d08), // Started(0000-12-30T00:00:00Z, d3275765-dde6-4dc9-91b5-097b7c8812bf), // Ended(+57502701-01-01T00:00:00.001Z, 1345fd86-4cb4-4a94-8b55-98404bc36d2f), // Ended(3354-09-07T08:07:20Z, 84d05966-bf03-4ab4-84d9-173283261b45), // Started(1970-01-01T00:00:00.001Z, 66e3a7d3-7cc8-4873-91c6-d2990d6ba87b), // Paused(0991-03-22T23:59:59.059Z, 56eb25ac-ad0a-4b0f-8b30-c17e6a9147c5) // ) ``` 😱 not useful! ??? * timestamp is nonsense. * each event has it's own session. * not a representative lifecycle. --- ## Requirements * Sessions modelled. * Representative delays between events. * Configure the number of concurrent sessions. -- Maybe this can be modelled as streams? -- Step in FS2:  ```scala import cats.effect._ import fs2.Stream import scala.concurrent.ExecutionContext import scala.concurrent.duration._ ``` --- ## Model a playback scenario ```scala def session: Stream[IO, PlaybackEvent] = for { now <- Stream.emit(Instant.parse("2019-11-26T12:00:00Z")) sessionId <- Stream.eval(IO(UUID.randomUUID())) s <- Stream.emits(List( Started(now, sessionId), Ended(now.plusSeconds(60), sessionId) )) } yield s ``` -- ```scala session.compile.toList.unsafeRunSync() // res2: List[PlaybackEvent] = List( // Started(2019-11-26T12:00:00Z, 62480eba-10e9-4a0a-a598-89e2fb374683), // Ended(2019-11-26T12:01:00Z, 62480eba-10e9-4a0a-a598-89e2fb374683) // ) ``` --- ## Nap between events 💤💤💤 Useful to model the progress of time between emitting events for a session. -- ```scala def nowS: Stream[IO, Instant] = Stream.eval(timer.clock.realTime(MILLISECONDS).map(Instant.ofEpochMilli)) def idealSession: Stream[IO, PlaybackEvent] = for { sessionId <- Stream.eval(IO(UUID.randomUUID())) s <- nowS.map(Started(_, sessionId)) ++ nowS.map(Ended(_, sessionId)).delayBy(30.minutes) } yield s ``` -- ```scala idealSession .evalMap(event => IO(println(event))) .compile .drain .unsafeRunAsyncAndForget() // Started(2018-12-20T00:00:00Z,d81d0a26-5608-41a4-8f5b-95e1c0ca55e3) testContext.tick(1.hour) // Ended(2018-12-20T00:30:00Z,d81d0a26-5608-41a4-8f5b-95e1c0ca55e3) ``` --- ## A stream of sessions  ??? Par join to choose how many concurrent sessions to have. --- ## Concurrent sessions ```scala val numConcurrent = 3 Stream.emit(idealSession) .repeat .take(5) // limit to 5 for example .parJoin(numConcurrent) .evalMap(event => IO(println(event))) .compile .drain .unsafeRunAsyncAndForget() ``` -- ```scala testContext.tick(1.hour) // Started(2018-12-20T01:00:00Z,55a0103d-4085-4820-b9e7-e008f4ba18c1) // Started(2018-12-20T01:00:00Z,298f0a5d-b88d-4d49-894a-f631bbc6afe4) // Started(2018-12-20T01:00:00Z,e0b57cd6-6925-4960-b17a-785801f109df) // Ended(2018-12-20T01:30:00Z,55a0103d-4085-4820-b9e7-e008f4ba18c1) // Ended(2018-12-20T01:30:00Z,298f0a5d-b88d-4d49-894a-f631bbc6afe4) // Ended(2018-12-20T01:30:00Z,e0b57cd6-6925-4960-b17a-785801f109df) // Started(2018-12-20T01:30:00Z,fdc11130-4cdc-438d-8b06-5e2a936df234) // Started(2018-12-20T01:30:00Z,04a41122-e9aa-4742-9e57-047becdbd2ed) // Ended(2018-12-20T02:00:00Z,04a41122-e9aa-4742-9e57-047becdbd2ed) // Ended(2018-12-20T02:00:00Z,fdc11130-4cdc-438d-8b06-5e2a936df234) ``` ??? Sessions are repeated forever. --- ## Add another scenario! ```scala def pauseSession: Stream[IO, PlaybackEvent] = for { sessionId <- Stream.eval(IO(UUID.randomUUID())) s <- nowS.map(Started(_, sessionId)) ++ nowS.map(Paused(_, sessionId)).delayBy(20.seconds) ++ nowS.map(Resumed(_, sessionId)).delayBy(5.seconds) ++ nowS.map(Ended(_, sessionId)).delayBy(20.minutes) } yield s ``` --- ## Running multiple scenarios ```scala Stream.emits(List(idealSession, pauseSession)) .repeat .take(4) // limit to 4 for example .parJoin(numConcurrent) .evalMap(event => IO(println(event))) .compile .drain .unsafeRunAsyncAndForget() ``` -- ```scala testContext.tick(1.hour) // Started(2018-12-20T02:00:00Z,fc2578f4-cc06-48ae-82d2-1ffbebe0941d) // Started(2018-12-20T02:00:00Z,aa0d1a91-b262-40b1-bf83-58c23cd6a690) // Started(2018-12-20T02:00:00Z,70500817-8eb2-4fdb-8155-664da3d068e7) // Paused(2018-12-20T02:00:20Z,aa0d1a91-b262-40b1-bf83-58c23cd6a690) // Resumed(2018-12-20T02:00:25Z,aa0d1a91-b262-40b1-bf83-58c23cd6a690) // Ended(2018-12-20T02:20:25Z,aa0d1a91-b262-40b1-bf83-58c23cd6a690) // Started(2018-12-20T02:20:25Z,57da679d-f630-4a37-9a41-78f4a9181a8c) // Paused(2018-12-20T02:20:45Z,57da679d-f630-4a37-9a41-78f4a9181a8c) // Resumed(2018-12-20T02:20:50Z,57da679d-f630-4a37-9a41-78f4a9181a8c) // Ended(2018-12-20T02:30:00Z,fc2578f4-cc06-48ae-82d2-1ffbebe0941d) // Ended(2018-12-20T02:30:00Z,70500817-8eb2-4fdb-8155-664da3d068e7) // Ended(2018-12-20T02:40:50Z,57da679d-f630-4a37-9a41-78f4a9181a8c) ``` --- ## Testing Can use `cats.effect.laws.util.TestContext` to simulate time. ```scala val f = idealSession .compile .toList .unsafeToFuture() ``` -- ```scala f.value // res9: Option[scala.util.Try[List[PlaybackEvent]]] = None ``` -- ```scala testContext.tick(1.hour) f.value // res11: Option[scala.util.Try[List[PlaybackEvent]]] = Some( // Success( // List( // Started(2018-12-20T03:00:00Z, 4a91bf8e-90c0-47d4-8f2b-624273408081), // Ended(2018-12-20T03:30:00Z, 4a91bf8e-90c0-47d4-8f2b-624273408081) // ) // ) // ) ``` --- ## Counting sessions ```scala val f2 = Stream.emit(idealSession) .repeat .parJoin(1000) .interruptAfter(1.hour) .compile .toList // count sessions .map(_.groupBy(_.sessionId).keySet.size) .unsafeToFuture() ``` `idealSession` lasts 30 minutes, so after 1 hour there should be ~2000 sessions. -- ```scala testContext.tick(1.hour) f2.value // res13: Option[scala.util.Try[Int]] = Some(Success(2000)) ``` 👌 --- ## Further reading and links * [FS2](https://github.com/functional-streams-for-scala/fs2) by [@mpilquist](https://github.com/mpilquist) * [cats-effect](https://github.com/typelevel/cats-effect) ## Thanks! * [mdoc](https://github.com/scalameta/mdoc) used to evaluate scala examples ## Questions?