CoinbaseRxJavaCallAdapterFactory.kt

/**
 * A [CallAdapter.Factory] which allows [CoinbaseResponse] objects to 
 * be returned from RxJava streams.
 *
 * Adding this class to [Retrofit] allows you to return [Observable], 
 * [Flowable], [Single], or [Maybe] types parameterized with [CoinbaseResponse] 
 * from service methods. This adapter must be registered before an adapter 
 * that is capable of adapting RxJava streams.
 *
 * For the type [Observable<CoinbaseResponse<SuccessBody, ErrorBody>>], 
 * the following semantics are provided:
 *
 * 1. 2xx responses call onNext with the deserialized body set as 
 * [CoinbaseResponse.body]
 * 2. non-2xx responses call onNext with the deserialized error body 
 * set as [CoinbaseResponse.errorBody]
 * 3. Calls that fail due to network issues call onNext with network errors set as
 * [CoinbaseResponse.networkError]
 */
class CoinbaseRxJavaCallAdapterFactory private constructor(): CallAdapter.Factory() {

    companion object {
        @JvmStatic
        fun create() = CoinbaseRxJavaCallAdapterFactory()
    }

    override fun get(
        returnType: Type,
        annotations: Array<Annotation>,
        retrofit: Retrofit
    ): CallAdapter<*, *>? {
        val rawType = getRawType(returnType)

        if (rawType == Completable::class.java) {
            // Completable is not parameterized (which is what the 
            // rest of this method deals with) so it can only be 
            // created with a single configuration.
            return retrofit.nextCallAdapter(this, returnType, annotations)
        }

        val isFlowable = rawType == Flowable::class.java
        val isSingle = rawType == Single::class.java
        val isMaybe = rawType == Maybe::class.java
        val isObservable = rawType == Observable::class.java

        if (!isFlowable && !isSingle && !isMaybe && !isObservable) {
            return null
        }

        if (returnType !is ParameterizedType) {
            val name = when {
                isFlowable -> "Flowable"
                isSingle -> "Single"
                isMaybe -> "Maybe"
                else -> "Observable"
            }
            throw IllegalStateException(
                "$name return type must be parameterized " +
                        "as $name<Foo> or $name<? extends Foo>"
            )
        }

        val observableEmissionType = CallAdapter.Factory.getParameterUpperBound(
           0, 
           returnType)
        if (getRawType(observableEmissionType) != CoinbaseResponse::class.java) {
            return null
        }

        if (observableEmissionType !is ParameterizedType) {
            throw IllegalStateException(
                "CoinbaseResponse must be parameterized " +
                        "as CoinbaseResponse<SuccessBody, ErrorBody>"
            )
        }

        val successBodyType = getParameterUpperBound(0, observableEmissionType)
        val delegateType = Types.newParameterizedType(
           getRawType(returnType), 
           successBodyType)
        val delegateAdapter = retrofit.nextCallAdapter(
            this,
            delegateType,
            annotations
        )

        val errorBodyType = getParameterUpperBound(1, observableEmissionType)
        val errorBodyConverter = retrofit.nextResponseBodyConverter<Any?>(
            null,
            errorBodyType,
            annotations)

        @Suppress("UNCHECKED_CAST") // CallAdapter type is not known at compile time.
        return CoinbaseRxJavaCallAdapter(
            successBodyType,
            delegateAdapter as CallAdapter<Any, Any>,
            errorBodyConverter,
            isObservable,
            isFlowable,
            isSingle,
            isMaybe
        )
    }
}