Bind to an Android service using RxJava!

Rx2ServiceBindingFactory.java

import android.app.Service;
import android.content.ComponentName;
import android.content.Context;
import android.content.Intent;
import android.content.ServiceConnection;
import android.os.Binder;
import android.os.IBinder;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;

public class Rx2ServiceBindingFactory {
    public <B extends Binder> Observable<B> bind(final Context context, final Intent launch) {
        return bind(context, launch, Service.BIND_AUTO_CREATE);
    }

    public <B extends Binder> Observable<B> bind(final Context context, final Intent launch, final int flags) {
        return Observable.using(Connection::new,
                                (final Connection<B> con) -> {
                                    context.bindService(launch, con, flags);
                                    return Observable.create(con);
                                },
                                context::unbindService);
    }

    private static class Connection<B extends Binder> implements ServiceConnection, ObservableOnSubscribe<B> {
        private ObservableEmitter<? super B> subscriber;

        @Override public void onServiceConnected(ComponentName name, IBinder service) {
            if ((subscriber != null) && (service != null)) {
                //noinspection unchecked - we trust this one
                subscriber.onNext((B)service);
            }
        }

        @Override public void onServiceDisconnected(ComponentName name) {
            if (subscriber != null) {
                subscriber.onComplete();
            }
        }

        @Override public void subscribe(ObservableEmitter<B> observableEmitter) throws Exception {
            this.subscriber = observableEmitter;
        }
    }
}

RxServiceBindingFactory.java

import android.content.ComponentName;
import android.content.Context;
import android.content.Intent;
import android.content.ServiceConnection;
import android.os.Binder;
import android.os.IBinder;

import rx.Observable;
import rx.Subscriber;

public class RxServiceBindingFactory {
    public <B extends Binder> Observable<B> bind(final Context context, final Intent launch, final int flags) {
        Connection<B> con = new Connection<>();
        return Observable.unsafeCreate(con)
                         .doOnSubscribe(() -> context.bindService(launch, con, flags))
                         .doOnUnsubscribe(() -> context.unbindService(con))
                         .onBackpressureLatest();
    }

    private static class Connection<B extends Binder> implements ServiceConnection, Observable.OnSubscribe<B> {
        private Subscriber<? super B> subscriber;

        @Override public void onServiceConnected(ComponentName name, IBinder service) {
            if (subscriber != null) {
                //noinspection unchecked - we trust this one
                subscriber.onNext((B)service);
            }
        }

        @Override public void onServiceDisconnected(ComponentName name) {
            if (subscriber != null) {
                subscriber.onCompleted();
            }
        }

        @Override public void call(Subscriber<? super B> subscriber) {
            this.subscriber = subscriber;
        }
    }
}

https://gist.github.com/MisterRager/3d40e4be8aa4624084528a49c803d6a5#file-rx2servicebindingfactory-java-L45 should be this.subscriber = observableEmitter;, right?

thanks

Hello!
Could you please provide an example of how to use your Rx2ServiceBindingFactory.java?

Context applicationContext = getApplicationContext(); // defined
Intent  dataServiceIntent  = getDataServiceIntent();  // defined

Rx2ServiceBindingFactory bindingFactory = 
new Rx2ServiceBindingFactory();

Observable<DataServiceBinder> dataServiceBinderObservable = 
bindingFactory
.bind(applicationContext, 
      dataServiceIntent);

dataServiceBinderObservable
.subscribe(binder -> /* ??? */);

What should I do next to call a method from DataService instance when connected to it?