import io.reactivex.Observable
import io.reactivex.Single
import io.reactivex.schedulers.Schedulers
import junit.framework.Assert
import org.junit.Test
import java.util.*

class SkipExceptionTest {

    private val data: Map<Int, String> = mapOf(
            Pair(1, "one"),
            Pair(2, "two"),
            Pair(4, "four"),
            Pair(5, "five")
    )

    @Test
    fun canFilterOnError() {
        val actual = getStream(listOf(1, 2, 3, 4, 5))
                .subscribeOn(Schedulers.trampoline())
                .observeOn(Schedulers.trampoline())
                .test()
                .assertComplete()
                .assertNoErrors()
                .assertValueCount(1)
                .values()[0]

        val expected = listOf("one", "two", "four", "five")
        Assert.assertEquals(expected, actual)

    }

    private fun getStream(list: List<Int>): Single<List<String>> {
        return Observable.fromIterable(list)
                .flatMapSingle {
                    getValue(it)
                            .map {
                                Optional.of(it)
                            }
                            .onErrorResumeNext {
                                when (it) {
                                    is NotFoundException -> Single.just(Optional.empty())
                                    else -> Single.error(it)
                                }
                            }
                }
                .filter { it.isPresent }
                .map { it.get() }
                .toList()
    }

    private fun getValue(id: Int): Single<String> {
        return Single.fromCallable {
            data[id] ?: throw NotFoundException("data with id $id does not exist")
        }
    }

    class NotFoundException(message: String) : Exception(message)
}