AsyncResult.kt

class AsyncResult<Element>(
    private val source: (AsyncResultEmitter<Element>) -> Unit
) {

    fun subscribe(
        onSuccess: (Element) -> Unit,
        onError: (Throwable) -> Unit
    ): AsyncResultDisposable {

        val emitter = AsyncResultEmitter(onSuccess, onError)

        source.invoke(emitter)

        return object : AsyncResultDisposable {
            override fun dispose() {
                emitter.disposable?.invoke()
            }
        }
    }

    class AsyncResultEmitter<Element>(
        val onSuccess: (Element) -> Unit,
        val onError: (Throwable) -> Unit,
        var disposable: (() -> Unit)? = null
    )

    companion object {

        fun <Element> create(source: (AsyncResultEmitter<Element>) -> Unit): AsyncResult<Element> {
            return AsyncResult(source)
        }

        fun <Element> from(callable: () -> Element): AsyncResult<Element> {
            return create { it.onSuccess.invoke(callable.invoke()) }
        }
    }
}