Caching RxJava2 call adapter for Retrofit2

CachingRxJava2CallAdapterFactory.kt

package retrofit2.adapter.rxjava2

import io.reactivex.Completable
import io.reactivex.Flowable
import io.reactivex.Maybe
import io.reactivex.Observable
import io.reactivex.Single
import retrofit2.Call
import retrofit2.CallAdapter
import retrofit2.Retrofit
import retrofit2.http.GET
import java.lang.reflect.Type

/**
 * Wrapper for [RxJava2CallAdapterFactory] that supports Rx-chains caching.
 *
 * By default all GET requests to API will be cached, if you want change this behavior add annotation
 * [DisableRxCache] to method that you don't want to cache.
 */
class CachingRxJava2CallAdapterFactory private constructor(
    private val delegate: CallAdapter.Factory
) : CallAdapter.Factory() {

    companion object {
        /** Returns [RxJava2CallAdapterFactory] with cache support. */
        fun create(): CallAdapter.Factory = CachingRxJava2CallAdapterFactory(RxJava2CallAdapterFactory.create())
    }

    override fun get(returnType: Type, annotations: Array<Annotation>, retrofit: Retrofit): CallAdapter<*, *>? {
        val adapter = delegate.get(returnType, annotations, retrofit) as? RxJava2CallAdapter<*>

        // Сaching only suitable for GET requests
        return if (annotations.any { it is GET } && !annotations.any { it is DisableRxCache }) {
            adapter?.let { CachingRxJava2CallAdapter(it) }
        } else {
            adapter
        }
    }
}

/** Disable Rx-chain caching for method. */
@MustBeDocumented
@Target(AnnotationTarget.FUNCTION)
@Retention(AnnotationRetention.RUNTIME)
annotation class DisableRxCache

private class CachingRxJava2CallAdapter<R>(private val delegate: CallAdapter<R, Any>) :
    CallAdapter<R, Any> by delegate {

    private var cache = mutableMapOf<String, Any>()

    @Synchronized
    override fun adapt(call: Call<R>): Any {
        // Calls with different parameters should be cached separately. So we use URL with query params as key.
        val url = call.request().url().toString()
        return cache.getOrPut(url) {
            when (val adaptedCall = delegate.adapt(call)) {
                is Completable -> adaptedCall.cache().doAfterTerminate { clearCache(url) }
                is Maybe<*> -> adaptedCall.cache().doAfterTerminate { clearCache(url) }
                is Single<*> -> adaptedCall.cache().doAfterTerminate { clearCache(url) }
                is Flowable<*> -> adaptedCall.cache().doAfterTerminate { clearCache(url) }
                is Observable<*> -> adaptedCall.cache().doAfterTerminate { clearCache(url) }
                else -> error("Unexpected return type: ${adaptedCall.javaClass.simpleName}")
            }
        }
    }

    @Synchronized
    private fun clearCache(url: String) {
        cache.remove(url)
    }
}