Rx-Java: Создание настраиваемого Observable

Question

Я новичок в RxJava, и мне интересно, как я могу создать настраиваемый Observable? Давайте представим, я мог бы написать передачу DB-к-БД так:

srcDb.getObservable(Bean.class) 
    .sql(selectSql) 
    .params(selectParams) 
    .subscribe(
      trgDb.getSubscriber(Bean.class) 
       .sql(insertSql) 
    ); 

я уже может сделать это с Абонентом, но как я могу получить некоторую небольшую конфигурацию таким же образом, чтобы сам Наблюдаемый?

Answer 1

Возможно, я нашел приемлемый способ обойти это. Кажется, что мне нужно сделать, это двойное связывание вне самого экземпляра Observable. Например. Мне нужен DbObservable и DbOnSubscribe пару, которая рассчитывает друг на друга, что-то вроде этого:

DbObservable класс:

public class DbObservable<T> extends Observable<T> { 

    //Some parameter 
    private String sql; 

    protected DbObservable(DbOnSubscribe<T> onSub) { 
     super(onSub); 
    } 

    //Getter for DbOnSubscribe 
    public String getSql() { 
     return sql; 
    } 

    //Chain parameter modifier 
    public DbObservable<T> sql(String sql) { 
     this.sql = sql; 
     return this; 
    } 
} 

DbOnSubscribe класс:

public class DbOnSubscribe<T> implements Observable.OnSubscribe<T> { 

    private DbObservable<T> dbObservable; 

    @Override 
    public void call(Subscriber<? super T> subscriber) { 
     String sql = dbObservable.getSql(); //Access SQL param 
     subscriber.onNext((T) sql); //Use subscriber 
     subscriber.onCompleted(); 
    } 

    //Set back-reference 
    public void setDbObservable(DbObservable<T> dbObservable) { 
     this.dbObservable = dbObservable; 
    } 
} 

И, наконец, наш предполагается класс DbConnector:

public class DbConnector { 

    public DbObservable<String> getObservable() { 
     DbOnSubscribe<String> onSub = new DbOnSubscribe<String>(); 
     DbObservable<String> obs = new DbObservable<>(onSub); 
     onSub.setDbObservable(obs); 
     return obs; 
    } 
} 

Поэтому, когда я пробую ...

public class DbObservableTest { 

    public static void main(String[] args) { 
     DbConnector srcDb = new DbConnector(); 

     srcDb.getObservable() 
       .sql("some SQL") 
       .subscribe(System.out::println); 
    } 
} 

... это действительно работает! Он печатает «некоторый SQL».

Заключение

1. Если вы хотите быть супер-чистым и не против одного или 2 дополнительных строк кода, идти на строитель, как предложил Джоэл и Тассосом Bassoukos.
2. Если вы не боитесь немного более сложного кода (который всегда должен быть инкапсулирован где-то), и вы действительно хотите, чтобы эти параметры находились внутри вашего собственного наблюдаемого, вы можете попробовать двойное связывание.
3. опции?

Answer 2

Там в 2 способа сделать это:

Вариант № 1: есть собственные объекты, сделать конфигурацию, а затем иметь execute(), query() или toObservable(), который переключает домены:

srcDb 
.find(Bean.class) 
.sql(selectSql) 
.params(selectParams) 
.execute() 
.subscribe(
     trgDb.getSubscriber(Bean.class) 
      .sql(insertSql) 
); 

Option # 2: использовать .compose() повторно использовать общие операции:

srcDb 
.getObservable(Bean.class) 
.compose(addSQLParameters()) 
.subscribe(
     trgDb.getSubscriber(Bean.class) 
      .sql(insertSql) 
); 

<T> Transformer<T,T> addSQLParameters() { 
    return obs -> obs.sql(selectSql).params(selectParams); 
} 

Я хотел бы предложить вам использовать вариант # 1, поскольку это позволяет намного лучше управлять вашей частью кода.