<commit_before>package com.zeyad.rxredux.core.viewmodel

import android.arch.lifecycle.LiveData
import android.arch.lifecycle.MutableLiveData
import android.os.Parcelable
import android.util.Log
import com.jakewharton.rx.ReplayingShare
import com.zeyad.rxredux.core.*
import io.reactivex.BackpressureStrategy
import io.reactivex.Flowable
import io.reactivex.Observable
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.disposables.CompositeDisposable
import io.reactivex.functions.BiFunction
import io.reactivex.schedulers.Schedulers
import io.reactivex.subjects.BehaviorSubject

@Throws(IllegalStateException::class)
inline fun <reified T> T.throwIllegalStateException(result: Any): Nothing =
        throw IllegalStateException("Can not reduce from $this to ${T::class.java.simpleName} with $result")

interface IBaseViewModel<R, S : Parcelable, E> {

    var disposables: CompositeDisposable

    val currentStateStream: BehaviorSubject<Any>

    fun reducer(newResult: R, currentStateBundle: S): S

    fun reduceEventsToResults(event: BaseEvent<*>, currentStateBundle: Any): Flowable<*>

    fun errorMessageFactory(throwable: Throwable, event: BaseEvent<*>, currentStateBundle: PEffect<E>): Message =
            StringMessage(throwable.localizedMessage)

    fun middleware(it: PModel<*>) {
        if (it is ErrorEffect) Log.e("IBaseViewModel", "Error", it.error)
        else Log.d("IBaseViewModel", "PModel: $it")
    }

    fun store(events: Observable<BaseEvent<*>>, initialState: S): LiveData<PModel<*>> {
        currentStateStream.onNext(initialState)
        val pModels = events.toFlowable(BackpressureStrategy.BUFFER)
                .toResult()
                .publish()
                .autoConnect(0)
        val liveState = MutableLiveData<PModel<*>>()
        val states = stateStream(pModels as Flowable<Result<R>>, initialState)
        val effects = effectStream(pModels as Flowable<Result<E>>)
        Flowable.merge(states, effects)
                .doAfterNext { middleware(it) }
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe { t: PModel<*> -> liveState.value = t }
                .let { disposables.add(it) }
        return liveState
    }

    private fun stateStream(pModels: Flowable<Result<R>>, initialState: S): Flowable<PModel<*>> =
            pModels.filter { it is SuccessResult }
                    .map { it as SuccessResult }
                    .scan<PModel<S>>(SuccessState(initialState), stateReducer())
                    .map {
                        if (currentStateStream.value == it.bundle) {
                            EmptySuccessState()
                        } else {
                            currentStateStream.onNext(it.bundle)
                            it
                        }
                    }
                    .compose(ReplayingShare.instance())

    private fun effectStream(pModels: Flowable<Result<E>>): Flowable<PModel<*>> =
            pModels.filter { it is EffectResult }
                    .map { it as EffectResult }
                    .scan<PModel<*>>(EmptySuccessEffect(), effectReducer())
                    .filter { t: PModel<*> -> t !is EmptySuccessEffect }
                    .distinctUntilChanged()
                    .doOnNext {
                        if (it is SuccessEffectResult<*>) {
                            currentStateStream.onNext(it.bundle!!)
                        }
                    }


    private fun Flowable<BaseEvent<*>>.toResult(): Flowable<Result<*>> =
            observeOn(Schedulers.computation())
                    .concatMap { event ->
                        Log.d("IBaseViewModel", "Event: $event")
                        reduceEventsToResults(event, currentStateStream.value!!)
                                .map<Result<*>> {
                                    if (it is EffectResult<*>) it
                                    else SuccessResult(it, event)
                                }.onErrorReturn { ErrorEffectResult(it, event) }
                                .startWith(LoadingEffectResult(event))
                    }
                    .distinctUntilChanged()

    private fun stateReducer(): BiFunction<PModel<S>, SuccessResult<R>, PModel<S>> =
            BiFunction { currentUIModel, result ->
                SuccessState(reducer(result.bundle, currentUIModel.bundle), result.event)
            }

    private fun effectReducer(): BiFunction<PModel<*>, in EffectResult<E>, PModel<*>> =
            BiFunction { currentUIModel, result ->
                result.run {
                    when (this) {
                        is LoadingEffectResult -> LoadingEffect(currentUIModel.bundle, event)
                        is SuccessEffectResult -> successEffect(currentUIModel as PEffect<E>)
                        is ErrorEffectResult -> errorEffect(currentUIModel as PEffect<E>)
                    }
                }
            }

    private fun SuccessEffectResult<E>.successEffect(currentUIModel: PEffect<E>): SuccessEffect<E> =
            when (currentUIModel) {
                is LoadingEffect -> SuccessEffect(bundle, event)
                is EmptySuccessEffect, is SuccessEffect, is ErrorEffect -> currentUIModel.throwIllegalStateException(this)
            }

    private fun ErrorEffectResult.errorEffect(currentUIModel: PEffect<E>): ErrorEffect<E> =
            when (currentUIModel) {
                is LoadingEffect -> ErrorEffect(error, errorMessageFactory(error, event, currentUIModel), currentUIModel.bundle, event)
                is EmptySuccessEffect, is SuccessEffect, is ErrorEffect -> currentUIModel.throwIllegalStateException(this)
            }
}<commit_msg>Fixed errorMessageFactory to receive effect payload<commit_after>package com.zeyad.rxredux.core.viewmodel

