|
|
package nova.monadic_sfx.util.reactive.store
import monix.bio.Task import monix.eval.Coeval import monix.reactive.Observer import monix.reactive.OverflowStrategy import monix.reactive.subjects.ConcurrentSubject import cats.effect.Resource import monix.{eval => me} import monix.catnap.ConcurrentQueue import nova.monadic_sfx.util.IOUtils import monix.reactive.Observable import io.odin.Logger import nova.monadic_sfx.implicits._
object Store { def createL[A, M]( initialAction: A, initialState: M, reducer: Reducer[A, M], middlewares: Seq[Middleware[A, M]] = Seq.empty, overflowStrategy: OverflowStrategy.Synchronous[A] = OverflowStrategy.DropOld(50) ): Task[Store[A, M]] = Task.deferAction { implicit s => Task { val subject = ConcurrentSubject.publish[A](overflowStrategy)
val fold: ((A, M), A) => Coeval[(A, M)] = { case ((_, state), action) => Coeval { val (newState, effects) = reducer(state, action)
effects.subscribe(subject.onNext _)
action -> newState } }
val obs = subject .scanEval0F[Coeval, (A, M)]( Coeval.pure(initialAction -> initialState) )(fold)
val res = middlewares .foldLeft(obs) { case (obs, middleware) => middleware(obs) } .doOnNextF(i => Coeval(println(s"Emitted item 1: $i"))) .behavior(initialAction -> initialState) .refCount
res.subscribe(Observer.empty)
// .doOnNextF(i => Coeval(println(s"Emitted item 2: $i")))
MonixProSubject.from( subject, res ) } }
// : Resource[Task, Store[A, M]]
def backpressured[A, M]( initialAction: A, initialState: M, reducer: Reducer[A, M], logger: Logger[Task], middlewares: Seq[Middleware[A, M]] = Seq.empty ) = {
for { queue <- ConcurrentQueue[Task].bounded[A](10) source <- Task.deferAction(implicit s => Task { val fold: ((A, M), A) => Task[(A, M)] = { case ((_, state), action) => for { _ <- Task.unit (newState, effects) = reducer(state, action) _ <- effects .doOnNextF(queue.offer) .completedL .toIO // .start
} yield action -> newState }
val obs = Observable .repeatEvalF(queue.poll) .scanEval0F[Task, (A, M)]( Task.pure(initialAction -> initialState) )(fold)
val res = // middlewares
// .foldLeft(obs) {
// case (obs, middleware) => middleware(obs)
// }
obs .doOnNextF(i => logger.debug(s"Emitted item 1: $i")) .behavior(initialAction -> initialState) .refCount
// res.subscribe(Observer.empty)
// .doOnNextF(i => Coeval(println(s"Emitted item 2: $i")))
res } ) } yield new MyStore(Sink2.concurrentQueue(queue), source)
}
}
final class MyStore[A, M]( val sink: Sink2[A], val source: Observable[(A, M)] )
|