RxBus.java

package fr.creditagricole.cats.dcpa.bus;

import java.util.concurrent.TimeUnit;

import fr.creditagricole.cats.dcpa.bus.events.DefinirEtatClickableBoutonsEvent;
import fr.creditagricole.cats.dcpa.bus.events.Getter;
import fr.creditagricole.cats.dcpa.bus.events.ValiderModificationAutresImpotEvent;
import fr.creditagricole.cats.dcpa.data.model.PensionRetraite;
import rx.Observable;
import rx.Subscriber;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.subjects.PublishSubject;
import rx.subjects.SerializedSubject;
import rx.subjects.Subject;

/**
 * Created by etp7220 on 30/03/2017.
 */

public abstract class RxBus {

    private class AskedEvent {
        public final Object askedObject;

        public AskedEvent(Object askedObject) {
            this.askedObject = askedObject;
        }
    }

    private final Subject<Object, Object> bus = new SerializedSubject<>(PublishSubject.create());

    protected Observable<String> ecouterEvenement(String nomEvenement){
        return bus.filter(nomEvenement::equals).cast(String.class);
    }

    public <CLASS> Observable<CLASS> ecouterEvenement(Class<CLASS> theClass) {
        return bus.filter(o -> theClass.isInstance(o) || theClass.equals(o)).cast(theClass);
    }

    public void envoyerEvenement(Object event) {
        bus.onNext(event);
    }

    private Observable<Object> envoyerEvenementObservable(Object event) {
        return Observable.create(subscriber -> {
            envoyerEvenement(event);

            final Object _useless = new Object();
            subscriber.onNext(_useless);
            subscriber.onCompleted();
        });
    }

    public <CLASS> Observable<CLASS> get(Class<CLASS> theClass) {
        return Observable.zip(
                ecouterEvenement(theClass),
                envoyerEvenementObservable(new AskedEvent(theClass)),
                (neededObject, _useless) -> neededObject);
    }

    public <CLASS> Observable<Getter<CLASS>> repondreGet(Class<CLASS> theClass) {
        return ecouterEvenement(AskedEvent.class)//j'attend une demande de type CLASS
                .filter(askedEvent -> askedEvent.askedObject.equals(theClass))
                .map(o -> new Getter<CLASS>() {
                    //puis j'envoie au presenter qui la contiens un objet getter
                    //auquel il devra donner la pension retraite
                    @Override
                    public void get(CLASS value) {
                        envoyerEvenement(value); //lorsque cette pension retraite est obtenue, j'en avertis les autre via le bus
                    }
                });
    }


    public interface Getter<T> {
        void get(T value);
    }

}