Polling problem

gistfile1.txt

//So here is the problem -

Using retrofit and RXjava to do server polling. I am new to RX java, so ignore my rookie mistakes.

Problem - To do command polling every 5 secs. Let say the server gives us a command polling object(Commands) which contains
two variables - "onComplete" and "result". I need to send broadcasts after my polling is completed - Success or failure.  

Inicially I wrote this code using RXjava operators.

-------------------------------------------------------------------------------------------------
webService.commandPolling(...params)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
                    @Override
                    public Observable<?> call(Observable<? extends Void> observable) {
                        Timber.tag(TAG).d("Repeat when is called, adding 5 secs interval");
                        return observable.delay(pollingInterval , TimeUnit.SECONDS);
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(Observable<? extends Throwable> observable) {
                        //please ignore for now, this was specific to my use case which I cant disclose.
                    }
                })
                .takeUntil(new Func1<Commands, Boolean>() {
                    @Override
                    public Boolean call(Commands commands) {
                        Timber.tag(TAG).d("Take Until is called returning %b", commands.isComplete());
                        return commands.isComplete();
                    }
                })
                .filter(new Func1<Commands, Boolean>() {
                    @Override
                    public Boolean call(Commands commands) {
                        return commands.isComplete() && commands.result;
                    }
                })
                .subscribe(new LoggingSubscriber<Commands>(TAG) {
                    @Override
                    public void onNext(Commands commands) {
                        Timber.tag(TAG).d("On Next called, Command object received");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Timber.tag   (TAG).d("On error called with %s", e.getLocalizedMessage());
                        super.onError(e);
                    }

                    @Override
                    public void onCompleted() {
                        Timber.tag(TAG).d("Command Completed and onnextwascalled is %b", onNextWasCalled);
                        super.onCompleted();
                    }
                });
 ------------------------------------------------------------------------------------------------- 
 
 So this can end in either two ways - 
 
 Lets say the command completes ie. isComplete = true and result = false, in this case only onComplete() 
 will get called. 
 
 Now for the case of isComplete = true and result = true. Both onComplete() and onNext() Will get called.  
 onNext() is called multiple times as I am returning an observable(Not a single.)
 
 Thus for me to know whether the command succeeded or failed, I need to know if onNext() was called before onComplete(); as onComplete is called in both the cases. 

 ------------------------------------------------------------------------------------------------- 
 
 
 I got it working by doing this.
 
  webService.commandPolling(..Params)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
                    @Override
                    public Observable<?> call(Observable<? extends Void> observable) {
                        //Ignore
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(Observable<? extends Throwable> observable) {
                        return observable.repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
                            @Override
                            public Observable<?> call(Observable<? extends Void> observable) {
                                Timber.tag(TAG).d("repeat when is called");
                                return observable.delay(pollingInterval , TimeUnit.SECONDS);
                            }
                        });
                    }
                })
                .takeUntil(new Func1<Commands, Boolean>() {
                    @Override
                    public Boolean call(Commands commands) {
                        Timber.tag(TAG).d("Take Until is called returning %b", commands.isComplete());
                        return commands.isComplete();
                    }
                })
                .subscribe(new LoggingSubscriber<Commands>(TAG) {
                    @Override
                    public void onNext(Commands commands) {
                        if(commands.isComplete() && commands.status) {
                            Timber.d("Polling is completed with result ");
                            unsubscribe();
                        }
                    }

                    @Override
                    public void onError(Throwable e) {
                        Timber.tag   (TAG).d("On error called with %s", e.getLocalizedMessage());
                        super.onError(e);
                    }

                    @Override
                    public void onCompleted() {
                        super.onCompleted();
                    }
                });


///////////////////////////////////////////////////////////////////////////////////////////////////////

Conclusion - 

I wanted to use the filter operator mode for efficient use. I am not stuck on this, but many of you folks wanted to 
know my use case, so thought I should write a gist.