A CallAdapterFactory capable of creating NetworkResponse instances.

KotlinRxJava2CallAdapterFactory.kt

/**
 * A [CallAdapter.Factory] which allows [NetworkResponse] objects to be 
 * returned from RxJava streams created by Retrofit.
 *
 * Adding this class to [Retrofit] allows you to write service methods like:
 *
 * fun getTokens(): Single<NetworkResponse<AccessToken,Error>>
 */
class KotlinRxJava2CallAdapterFactory : CallAdapter.Factory() {

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

        // This adapter only handles creating RxJava streams. If the caller
        // isn't asking for an RxJava stream, return null, this isn't the right
        // adapter!
        val rawType = getRawType(returnType)
        val isFlowable = rawType === Flowable::class.java
        val isSingle = rawType === Single::class.java
        val isMaybe = rawType === Maybe::class.java
        if (rawType !== Observable::class.java
            && !isFlowable
            && !isSingle
            && !isMaybe
        ) {
            return null
        }

        // Check to see if the RxJava stream is emitting instances of NetworkResponse. 
        // If not this isn't the right adapter, return null!
        val observableEmissionType = getParameterUpperBound(0, returnType)
        if (getRawType(observableEmissionType) != NetworkResponse::class.java) {
            return null
        }

        // Ask Retrofit for an adapter that's capable of creating an instance 
        // of Observable<AccessToken>
        val successBodyType = getParameterUpperBound(0, observableEmissionType)
        val delegateType = Types.newParameterizedType(
            Observable::class.java,
            successBodyType
        )
        val delegateAdapter = retrofit.nextCallAdapter(
            this,
            delegateType,
            annotations
        )

        // Ask Retrofit for a serializer than can serialize an instance of Error
        val errorBodyType = getParameterUpperBound(1, observableEmissionType)
        val errorBodyConverter = retrofit.nextResponseBodyConverter<Any>(
            null,
            errorBodyType,
            annotations
        )

        return KotlinRxJava2CallAdapter(
            successBodyType,
            delegateAdapter,
            errorBodyConverter,
            isFlowable,
            isSingle,
            isMaybe
        )
    }
}