I/O關係可觀測量

Question

我希望隱藏通過觀察到的運營商，以便橫穿integrety被保持的數據的一部分。

例如，如果我有一個像{ private, public }數據對象和我想做的事：

1. 採摘public屬性，並將其傳遞到自定義流custom$
2. custom$後，已經完成了我想恢復關係，以便我有一個對象{ private, data }，其中數據是custom$流的結果。

我不想使用concatMap，因爲custom$可能會阻塞傳入的數據並使流緩慢。

我目前使用以下，但它似乎很繁瑣，感覺就像我做錯了什麼：

console.clear(); 
 

 
const input$ = new Rx.Subject(); 
 
const custom = data$ => data$.switchMap(
 
    ({ value, delay }) => Rx.Observable.of(`data ${value}`).delay(delay) 
 
); 
 

 
const out$ = input$.mergeMap(
 
    ({ secret, data }) => 
 
    Rx.Observable.zip(
 
     Rx.Observable.of(data).let(data$ => custom(data$)), 
 
     Rx.Observable.of(secret) 
 
    )); 
 

 

 
out$.subscribe(([data, secret]) => console.log(data, '=', secret)); 
 

 
input$.next({ secret: 1, data: { value: 1, delay: 4000 }}); 
 
input$.next({ secret: 2, data: { value: 2, delay: 1000 }}); 
 
input$.next({ secret: 3, data: { value: 3, delay: 3000 }});

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

有沒有更簡單/更理智的方式來做到這一點？

Answer 1

所有*Map運算符的第二個參數是一個結果選擇器，該結果選擇器將平坦化流的結果並允許您使用原始參數重新加入它。

const input$ = new Rx.Subject(); 
const custom = data$ => data$.switchMap(
    ({ value, delay }) => Rx.Observable.of(`data ${value}`).delay(delay) 
); 

const out$ = input$.mergeMap(
    ({ secret, data }) => Rx.Observable.of(data).let(custom), 
    ({secret}, newData) => ({secret, data: newData}) 
); 


out$.subscribe(({data, secret}) => console.log(data, '=', secret)); 

Answer 2

我想這應該做你想要什麼：

你的數據

const { Observable } = Rx; 

const init$ = Observable.from([ 
    { secret: 1, data: { value: 1, delay: 4000 }}, 
    { secret: 2, data: { value: 2, delay: 1000 }}, 
    { secret: 3, data: { value: 3, delay: 3000 }} 
]); 

流：

const public$ = init$.pluck('data'); 

const final$ = public$.last().withLatestFrom(init$).map(x => x[0]); 

如何檢查，如果你想要做什麼：

final$ 
    .do(console.log) 
    .subscribe(); 

輸出：

Object {value: 3, delay: 3000} 

你不如有沒有祕密數據使用public$安全地在這裏和那麼對於你來說，如果你想要最初（完整）的初始流的值，你可以使用final$。

下面是演示：https://codepen.io/anon/pen/VbJEvW?editors=0010