import android.arch.lifecycle.LiveData
import android.arch.lifecycle.MutableLiveData
import android.os.Parcelable
import android.util.Log
import com.jakewharton.rx.ReplayingShare
import com.zeyad.rxredux.core.*
import io.reactivex.BackpressureStrategy
import io.reactivex.Flowable
import io.reactivex.Observable
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.disposables.CompositeDisposable
import io.reactivex.functions.BiFunction
import io.reactivex.schedulers.Schedulers
import io.reactivex.subjects.BehaviorSubject

@Throws(IllegalStateException::class)
inline fun <reified T> T.throwIllegalStateException(result: Any): Nothing =
        throw IllegalStateException("Can not reduce from $this to ${T::class.java.simpleName} with $result")

interface IBaseViewModel<R, S : Parcelable, E> {

    var disposables: CompositeDisposable

    val currentStateStream: BehaviorSubject<Any>

    fun reducer(newResult: R, currentStateBundle: S): S

    fun reduceEventsToResults(event: BaseEvent<*>, currentStateBundle: Any): Flowable<*>

    fun errorMessageFactory(throwable: Throwable, event: BaseEvent<*>, currentStateBundle: E): Message =
            StringMessage(throwable.localizedMessage)

    fun middleware(it: PModel<*>) {
        if (it is ErrorEffect) Log.e("IBaseViewModel", "Error", it.error)
        else Log.d("IBaseViewModel", "PModel: $it")
    }

    fun store(events: Observable<BaseEvent<*>>, initialState: S): LiveData<PModel<*>> {
        currentStateStream.onNext(initialState)
        val pModels = events.toFlowable(BackpressureStrategy.BUFFER)
                .toResult()
                .publish()
                .autoConnect(0)
        val liveState = MutableLiveData<PModel<*>>()
        val states = stateStream(pModels as Flowable<Result<R>>, initialState)
        val effects = effectStream(pModels as Flowable<Result<E>>)
        Flowable.merge(states, effects)
                .doAfterNext { middleware(it) }
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe { t: PModel<*> -> liveState.value = t }
                .let { disposables.add(it) }
        return liveState
    }

    private fun stateStream(pModels: Flowable<Result<R>>, initialState: S): Flowable<PModel<*>> =
            pModels.filter { it is SuccessResult }
                    .map { it as SuccessResult }
                    .scan<PModel<S>>(SuccessState(initialState), stateReducer())
                    .map {
                        if (currentStateStream.value == it.bundle) {
                            EmptySuccessState()
                        } else {
                            currentStateStream.onNext(it.bundle)
                            it
                        }
                    }
                    .compose(ReplayingShare.instance())

    private fun effectStream(pModels: Flowable<Result<E>>): Flowable<PModel<*>> =
            pModels.filter { it is EffectResult }
                    .map { it as EffectResult }
                    .scan<PModel<*>>(EmptySuccessEffect(), effectReducer())
                    .filter { t: PModel<*> -> t !is EmptySuccessEffect }
                    .distinctUntilChanged()
                    .doOnNext {
                        if (it is SuccessEffectResult<*>) {
                            currentStateStream.onNext(it.bundle!!)
                        }
                    }


    private fun Flowable<BaseEvent<*>>.toResult(): Flowable<Result<*>> =
            observeOn(Schedulers.computation())
                    .concatMap { event ->
                        Log.d("IBaseViewModel", "Event: $event")
                        reduceEventsToResults(event, currentStateStream.value!!)
                                .map<Result<*>> {
                                    if (it is EffectResult<*>) it
                                    else SuccessResult(it, event)
                                }.onErrorReturn { ErrorEffectResult(it, event) }
                                .startWith(LoadingEffectResult(event))
                    }
                    .distinctUntilChanged()

    private fun stateReducer(): BiFunction<PModel<S>, SuccessResult<R>, PModel<S>> =
            BiFunction { currentUIModel, result ->
                SuccessState(reducer(result.bundle, currentUIModel.bundle), result.event)
            }

    private fun effectReducer(): BiFunction<PModel<*>, in EffectResult<E>, PModel<*>> =
            BiFunction { currentUIModel, result ->
                result.run {
                    when (this) {
                        is LoadingEffectResult -> LoadingEffect(currentUIModel.bundle, event)
                        is SuccessEffectResult -> successEffect(currentUIModel as PEffect<E>)
                        is ErrorEffectResult -> errorEffect(currentUIModel as PEffect<E>)
                    }
                }
            }

    private fun SuccessEffectResult<E>.successEffect(currentUIModel: PEffect<E>): SuccessEffect<E> =
            when (currentUIModel) {
                is LoadingEffect -> SuccessEffect(bundle, event)
                is EmptySuccessEffect, is SuccessEffect, is ErrorEffect -> currentUIModel.throwIllegalStateException(this)
            }

    private fun ErrorEffectResult.errorEffect(currentUIModel: PEffect<E>): ErrorEffect<E> =
            when (currentUIModel) {
                is LoadingEffect -> ErrorEffect(error, errorMessageFactory(error, event, currentUIModel.bundle),
                        currentUIModel.bundle, event)
                is EmptySuccessEffect, is SuccessEffect, is ErrorEffect -> currentUIModel.throwIllegalStateException(this)
            }
}
{'fraction_non_alphanumeric': 0.09836493303183162, 'fraction_numerical': 0.00017394329448599757, 'mean_word_length': 2.077067166176077, 'pattern_counts': {'":': 0, '<': 112, '<?xml version=': 0, '>': 136, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '139755', 'n_tokens_mistral': 3081, 'n_tokens_neox': 2922, 'n_words': 593}