Custom rx Func1 for retrofit response code handlin

I am new in rxjava, so please don't be strict...

I have request lice next one:

Observable<Login>login(String l, String p){
        return api.loginUser(l,p)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .flatMap(new Func1<Response<Login>, Observable<? extends Login>>() {
                    @Override
                    public Observable<? extends Login> call(Response<Login> r) {
                        switch (r.code()){
                            case 200:
                                requestOk(r);
                                break;
                            case 202:
                                //need to Repeat
                                login.timeout(2000,TimeUnit.Milliseconds);
                                break;
                            default:
                                //
                        }
                    }
                });
    }

Observable<Login> requestOk(final Response<Login> r){
            return Observable.create(new Observable.OnSubscribe<Login>(){
                @Override
                public void call(Subscriber<? super Login> subscriber) {
                    if (subscriber.isUnsubscribed()) return;
                    subscriber.onNext(r.body());
                    subscriber.onCompleted();
                }
            });
        }

It works ok. But there will be a lot of other requests where I need to check request code in case of repeat.

So I was trying to create custom Func1 - universal for all request:

private <T>Func1<? super Response<T>, ? extends Observable<? extends T>> customFunc(final Observable<T> o) {
        return new Func1<Response<T>, Observable<? extends T>>() {
            @Override
            public Observable<? extends T> call(Response<T> r) {
                switch (r.code()){
                    case 200:
                        //
                        break;
                    case 202:
                        //need to Repeat
                        return o.timeout(2000,TimeUnit.Milliseconds);
                        break;
                    default:
                        //
                }
            };
        };
    }

I am stuck with putting current observable from login observable into customFunc.

I believe there must be some other, easier and correct way how to do this. Will be glad any help!

You can use Transformer in order to create a generic status code verifier:

class StatusCodeVerifierTransformer<T> implements Observable.Transformer<Response<T>, T> {

    @Override
    public Observable<T> call(Observable<Response<T>> responseObservable) {
        return responseObservable.flatMap(new Func1<Response<T>, Observable<T>>() {
            @Override
            public Observable<T> call(Response<T> loginResponse) {
                switch (loginResponse.code()) {
                    case 200:
                        return Observable.just(loginResponse.body());
                    case 202:
                        //need to Repeat
                        return Observable.error(new Status202Exception());
                    default:
                        return Observable.error(new Exception("unknown error"));
                }
            }
        });

    }
}

this transformer will take any Observable of Response<T> and will transform it to Observable of T, that emits errors according to your strategy.
Then use it with compose() operator:

  Observable<Login> login(String l, String p) {
    return api.loginUser(l, p)
            .compose(new StatusCodeVerifierTransformer<>())
   }

Now you have an Observable that will emit onError with your desired exception according to your status code handling.

Now, you can retry on error like with any Observable with retry operator, for instance like this:

 api.loginUser(l, p)
            .compose(new StatusCodeVerifierTransformer<>())
            .retry(new Func2<Integer, Throwable, Boolean>() {
                @Override
                public Boolean call(Integer retryCount, Throwable throwable) {
                    return throwable instanceof Login202Exception && retryCount < MAX_RETRY;
                }
            })

BTW, some comments:

requestOk - you definitely don't need to create Observable for emitting specific value once, just use the just() operator, like in the StatusCodeVerifierTransformer example. (or for any other synchronize operation, you have plenty of operators like fromCallable(), just(), from())
Generally, create() is not a safe method of creating an Observable anymore.