Incremental Store

IncStore.scala

import monix.execution.{Ack, Cancelable, Scheduler}
import monix.reactive.observers.Subscriber
import monix.reactive.{Observable, Observer}
import monix.reactive.subjects.PublishSubject

import scala.concurrent.Future

trait IncStore[State, InAction, OutAction] { self =>
  def initialState: State
  def inAction: Observer[InAction]
  def output: Observable[(State, OutAction)]
  val state: Observable[State] = output.map(_._1)
  val outAction: Observable[OutAction] = output.map(_._2)

  def mapIncremental[NextState, NextOutAction](mapInputState:State => NextState, onAction:(NextState,OutAction) => (NextState, NextOutAction)): IncStore[NextState, InAction, NextOutAction] = new IncStore[NextState, InAction, NextOutAction] {
    override def initialState: NextState = mapInputState(self.initialState)
    override def inAction: Observer[InAction] = self.inAction
    override def output: Observable[(NextState, NextOutAction)] = self.output.map { case (state, action) => onAction(mapInputState(state), action)}
  }

  def map[NextState](mapInputState:State => NextState, onAction:(NextState,OutAction) => NextState): IncStore[NextState, InAction, OutAction] = new IncStore[NextState, InAction, OutAction] {
    override def initialState: NextState = mapInputState(self.initialState)
    override def inAction: Observer[InAction] = self.inAction
    override def output: Observable[(NextState, OutAction)] = self.output.map { case (state, action) => (onAction(mapInputState(state), action), action) }
  }

  def mapState(mapInputState: State => State): IncStore[State, InAction, OutAction] = new IncStore[State, InAction, OutAction] {
    override def initialState: State = mapInputState(self.initialState)
    override def inAction: Observer[InAction] = self.inAction
    override def output: Observable[(State, OutAction)] = self.output.map { case (state, action) => (mapInputState(state), action) }
  }
}

case class ConcreteIncStore[State, InAction, OutAction](initialState: State, inAction: Observer[InAction], output: Observable[(State, OutAction)]) extends IncStore[State, InAction, OutAction]

object IncStore {
  type Store[State, Action] = IncStore[State, Action, Action]

  def apply[State, Action](init:State, f:(State,Action) => State): Store[State, Action] = {
    val subject = PublishSubject[Action]()
    var lastState = init
    val observable = subject.map { inAction =>
      lastState = f(lastState, inAction)
      (lastState, inAction)
    }
    new ConcreteIncStore[State, Action, Action](init, subject, observable)
  }
  def incremental[State, InAction, OutAction](init: State, f:(State,InAction) => (State, OutAction)): IncStore[State, InAction, OutAction] = {
    val subject = PublishSubject[InAction]()
    var lastState = init
    val observable = subject.map { inAction =>
      val result = f(lastState, inAction)
      lastState = result._1
      result
    }
    new ConcreteIncStore[State, InAction, OutAction](init, subject, observable)
  }
}