CachedObservableをシャットダウンする方法

Question

私は無限のストリームを持っています（これはoncompleteとは全く呼ばれません）。そして、私はそれに最後の値をキャッシュ：

Observable<T> endlessStream = createStream().cache(); 

Subscription s1 = endlessStream.subscribe(...) 
Subscription s2 = endlessStream.subscribe(...) 

いくつかの条件によってendlessStreamは無効になり、私はそれを置き換える（switchMapと、それは重要ではありません）。

s1.unsubscribe() 
s2.unsubscribe() 

しかしCachedObservableは常に（createStream()から返された）ソースストリームへの接続を保存します。これはメモリリークを引き起こします。 CachedObservableをソースから観測する方法を切り離すには？

詳細情報：

CachedObservableソース観測（connection）にSerialSubscriptionを含むフィールドstateを含んでいます。 私は次のハックを呼び出すと、すべてがOKになり：

private void disconnectCachedObservable(CachedObservable<T> observable) { 
    try { 
     Field fieldState = CachedObservable.class.getDeclaredField("state"); 
     fieldState.setAccessible(true); 
     Object state = fieldState.get(observable); 
     Field fieldConnection = state.getClass().getDeclaredField("connection"); 
     fieldConnection.setAccessible(true); 
     SerialSubscription subscription = (SerialSubscription) fieldConnection.get(state); 
     subscription.unsubscribe(); 
    } catch (NoSuchFieldException e) { 
     e.printStackTrace(); 
    } catch (IllegalAccessException e) { 
     e.printStackTrace(); 
    } 
} 

しかし、反射が良い解決策ではありません:(

Answer 1

見つかりソリューション：

オペレータキャッシュは、オペレータリプレイ+ AUTOCONNECT（1と同様の振る舞いを持っていますあなたはあなたの契約を与えることによって、上流を停止してみましょう）。

replay().autoConnect(1, toStop -> { /* store Subscription to cancel later */ }); 

。