redukt-v2.kt

fun <S, E> createKnot(
    initial: S,
    eventsSource: Observable<E>,
    reducer: (S, E) -> S,
    transformer: ObservableTransformer<in Triple<S, S, E>, out E>
): Observable<E> {
  val state = behaviorSubjectDefault(initial)
  val events = publishSubject<E>()
  eventsSource.subscribe(events::onNext)
  events.withLatestFrom(state, BiFunction<E, S, Triple<S, S, E>> { ev, oldState ->
    val newState = reducer(oldState, ev)
    state.onNext(newState)
    Triple(oldState, newState, ev)
  })
      .compose(transformer)
      .doOnError { RxJavaPlugins.onError(it) }
      .subscribe(events)
  return events
}