Upgrade to Pro — share decks privately, control downloads, hide ads and more …

実践Monix導入 / Monix in Action

Taisuke Oe
October 27, 2019

実践Monix導入 / Monix in Action

2019/10/27 Scala関西サミット
2019/9/30 セプテーニ・オリジナル社内勉強会

Taisuke Oe

October 27, 2019
Tweet

More Decks by Taisuke Oe

Other Decks in Technology

Transcript

  1. WHO AM I ▸ Taisuke Oe ▸ ϑϦʔϥϯεΤϯδχΞ ▸ ٕज़ΞυόΠβʔ

    @ Septeni Original, Inc ▸ ScalaMatsuri chairperson
  2. MONIXͬͯͳʹʁ MONIX ▸ ฒߦϓϩάϥϛϯάΛޮ཰Α͘ɺศརʹ͢ΔͨΊͷϥΠϒ ϥϦ ▸ ࠷৽͸v3.0.0 ▸ cats, cats-effectʹґଘ

    ▸ ࡞ऀ͸cats-effectͷओͨΔ։ൃऀͷҰਓ(ίϛοτ਺1Ґ) ▸ Scala Best Practices
  3. MONIXͬͯͳʹʁ MONIXͷྺ࢙(͓·͚) ▸ ~v1.0 (2015/12): monifu ▸ Reactive Extensions (Rx)ʹΠϯεύΠΞ͞Εͨɺobservable

    sequenceʹΑΔඇಉظϓϩάϥϛϯά ϥΠϒϥϦ ▸ Zero dependency / compatible with Scalaz ▸ v1.0 monifu => monix΁໊લมߋ ▸ ‘Monifu has a lame English pronunciation’ (ӳޠͰ޷·͘͠ͳ͍ൃԻͷͨΊ) ▸ v2.0-M1 (2016/3): Taskͷ௥Ճ ▸ v2.0.0 release (2016/9) ▸ Optional packages: moni-cats / monix-scalaz-72 ▸ v3.0.0-M1 (2017/8) Iterantͷ௥Ճ ▸ deep integration with Typelevel Cats and Cats-Effect ▸ v2.3.3 release (2018/1) ▸ v3.0.0 release (2019/9) <- ΠϚίί
  4. MONIXͬͯͳʹʁ MONIXͷߏ੒ ▸ monix-execution … ฒߦॲཧͰ෭࡞༻Λ͏·͘ѻ͏ͨΊͷLow Level APIɻ SchedulerͳͲɻ ▸

    monix-eval … ฒߦॲཧͰ෭࡞༻Λ͏·͘ѻ͏ͨΊͷɺ७ਮؔ਺ܕσʔλߏ ଄ɻTask, CoevalͳͲɻ ▸ monix-reactive … ReactiveXͷScala࣮૷ɻObservableͳͲɻ ▸ monix-tail … PullܕετϦʔϜͷ७ਮؔ਺ܕσʔλߏ଄ɻIterantͳͲɻ ▸ monix-catnap … cats-effectͷܕΫϥεΛجʹͨ͠७ਮؔ਺ܕͷฒߦॲཧ ϢʔςΟϦςΟɻMVar, CircuitBreakerͳͲɻ
  5. MONIXͬͯͳʹʁ MONIXͷߏ੒ ▸ monix-execution … ฒߦॲཧͰ෭࡞༻Λ͏·͘ѻ͏ͨΊͷLow Level APIɻSchedulerͳͲɻ ▸ monix-eval

    … ฒߦॲཧͰ෭࡞༻Λ͏·͘ѻ͏ͨΊͷɺ७ਮؔ਺ܕσʔλߏ ଄ɻTask, CoevalͳͲɻ ▸ monix-reactive … ReactiveXͷScala࣮૷ɻObservableͳͲɻ ▸ monix-tail … PullܕετϦʔϜͷ७ਮؔ਺ܕσʔλߏ଄ɻIterantͳͲɻ ▸ monix-catnap … cats-effectͷܕΫϥεΛجʹͨ͠७ਮؔ਺ܕͷฒߦॲཧ ϢʔςΟϦςΟɻMVar, CircuitBreakerͳͲɻ
  6. MONIX TASKͷ࢖͍ํ MONIX TASKʹΑΔฒߦϓϩάϥϛϯάୈҰา import Scheduler.Implicits.global def task(i: Int) =

    Task.eval { printCurrentThread() i } val cubeSum = for { i <- task(11) //on run-main-0 iCube <- Task.evalAsync(i * i * i) j <- task(13) //on scala-execution-context-global-57 jCube <- Task.evalAsync(j * j * j) } yield { printCurrentThread() //on scala-execution-context-global-57 iCube + jCube } cubeSum.runAsync { case Right(value) => println(value) case Left(err) => println(err.getMessage) }
  7. MONIX TASKͷධՁઓུ MONIX TASK͸஗ԆධՁ ▸ ஗ԆධՁTaskΠϯελϯεΛੜ੒ͨ࣌͠఺Ͱ ͸ɺ࣮ߦ͞Εͳ͍ͷͰࢀরಁաɻ ▸ ஗ԆධՁ͔ͭಉظ ▸

    Task.apply ▸ v3.0.0͔Βಉظʹͳͬͨ ▸ Task.eval ▸ Task.evalOnce ▸ ϝϞԽ ▸ ஗ԆධՁ͔ͭඇಉظ ▸ Task.evalAsync ▸ ઌߦධՁ͔ͭಉظ ▸ Task.now ▸ Task.raiseError ▸ ܭࢉࡁΈͷ஋Λϥοϓ͢Δ
  8. MONIX TASKͷධՁઓུ ஗ԆධՁTASKΛ࣮ߦ͢Δ ▸ ࣮ߦ͢Δࡍʹ͸͡ΊͯɺScheduler (ExecutionContextΛϥοϓͨ͠΋ͷʣΛ҉໧ͷҾ਺Ͱ ཁٻ͢Δ ▸ ࣮ߦ͢Δͱɺ෭࡞༻͕ൃੜ͢ΔʢՄೳੑ͕͋Δʣ ▸

    Task#runAsync ▸ Task#runAsync(cb: (Either[Throwable, A]) => Unit)(implicit s: Scheduler): Cancelable ▸ Task#runAsyncUncancelable ▸ Task#runAsyncUncancelable(cb: (Either[Throwable, A]) => Unit)(implicit s: Scheduler): Unit ▸ Task#runAsyncAndForget ▸ Task#runAsyncAndForget(implicit s: Scheduler): Unit ▸ Task#runToFuture ▸ Task#runToFuture(implicit s: Scheduler): CancelableFuture[A]
  9. MONIX TASKͷ࢖͍ํ MONIX TASKʹ͓͚ΔඇಉظڥքΛཧղ͠Α͏ import Scheduler.Implicits.global def task(i: Int) =

    Task.eval { printCurrentThread() i } val cubeSum = for { i <- task(11) //on run-main-0 iCube <- Task.evalAsync(i * i * i) j <- task(13) //on scala-execution-context-global-57 jCube <- Task.evalAsync(j * j * j) } yield { printCurrentThread() //on scala-execution-context-global-57 iCube + jCube } cubeSum.runAsync{ case Right(value) => println(value) case Left(err) => println(err.getMessage) }
  10. MONIX TASKͷ࢖͍ํ MONIX TASKʹ͓͚ΔඇಉظڥքΛཧղ͠Α͏ import Scheduler.Implicits.global def task(i: Int) =

    Task.eval { printCurrentThread() i } val cubeSum = for { i <- task(11) //on run-main-0 iCube <- Task.evalAsync(i * i * i) j <- task(13) //on scala-execution-context-global-57 jCube <- Task.evalAsync(j * j * j) } yield { printCurrentThread() //on scala-execution-context-global-57 iCube + jCube } cubeSum.runAsync{ case Right(value) => println(value) case Left(err) => println(err.getMessage) } ඇಉظڥքʁ
  11. ඇಉظڥք ඇಉظڥք<ASYNCHRONOUS BOUNDARY>ͱ͸ʁ ▸ ʮReactive Manifest GlossaryʯΑΓ ▸ Isolation can

    be defined in terms of decoupling …… Decoupling in time means that the sender and receiver …… do not need to be present at the same time for communication to be possible. It is enabled by adding asynchronous boundaries between the components ……
  12. ඇಉظڥք ඇಉظڥք<ASYNCHRONOUS BOUNDARY>ͱ͸ʁ ▸ ʮReactive Manifest GlossaryʯΑΓ ▸ ִ཭<Isolation> ͸ɺ੾Γ཭͢͜ͱʹΑͬͯୡ੒͞ΕΔ……࣌ؒʹ

    ͓͚Δ੾Γ཭͠ͱ͸ɺૹΓखͱड͚ख͕…….ಉ࣌ʹίϛϡχέʔ γϣϯ͢Δඞཁ͕ͳ͍ͱ͍͏͜ͱΛҙຯ͢Δɻ͜ͷ੾Γ཭͠͸ɺ ίϯϙʔωϯτؒʹඇಉظڥքΛૠೖ͢Δ͜ͱͰՄೳʹͳΔ…… ▸ ͢ͳΘͪඇಉظڥք͸ɺૹΓखͱड͚ख͕ʮಉظ௨৴͢Δඞཁ͕ͳ ͍ʯͱ͍͏ؔ܎Λ໌ࣔ͢ΔͨΊͷ΋ͷ
  13. MONIX TASKͷඇಉظڥք TASK ʹ͓͚Δඇಉظڥք ▸ (task: Task).executeAsync ▸ taskͷલʹඇಉظڥքΛૠೖ ▸

    Task.evalAsync(…) 㱻 Task.eval(…).executeAsync ▸ (task: Task).asyncBoundary ▸ taskͷޙʹඇಉظڥքΛૠೖ
  14. MONIX TASKͷඇಉظڥք ඇಉظڥք͸Ͳ͜ʁ import Scheduler.Implicits.global def task(i: Int) = Task.eval

    { printCurrentThread() i } val cubeSum = for { i <- task(11) //on run-main-0 iCube <- Task.evalAsync(i * i * i) j <- task(13) //on scala-execution-context-global-57 jCube <- Task.evalAsync(j * j * j) } yield { printCurrentThread() //on scala-execution-context-global-57 iCube + jCube } cubeSum.runAsync{ case Right(value) => println(value) case Left(err) => println(err.getMessage) }
  15. MONIX TASKͷඇಉظڥք ඇಉظڥք͸͜͜ import Scheduler.Implicits.global def task(i: Int) = Task.eval

    { printCurrentThread() i } val cubeSum = for { i <- task(11) //on run-main-0 iCube <- Task.evalAsync(i * i * i) j <- task(13) //on scala-execution-context-global-57 jCube <- Task.evalAsync(j * j * j) } yield { printCurrentThread() //on scala-execution-context-global-57 iCube + jCube } cubeSum.runAsync{ case Right(value) => println(value) case Left(err) => println(err.getMessage) } ඇಉظڥք ඇಉظڥք
  16. MONIX TASKͷඇಉظڥք εϨου͕੾ΓସΘͬͯΔͷ͸͜͜ import Scheduler.Implicits.global def task(i: Int) = Task.eval

    { printCurrentThread() i } val cubeSum = for { i <- task(11) //on run-main-0 iCube <- Task.evalAsync(i * i * i) j <- task(13) //on scala-execution-context-global-57 jCube <- Task.evalAsync(j * j * j) } yield { printCurrentThread() //on scala-execution-context-global-57 iCube + jCube } cubeSum.runAsync{ case Right(value) => println(value) case Left(err) => println(err.getMessage) } ඇಉظڥք ඇಉظڥք ݺͼग़͠εϨουʢϝΠϯʣ ExecutionContext಺ͷεϨουϓʔϧ ExecutionContext಺ͷεϨουϓʔϧ
  17. εϨου͕੾ΓସΘͬͯΔͷ͸͜͜ import Scheduler.Implicits.global def task(i: Int) = Task.eval { printCurrentThread()

    i } val cubeSum = for { i <- task(11) //on run-main-0 iCube <- Task.evalAsync(i * i * i) j <- task(13) //on scala-execution-context-global-57 jCube <- Task.evalAsync(j * j * j) } yield { printCurrentThread() //on scala-execution-context-global-57 iCube + jCube } cubeSum.runAsync{ case Right(value) => println(value) case Left(err) => println(err.getMessage) } ඇಉظڥք ඇಉظڥքʁ ݺͼग़͠εϨουʢϝΠϯʣ ExecutionContext಺ͷεϨουϓʔϧ MONIX TASKͷඇಉظڥք
  18. MONIX TASKͷඇಉظڥք MONIX TASKͷඇಉظڥքͷऔΓѻ͍ ▸ ඇಉظڥքΛૠೖͨ͠ͱ͜ΖͰ ▸ ಉ࣌ʹSchedulerΛ౉͍ͯ͠Δ৔߹͸ɺͦͷSchedulerͷ΋ͭεϨουϓʔϧͰ࣮ߦ͞Ε Δɻ ▸

    Scheduler͕ࢦఆ͞Εͳ͍৔߹͸ɺrunAsync౳Ͱ҉໧తʹ౉ͨ͠SchedulerʢσϑΥϧτ ͷSchedulerʣ಺ͷεϨουϓʔϧ΁੾ΓସΘΔ ▸ طʹσϑΥϧτͷScheduler಺ͷεϨουϓʔϧͰ࣮ߦதͷ৔߹͸ɺ࣮ߦεϨου͸ ੾ΓସΘΒͳ͍͜ͱ͕͋Δ ▸ Scalaඪ४Future͸ɺmap / flatMap͝ͱʹڧ੍తʹεϨουϓʔϧʹ౉͞ΕΔ ▸ Φʔόʔϔου͕େ͖͍
  19. TASK ʹ͓͚ΔඇಉظڥքʴSCHEDULERΛࢦఆ ▸ (task: Task).asyncBoundary(s: Scheduler) ▸ taskͷޙʹඇಉظڥքΛૠೖ͠ɺ͔࣮ͭߦ͢ΔSchedulerΛࢦఆ͢Δɻ ▸ (task:

    Task).executeOn(s: Scheduler) ▸ taskͷલʹඇಉظڥքΛૠೖ͠ɺ͔࣮ͭߦ͢ΔSchedulerΛࢦఆ͢Δɻ ▸ taskͷޙʹ΋ඇಉظڥքΛૠೖ͠ɺ͔࣮ͭߦ͢ΔSchedulerΛσϑΥϧ τ΁੾Γସ͑Δ ▸ v3.0.0͔ΒɺSwitch Back͢ΔΑ͏ʹͳͬͨ MONIX TASKͷඇಉظڥք
  20. ࣮ߦ͢ΔSCHEDULERΛࢦఆ͢Δ import monix.eval.Task def task(i: Int) = Task.eval { printCurrentThread()

    i } def heavyCalculation(i: Int): Task[BigDecimal] = Task.eval { printCurrentThread() Seq.fill(100)(i).foldLeft(BigDecimal(i))(_ * _) } import Scheduler.Implicits.global val io = Scheduler.io("blocking") val result = for { value <- task(151) // on run-main-24 another <- task(100) // on run-main-24 calculated <- heavyCalculation(value).executeOn(io) // on blocking-808 calcu <- heavyCalculation(another).executeOn(io) // on blocking-808 } yield calculated + calcu result.runAsync { case Right(value) => printCurrentThread() // on scala-execution-context-global-809 println(value) case Left(err) => println(err) } MONIX TASKͷඇಉظڥք
  21. MONIX TASKͷ࢖͍ํ SCALAඪ४FUTUREͱͷINTEROP ▸ Scalaඪ४ Futureͱͷ૬ޓม׵͕ޮ཰Α͘؆୯ʹͰ͖Δ ▸ Task.fromFuture[A](f: Future[A]): Task[A]

    ▸ Task.deferFuture[A](fa: => Future[A]): Task[A] ▸ Task.deferFutureAction[A](f: (Scheduler) => Future[A]): Task[A] ▸ Task#runToFuture(implicit s: Scheduler): CancelableFuture[A]
  22. SCALAඪ४FUTUREͱͷINTEROP TASK . FROM FUTURE ▸ Task.fromFuture[A](f: Future[A]): Task[A] ▸

    ઌߦධՁ(eager)͞ΕΔTaskΛ࡞Δ ▸ Future.successful΍Future.faildͰఆٛ͞ΕͨFutureͳ ͲɺܭࢉࡁΈͷFutureΛแΉɻ
  23. SCALAඪ४FUTUREͱͷINTEROP TASK . DEFER FUTURE ▸ Task.deferFuture[A](fa: => Future[A]): Task[A]

    ▸ ஗ԆධՁ(lazy)͞ΕΔTaskΛ࡞Δ ▸ (ExecutionContextΛཁٻ͠ͳ͍) FutureΛฦ͢ܭࢉΛแ Ήɻ
  24. SCALAඪ४FUTUREͱͷINTEROP TASK . DEFER FUTURE ACTION ▸ Task.deferFutureAction[A](f: (Scheduler) =>

    Future[A]): Task[A] ▸ ஗ԆධՁ(lazy)͞ΕΔTaskΛ࡞Δɻ ▸ Scheduler (ExecutionContext)ΛҾ਺ʹͱΔɺFutureΛฦ ͢ܭࢉΛแΉɻ͜ͷFutureʹ౉͞ΕΔExecutionContext ͸ɺ࠷ޙʹTask#runAsyncͰ౉ͨ͠SchedulerͷΠϯελ ϯεͰ͋Δɻ
  25. SCALAඪ४FUTUREͱͷINTEROP TASK . RUN TO FUTURE ▸ Task#runToFuture(implicit s: Scheduler):

    CancelableFuture[A] ▸ TaskͷධՁΛ։࢝͠ɺFuture΁ม׵͢Δ
  26. MONIX TASKͷ࢖͍ํ ΤϥʔϋϯυϦϯά import monix.eval.Task //ࣦഊ͢Δ͔΋ def loadHeavyJson(id: Long): Task[JSON]

    = ??? import scala.concurrent.duration._ // Exponential BackoffͰ࠶ࢼߦ // 100 millis, 200 millis, 400 millis, 800 millis, 1600 millis, 3200 millis val heavyJson: Task[JSON] = ɹloadHeavyJson(123L).onErrorRestartLoop(100.millis) { (e, delay, retry) => if(delay < 4.seconds) retry(delay * 2).delayExecution(delay) else Task.raiseError(e) ɹ}.memoizeOnSuccess //੒ޭͨ͠ΒϝϞԽ
  27. MONIX TASKͷ࢖͍ํ Ωϟϯηϧ import monix.eval.Task import monix.execution.Scheduler.Implicits.global val cancelable =

    heavyJson.runAsync { case Left(e) => println(e) case Right(json) => println(json) } //΋͠ඞཁͳΒ cancelable.cancel()
  28. MONIX TASKͷ࢖͍ํ Ϧιʔεղ์ import monix.eval.Task class Client() { def get(id:

    Long): JSON = ??? def release(): Unit = ??? } def loadHeavyJsonByClient(id: Long): Task[JSON] = Task.eval(new Client()) .bracket { client => Task.evalAsync(client.get(id)) } { client => //੒൱ʹؔΘΒͣϦϦʔε Task.eval(client.release()) }
  29. ฒྻܭࢉAPI ▸ tasks: Seq[Task[T]]͕͋ͬͨͱ͖ ▸ Task.sequence(tasks) Ͱ௚ྻܭࢉ͢ΔTask[Seq[T]] ▸ Task.gather(tasks)Ͱฒྻܭࢉ͢ΔTask[Seq[T]] ▸

    Task.gatherN(parallerism:Int)(tasks)Ͱɺಉ࣌ฒྻ਺ࢦఆͰ ฒྻܭࢉ͢ΔTask[List[A]] ▸ Task.parMap2(taskA, taskB)((A,B) => R)Ͱɺฒྻܭࢉ͠Task[R] MONIX TASKͷ࢖͍ํ
  30. ؔ਺ܕ ܕΫϥεෆཁͷTASK API TASK.ATTEMPT ▸ Task#attempt: Task[Either[Throwable, A]] ▸ MonadErrorܕΫϥεͷattemptͱಉ౳

    ▸ ੒ޭ஋΋͘͠͸ࣦഊ஋ΛEitherܕ΁্࣋ͪ͛Δ ▸ ੒ޭ͍ͯ͠Δ৔߹͸Task(Right(_: A)) ▸ ࣦഊ͍ͯ͠Δ৔߹͸Task(Left(_: Throwable))
  31. ؔ਺ܕ ܕΫϥεෆཁͷTASK API TASK.MAP2 / TASK.PARMAP2 OR MORE ▸ Task.map2[A1,

    A2, R](fa1: Task[A1], fa2: Task[A2])(f: (A1, A2) => R): Task[R] ͳͲͷmapN, parMapN ▸ ApplicativeܕΫϥεͷmapNͱಉ౳(࣮૷ґଘͰ௚ྻorฒྻܭࢉ ͞ΕΔ) ▸ map2͸fa1ͱfa2͸௚ྻʹܭࢉ͞ΕΔ ▸ fa1ͷ׬ྃޙʹfa2͕ܭࢉ͞ΕΔ ▸ parMap2͸fa1ͱfa2͸ฒྻʹܭࢉ͞ΕΔ
  32. ؔ਺ܕ ܕΫϥεෆཁͷTASK API TASK.SEQUENCE ▸ Task.sequence[A, M[X] <: Iterable[X]](in: M[Task[A]])

    (implicit bf: BuildFrom[M[Task[A]], A, M[A]]): Task[M[A]] ΍ɺTask.traverse ▸ TraverseܕΫϥεͷsequence , traverseͱ΄΅ಉ౳ ▸ MͱTaskͷೖΕࢠͷॱ൪ΛೖΕସ͑Δɻ
  33. ؔ਺ܕ ܕΫϥεෆཁͷTASK API TASKͱܕΫϥε ▸ Task͸஗ԆධՁ͢Δࢀরಁաͳσʔλߏ଄ɻlawfulͳܕΫϥεΠϯελϯε΋༻ҙ͞Ε͍ͯΔɻ ▸ ७ਮؔ਺ܕͷઃܭΛͨ͠ιϑτ΢ΣΞͰ΋࢖͍΍͍͢ɻ ▸ cats-effect:

    ▸ Effect, ConcurrentEffect, Async, Concurrent, ContextShift ▸ cats: ▸ MonadError, Applicative, Monad, CoFlatMap ▸ Parallel, Monoid, Semigroup ▸ Future͸ઌߦධՁͰ෭࡞༻ΛҾ͖ىͨ͜͢ΊɺMonad౳ͷܕΫϥεͷlawΛຬͨ͞ͳ͍ɻ MonadΛཁٻ͍ͯ͠Δͱ͜ΖͰFutureΛ࢖͏ͱɺ༧ظͤ͵ΤϥʔΛҾ͖ى͜͢Մೳੑ͋Γɻ
  34. MONIX TASKͷ࢖͍ํ MONIX TASK ͷخ͍͠ͱ͜Ζʢ·ͱΊʣ ▸ ॊೈʹඇಉظڥքΛઃఆՄೳ ▸ ඞཁͳ৔ॴͰඞཁͳ͚ͩɺඇಉظॲཧʹ͢Δ͜ͱ͕Մೳ ▸

    Scala Futureͱͷinterop͕༏ल ▸ Lightbend Ecosystemʹ৐ͬͨΞϓϦέʔγϣϯͰ΋ಋೖ͠΍͍͢ ▸ ਖ਼ৗύεҎ֎ͷϋϯυϦϯά͕ॊೈ ▸ ΤϥʔɺϦτϥΠɺΩϟϯηϧɺϦιʔεղ์ͳͲ ▸ ฒྻܭࢉAPI͕๛෋ ▸ ஗ԆධՁͷϝϦοτΛڗड ▸ ؔ਺ܕ༝དྷͷܕΫϥεͷ஌͕ࣝͳͯ͘΋े෼࢖͑Δ
  35. FUTUREͰ࣮૷͞ΕͨγεςϜ D0MAIN trait UserRepository { type Ctx <: Context def

    resolveById(id: UserId)(implicit C: Ctx): Future[User] def update(user: User)(implicit C: Ctx): Future[Unit] } trait UserNameChangeService { val repository: UserRepository def changeName(userId: UserId, newName: String)(implicit C: repository.Ctx): Future[Unit] = { implicit val ec = C.ec for { user <- repository.resolveById(userId) modifiedUser = user.copy(name = newName) _ <- repository.update(modifiedUser) } yield () } } trait Context { def ec: ExecutionContext } case class UserId(value: Long) extends AnyVal case class User(id: UserId, name: String)
  36. APPLICATION class UserRepositoryImpl extends UserRepository { override type Ctx =

    FutureIOContext import UserRecord.{u, userIdBinder} override def resolveById(id: UserId)(implicit C: FutureIOContext): Future[User] = { import C._ Future( Future.fromTry( withSQL { select.from(UserRecord as u).where.eq(u.id, id).limit(1) }.map(UserRecord(_, u.resultName)).single().apply() .fold[Try[User]](Failure( /* … */ )))(ur => Success(ur.toUser)) ) ).flatten } override def update(user: User)(implicit C: FutureIOContext): Future[Unit] = { import C._ Future { withSQL { QueryDSL.update(UserRecord as u).set(u.id -> user.id.value, u.name -> user.name) }.update().apply() } } } class UserNameChangeServiceImpl(override val repository: UserRepositoryImpl) extends UserNameChangeService { def changeNameTx(userId: UserId, newName: String)(implicit ec: ExecutionContext): Future[Unit] = DB.localTx { /* … * / } } case class FutureIOContext(session: DBSession, ec: ExecutionContext) extends Context{ implicit val dbSession: DBSession = session implicit val executionContext: ExecutionContext = ec } FUTUREͰ࣮૷͞ΕͨγεςϜ
  37. CONTROLLER object Runner extends App { val repository = new

    UserRepositoryImpl() val service = new UserNameChangeServiceImpl(repository) import ExecutionContext.Implicits.global service.changeNameTx(UserId(1L), "Martin Odersky").onComplete { case Success(_) => //do something case Failure(_) => //do something } } FUTUREͰ࣮૷͞ΕͨγεςϜ
  38. Domain Service Repository Repository Impl Application Service Controller Future Future

    Task Task.deferFutureAction Future TASKΛͲ͔͜Βಋೖͨ͠ΒΑ͍͔ʁ MONIX TASKͷಋೖ
  39. CONTROLLERͰTASKΛ࢖͏ object Runner extends App { val repository = new

    UserRepositoryImpl() val service = new UserNameChangeServiceImpl(repository) val task = Task.deferFutureAction { implicit scheduler => service.changeNameTx(UserId(1L), "Martin Odersky") }.onErrorRestartLoop(0) { (t, current, handle) => if (current < 5) handle(current + 1).delayExecution(300.millis) else Task.raiseError(t) } import monix.execution.Scheduler.Implicits.global task.runAsync { case Left(_) => //do something case Right(_) => //do something } } FUTUREͰ࣮૷͞ΕͨγεςϜ
  40. ΋͠APPLICATION SERVICEͰTASKΛ࢖͍ͨ͘ͳͬͨΒ Domain Service Repository Repository Impl Application Service Controller

    Future Future Task Task.deferFutureAction Future Task Task.deferFutureAction MONIX TASKͷಋೖ
  41. APPLICATION SERVICEͰTASKΛ࢖͏ class UserNameChangeServiceImpl(override val repository: UserRepositoryImpl) extends UserNameChangeService {

    def changeNameTx(userId: UserId, newName: String): Task[Unit] = DB.localTx { session => Task.deferFutureAction{ scheduler => implicit val ctx = FutureIOContext(session, scheduler) changeName(userId, newName) } } } FUTUREͰ࣮૷͞ΕͨγεςϜ
  42. ΋͠REPOSITORYͷ࣮૷ͰTASKΛ࢖͍ͨ͘ͳͬͨΒ Domain Service Repository Repository Impl Application Service Controller Task

    Future Future Task Task.deferFutureAction Task Task.deferFutureAction MONIX TASKͷಋೖ
  43. MONIX TASKͷಋೖͷ࢓ํ - ·ͱΊ ▸ Task͸ॊೈͳඇಉظڥքͷઃఆɺࣦഊύεͷϋϯυϦϯάͳͲɺ࣮༻ తͳAPI͕ἧ͍ͬͯΔ ▸ ControllerͳͲɺෳࡶͳඇಉظॲཧ΍ɺࣦഊͷϋϯυϦϯά͕ٻΊΒ ΕΔͱ͜ΖͰಋೖ͠΍͍͢

    ▸ Task͸Futureͱͷinterop͕ॆ࣮͍ͯ͠ΔͨΊɺincrementalʹಋೖ͢Δ ͜ͱ͕Ͱ͖Δ ▸ ಉظॲཧͷͨΊʹFuture / Try͕ࠞࡏ͍ͯ͠Δ৔߹ɺඇಉظTaskͱಉ ظTaskʹͦΕͧΕஔ׵Մೳ MONIX TASKͷಋೖ
  44. MONIXΛ࢖͏ͱ͖ -༨ஊ ▸ Effectܕ Injection (a.k.a “tagless final”) ͢Δ΂͖͔ ▸

    Monix TaskΛ࢖͏্Ͱ͸ඞਢͰ͸ͳ͍ɻReaderTͱ߹Θͤ ͯ࢖͍͍ͨ࣌ͳͲɺศརͳ࣌΋͋Δɻ ▸ ςελϏϦςΟͳͲɺଞͷ࣠Ͱಋೖ͢΂͖͔ݕ౼͠·͠ΐ ͏ɻͦͷ্Ͱ͸ϝϦοτ͕͋ΔͳΒಋೖ͢Ε͹ྑ͍ɻ MONIX TASKͷಋೖ
  45. APPENDIX MONIX 3.0.0ͷओͳมߋ఺ ▸ ґଘϥΠϒϥϦͷมߋ ▸ APIͷमਖ਼ ▸ ڍಈͷมߋ ▸

    ໊લͷमਖ਼ ▸ ฒྻܭࢉAPIͷ֦ॆ ▸ Iterant, Fiber, Localͷ௥Ճ
  46. MONIX 3.0.0ͷมߋ఺ APIͷमਖ਼ - ڍಈͷมߋ ▸ Task#executeOn͕ɺσϑΥϧτscheduler΁ࣗಈͰswitch back ▸ https://github.com/monix/monix/pull/670

    ▸ https://gitter.im/monix/monix/archives/2018/05/10? at=5af3fffab84be71db9f6bfdb ▸ https://gitter.im/monix/monix/archives/2018/05/10? at=5af3fe575a1d895fae2bb6fc
  47. APIͷमਖ਼ - RUNܥAPIͷ੔ཧ ▸ Task#runAsync => Task#runToFuture ▸ Task#runOnComplete(Try[A] =>

    Unit) => Task#runAsync(Either[Throwable, A] => Unit) MONIX 3.0.0ͷมߋ఺
  48. MONIX 3.0.0ͷมߋ఺ APIͷ௥Ճ - CATS-EFFECT COMPATIBLEͳܕม׵ ▸ Task#toܥ ▸ to[F[_]],

    toAsync[F[_]], toConcurrent[F[_]] ▸ Task.fromܥ ▸ fromEffect, fromConcurrentEffectͳͲ
  49. MONIX 3.0.0ͷมߋ఺ MONIX 3.0.0 ͷ஫ҙࣄ߲ ▸ 2019/10/27࣌఺Ͱ͸υΩϡϝϯτͷߋ৽͕௥͍͍͍ͭͯͳ͍ ▸ Documentation, scaladocͷίϝϯτ

    ▸ େےͰ͸มߋ͸ͳ͍͕ɺࡉ෦ͷهड़͸৴༻͠ͳ͍ํ͕͍͍ ▸ ඞͣ3.0.0Ͱಈ࡞֬ೝ͠Α͏ ▸ GitHub issues / PullRequests΋ඞཁʹԠͯ͡ࢀর͠Α͏ ▸ PullReqνϟϯε!!