|
|
package nova.monadic_sfx.util.reactive
import cats.effect.Sync import monix.bio.Task import monix.execution.Scheduler import monix.reactive.Observable import monix.reactive.OverflowStrategy import monix.reactive.subjects.ConcurrentSubject
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.deferAction { implicit s => Task { val subject = ConcurrentSubject.publish[A](overflowStrategy)
val fold: ((A, M), A) => (A, M) = { case ((_, state), action) => { val (newState, effects) = reducer(state, action)
effects.subscribe(subject.onNext _)
action -> newState } }
val obs = subject .scan[(A, M)](initialAction -> initialState)(fold) .behavior(initialAction -> initialState) .refCount
val res = middlewares.view.reverse.foldLeft(obs) { case (obs, middleware) => middleware(obs) }
MonixProSubject.from( subject, res ) } }
def create[F[_], A, M]( initialAction: A, initialState: M, reducer: Reducer[A, M] )(implicit s: Scheduler, F: Sync[F]): F[Observable[(A, M)]] = F.delay { val subject = ConcurrentSubject.publish[A]
val fold: ((A, M), A) => (A, M) = { case ((_, state), action) => { val (newState, effects) = reducer(state, action)
effects.subscribe(subject.onNext _)
action -> newState } }
subject .scan[(A, M)](initialAction -> initialState)(fold) .behavior(initialAction -> initialState) .refCount }
}
|