redukt-v3.kt

fun <S, E> createConnectableKnot(
    initial: S,
    eventsSource: Observable<E>,
    reducer: (S, E) -> S,
    transformer: ObservableTransformer<in Triple<S, S, E>, out E>
): ConnectableObservable<E> {
  return Observable.create<E> { observer ->
    val state = behaviorSubjectDefault(initial)
    val events = publishSubject<E>()

    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 { t ->
          observer.onNext(t)
          events.onNext(t)
        }
    eventsSource.subscribe(events)
  }
      .publish()
}