Wie kann ich meine eigene Observable anpassen?

Question

Ich benutze Retrofit2, Rxjava2 und Adapter-Rxjava, um meinen http-API-Anruf zu implementieren.

//API definition 
Observable<String> queryProducts(@Body Query query); 

//API implementation. 
serviceApi.queryProducts(query) 
       .subscribeOn(new Scheduler().ioThread()) 
       .observeOn(new Scheduler().mainThread()) 
       .subscribe(new Observer()); 

Wenn ich eine Menge apis umgesetzt werden müssen, und jeder einzelne api Umsetzung muss diese beiden Zeilen hinzuzufügen:

.subscribeOn(new Scheduler().ioThread()) 
.observeOn(new Scheduler().mainThread()) 

Ich will sie nicht in jeder api Implementierung hinzufügen . Ich möchte MyObservable als Ergebnistyp meiner API-Definition verwenden.

Meine Idee sieht unten wie:

//API definition 
MyObservable<String> queryProducts(@Body Query query); 

//MyObservable definition 
    public class MyObservable<T> extends Observable<T> { 
     /** 
     * Creates an Observable with a Function to execute when it is subscribed to. 
     * <p> 
     * <em>Note:</em> Use {@link #create(OnSubscribe)} to create an Observable, instead of this constructor, 
     * unless you specifically have a need for inheritance. 
     * 
     * @param f {@link OnSubscribe} to be executed when {@link #subscribe(Subscriber)} is called 
     */ 
     protected MyObservable(OnSubscribe<T> f) { 
      super(f); 
      this.subscribeOn(new Scheduler().ioThread()); 
      this.observeOn(new Scheduler().mainThread()); 
     } 
    } 

Wenn ich es laufen, ich unten Ausnahme bekam:

java.lang.IllegalArgumentException: Kann Anruf Adapter für MyObservable erstellen.

Ich verfolgt RxJavaCallAdapterFactory.java Code bei https://github.com/square/retrofit/blob/master/retrofit-adapters/rxjava/src/main/java/retrofit2/adapter/rxjava/RxJavaCallAdapterFactory.java. Ich habe RxJavaCallAdapterFactory in Zeile 100 gefunden, es scheint, dass es nur die Observable-Klasse diesen Checkpoint passieren lässt. Ich konnte diese Methode nicht erweitern und überschreiben, da diese Klasse eine letzte Klasse ist.

if (rawType != Observable.class && !isSingle && !isCompletable) { 
     return null; 
} 

Gibt es eine Möglichkeit, diese beiden Zeilen in einer Superklasse hinzufügen, ich will sie nicht in jeder api Implementierung hinzufügen? Vielen Dank.

Answer 1

Während in RxJava2 Sie können safely extend Observable, es weniger wahrscheinlich eignet sich für diese Art von Situation, in der Sie gemeinsamen Code wiederverwenden möchten, anstatt es zu duplizieren (aber zum Erstellen von Observable von Grund auf neu, in der Regel zum Umschließen von externen asynchronen Callback-Code).
Stattdessen können Sie compose() Operator verwenden, die Observable mit Ihrem benutzerdefinierten Code transformieren, und es ist klassisch zum Hinzufügen von gemeinsamer Logik zu Observable.
Sie können Dan Lew's article für ein Beispiel genau für Ihren Bedarf folgen (Hinzufügen Schedulers).

Nachrüstadapter, da er Dienste mit Reflektion erstellt, kann er keine benutzerdefinierten Klassen unterstützen, generiert jedoch die vorhandenen RxJava-Klassen.
BTW, Sie suchen/mit Retrofit RxJava1 Adapter mit RxJava2, müssen Sie auf RxJava2 Adapter wechseln, mit RxJava2 Adapter können Sie sehen, dass tatsächlich Retrofit verwendet seine eigenen benutzerdefinierten Observable Klassen.

Wenn compose() mit nicht genug für Sie ist (wie Sie es immer noch zu jeder API hinzufügen müssen), die offizielle Weg, um es Ihre eigene CallAdapter.Factory und implementieren CallAdapter Nachrüstung zu schaffen durch Umwickeln RxJava2CallAdapterFactory, um es zu delegieren anzupassen, und dann wickeln die Rückkehr Observable mit Ihrem kundenspezifischen Code/operators/scheulers. Siehe hierzu tutorial. oder Beispiel mit RxJava2 (ziemlich gleich) bei einigen library I'm working on.