Come creare la cache/versione hot di rx.Single?

Question

Il RxJava v1.0.13 ha introdotto il nuovo tipo di un Osservabile: rx. Singolo. Si adatta perfettamente al modello richiesta-risposta ma manca degli effetti collaterali standard che introducono operatori come doOnNext(). Quindi, è molto più difficile far accadere più cose come risultato.

La mia idea era di sostituire doOnNext() con più abbonamenti alla stessa istanza singola. Ma questo può causare il lavoro di underlaying più volte: una volta per ogni abbonamento. implementazione

Esempio rx.Single:

private class WorkerSubscribe<SomeData>() : Single.OnSubscribe<SomeData> { 
    override fun call(sub: SingleSubscriber<in SomeData>) { 
     try { 
      val result = fetchSomeData() 
      sub.onSuccess(result) 
     } catch(t: Throwable) { 
      sub.onError(t) 
     } 
    } 
} 

val single = Single.create<SomeData>(WorkerSubscribe()) 

Usage:

single.subscribe({}, {}) 
single.subscribe({}, {}) // Data is fetched for the second time 

E 'possibile creare un'istanza di unico che non sarà fetchSomeData() più volte, anche quando single.subscribe() è chiamato più volte, ma cache e restituisce lo stesso risultato?

Answer 1

È necessario RxJava Subject: BehaviorSubject o AsyncSubject

Answer 2

Si prega di controllare il cache() operator. Si suppone di memorizzare le emissioni dallo Observable e di riprodurle nei successivi Subscriber s.

Answer 3

È possibile creare BehaviorSubject/ReplaySubject/AsyncSubject e quindi chiamare toSingle su di esso.

Answer 4

ho fatto soluzione, che io non sono felice con lui ma funziona:

public class Network { 
    private Object data; 
    private Single<Object> dataSingle; 

    public Single<Object> getData { 
     if (data == null) { 
     if (dataSingle == null) { 
      dataSingle = Single.create(...) 
      .doOnSuccess(data -> this.data = data;) 
      .sibscribeOn(..); 
     } 
     return dataSingle; 
     } else { 
     return Single.just(data); 
     } 
    } 
} 

Answer 5

ho solo bisogno di un comportamento simile e ho trovato qualche soluzione.

È possibile convertire Single a Observable applicare cache() e quindi riconvertirlo a Single.

yourSingle.toObservable().cacheWithInitialCapacity(1).toSingle() 

io uso cacheWithInitialCapacity(1) invece di cache() come un'ottimizzazione - Single non potrà mai emettere più di una voce.

 

E 'anche buona idea quella di fornire Transformer realizzazione

public class SingleUtils { 

    public static <T> Single.Transformer<T, T> cached() { 
     return single -> single.toObservable() 
      .cacheWithInitialCapacity(1) 
      .toSingle(); 
    } 
} 

in modo da poter utilizzare la cache dove vuoi chiamando semplicemente

yourSingle.compose(SingleUtils.cached()) 

Edit: A partire da rxJava 1.2.2 è stato aggiunto (https://github.com/ReactiveX/RxJava/releases/tag/v1.2.2)

Implementato esattamente in questo modo (https://github.com/ReactiveX/RxJava/pull/4757)