角度およびrxJS。ストリームは決して完了しません

Question

私は角型アプリケーションのメニューを持っており、このメニューはWebアプリケーションの選択されたクライアントを制御します。

それがあるとして：

• 私のメニューは、サービスを経由して、選択したクライアントを共有するコンポーネントです。
• 選択したクライアントはObservableとして共有されます。
• 観測可能なものは決して完了しません。メニューがクリックされたときに新しい選択されたクライアントを送出するだけです。
• これは再生の対象です。新しい各サブスクライバは、最後に発行されたクライアントを取得します。

それは良いデザインのようでしたが、私はクライアントに基づいて新しい観察可能なものを作成するときに（観察可能なものを完成させることはありません）いくつかの問題に遭遇しました。 AFAIKこれは、最初の観測が完了せず、このプロパティが伝播するためです。

//Command 
export interface Command { 
    productName: string; 
    qty: number; 
} 

//The service 
@Injectable() 
export class ClientService { 
private _client: ReplaySubject<Client> = new ReplaySubject(1); 

    setClient(client: Client) { // The menu component calls setClient on a user click 
     this._client.next(client); 
    } 
    getClient(): Observable<Client> { // getClient is heavilly called in child component to observe client selection events. 
     return this._client.asObservable(); 
    } 
} 


getCommands(): Observable<Command> { //Used in a template with async pipe 

    //In a child component 
    const commandsObs = this.clientService.getClient() 
    //.take(1) //I am forced to use take(1) to get getCommands() observer to complete 
    .flatMap(c => { 
     return Observable.merge(getCommandsPromise1(c), getCommandsPromise2(c)); 
    }) 
    .reduce((acc, next) => Object.assign({}, acc, next)) 
    .map(next => { 
     return finishMakingCommand(next)); 
    }) 
    .catch(err => /* Some error management code*/) 
} 

getCommandsPromise1(client: Client): Promise<any> { 
    //REST call returning a promise 
    return Promise.resolve({ 'command1': { 'productName': 'toy', qty: 1 } }); 
} 
getCommandsPromise2(client: Client): Promise<any> { 
    //REST call returning a promise 
    return Promise.resolve({ 'command2': { 'productName': 'another toy', qty: 1 } }); 
} 

finishMakingCommand(commands: any): Command[] { 
    // Flattens 'command1' and 'command2' to a list of commands 
    return [{'productName': 'toy', qty: 1}, {'productName': 'another toy', qty: 2}]; 
} 

私はより多くの経験を積んだ開発者が終わることのない、観察が良いデザインだと思うし、観測を終了することはありません回避するために、選択肢が何であるかどうかを知りたいのです。

Answer 1

私は前述したように、パイプラインの敷設などの観測を考えます。唯一の問題があるのは、水がいつも来ていることです（Observable.interval(1000)のように - それは絶え間なく進んでいます）。その状況でを手動で購読すると、購読を解除する必要があります。

しかし、あなたが言ったように、あなたは非同期パイプを使用しており、Angularはそこでの購読を中断します。

メニューをクリックすると、1回クリックすると1つの値が送信されます。観察可能な決して決して完了しない（すなわち、決して完了したイベントを受け取ることはない）のはよくあることですが、それは要件ではありません。

completedイベントは、通常、toArray()などの演算子を集計して、値のセット全体を知るのに便利です。

は、私はちょうど空想何かをするつもりだ場合を除き

<div>{{ commandObs | async }}</div> 

は、flatMapを忘れる

const commandsObs = this.clientService.getClient(); 

を使用して、テンプレート（例）に示唆している - 私に知らせてください。

編集 - あなたは、彼らが（Observable.merge）の内側に観察の結果を処理することを意図しているとして、flatmapの内側に削減し、マップ移動しようとすることができ、新たなサンプルコード

への変更を提案しました。

const commandsObs = this.clientService.getClient() 
    .flatMap(c => { 
    return Observable.merge(getCommandsPromise1(c), getCommandsPromise2(c)) 
     .reduce((acc, next) => Object.assign({}, acc, next)) 
     .map(next => finishMakingCommand(next)) 
    }); 

しようとする別のバージョン、

const commandsObs = this.clientService.getClient() 
    .map(c => Observable.forkJoin(getCommandsPromise1(c), getCommandsPromise2(c)); 

Answer 2

これを停止するには、サブスクライブ解除機能を使用できます。 例： mySubscribtion：Subscription;

this.mySubscription = somthing.subscribe((...) => {...}) 

、その後、あなたがイベントやその方法onDestroy unsubscibeことができます。

this.mySubscribtion.unsubscribe();