You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

37 lines
990 B

  1. package nova.monadic_sfx.implicits
  2. import monix.reactive.Observable
  3. import monix.reactive.Observer
  4. import monix.{eval => me}
  5. import monix.execution.Scheduler
  6. class ActionObservableExecuter[T](
  7. delegate: Observable[T]
  8. ) {
  9. def -->(sub: Observer[T])(implicit s: Scheduler) =
  10. delegate
  11. .doOnNext(el => me.Task(sub.onNext(el)))
  12. .subscribe()
  13. }
  14. class ActionObservableBuilder[A](
  15. observableAction: Observable[A]
  16. ) {
  17. def useLazy[T](v: => T) =
  18. new ActionObservableExecuter[T](observableAction.mapEval(_ => me.Task(v)))
  19. def use[T](cb: A => T) =
  20. new ActionObservableExecuter[T](
  21. observableAction.mapEval(ae => me.Task(cb(ae)))
  22. )
  23. def useIterable[T](cb: A => collection.immutable.Iterable[T]) =
  24. new ActionObservableExecuter[T](
  25. observableAction.flatMap(a =>
  26. Observable.suspend(Observable.fromIterable(cb(a)))
  27. )
  28. )
  29. def map[B](cb: A => B) =
  30. new ActionObservableBuilder(observableAction.mapEval(v => me.Task(cb(v))))
  31. }