CoinbaseRxJava2CallAdapter.kt

/**
 * Decorates stream emissions from [delegateAdapter] in [CoinbaseResponse].
 */
internal class CoinbaseRxJava2CallAdapter(
    private val successBodyType: Type,
    private val delegateAdapter: CallAdapter<Any, Any>,
    private val errorConverter: Converter<ResponseBody, Any?>,
    private val supportedType: SupportedType<Any, Any>
) : CallAdapter<Any, Any> {

    override fun adapt(call: Call<Any>): Any {
        val stream = delegateAdapter.adapt(call) //ie Single<AccessToken>

        // Turn all types to an Observable. ie Observable<AccesToken>
        val coinbaseStream = supportedType.toObservable(stream)
            .map { response ->
                when {
                    response.isSuccessful -> CoinbaseResponse<Any, Any>(
                        response,
                        response.errorBody(),
                        null
                    )
                    else -> {
                        val error = response.errorBody()
                        val errorBody = when {
                            error == null -> null
                            error.contentLength() == 0L -> null
                            else -> errorConverter.convert(error)
                        }
                        CoinbaseResponse(
                            response,
                            errorBody,
                            null
                        )
                    }
                }
            }.onErrorResumeNext(
                Function<Throwable, Observable<CoinbaseResponse<Any, Any>>> { throwable ->
                    val response: CoinbaseResponse<Any, Any> = CoinbaseResponse(
                        null,
                        null,
                        throwable as IOException
                    )
                    Observable.just(response)
                })

        return supportedType.fromObservable(coinbaseStream)
    }

    override fun responseType(): Type = successBodyType
}