Bridge between RxJava and Resilience4j for Retrofit.

ReactiveResillience4jCallAdapter.java

package com.github.wreulicke.resilience4j.extras;

import java.lang.annotation.Annotation;
import java.lang.reflect.Type;
import java.util.function.Predicate;

import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.retrofit.RetrofitCircuitBreaker;
import retrofit2.Call;
import retrofit2.CallAdapter;
import retrofit2.Response;
import retrofit2.Retrofit;
import retrofit2.adapter.rxjava2.RxJava2CallAdapterFactory;

/**
 * Reactive resilience4j call adaptor.
 *
 * @author wreulicke
 */
public class ReactiveResillience4jCallAdapter extends CallAdapter.Factory {
	
	private final CircuitBreaker circuitBreaker;
	
	private final Predicate<Response> successResponse;
	
	private final RxJava2CallAdapterFactory delegates;
	
	public static ReactiveResillience4jCallAdapter of(final CircuitBreaker circuitBreaker,
		RxJava2CallAdapterFactory delegates) {
		return of(circuitBreaker, Response::isSuccessful, delegates);
	}
	
	public static ReactiveResillience4jCallAdapter of(final CircuitBreaker circuitBreaker,
		final Predicate<Response> successResponse, RxJava2CallAdapterFactory delegates) {
		return new ReactiveResillience4jCallAdapter(circuitBreaker, successResponse, delegates);
	}
	
	private ReactiveResillience4jCallAdapter(final CircuitBreaker circuitBreaker,
		final Predicate<Response> successResponse
		, final RxJava2CallAdapterFactory delegates) {
		this.circuitBreaker = circuitBreaker;
		this.successResponse = successResponse;
		this.delegates = delegates;
	}
	
	@Override
	public CallAdapter<?, ?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
		CallAdapter<?, ?> callAdapter = delegates.get(returnType, annotations, retrofit);
		if (callAdapter != null) {
			return new DecorateCallAdaptor<>(callAdapter, circuitBreaker, successResponse);
		}
		return null;
	}
	
	public static class DecorateCallAdaptor<R, T> implements CallAdapter<R, T> {
		
		private final CallAdapter<R, T> callAdapter;
		
		private CircuitBreaker circuitBreaker;
		
		private Predicate<Response> successResponse;
		
		public DecorateCallAdaptor(CallAdapter<R, T> callAdapter,
			CircuitBreaker circuitBreaker, Predicate<Response> successResponse) {
			this.callAdapter = callAdapter;
			this.circuitBreaker = circuitBreaker;
			this.successResponse = successResponse;
		}
		
		@Override
		public Type responseType() {
			return callAdapter.responseType();
		}
		
		@Override
		public T adapt(Call<R> call) {
			return callAdapter.adapt(RetrofitCircuitBreaker.decorateCall(circuitBreaker, call, successResponse));
		}
	}
}