Использование takeUntil для завершения потока RxJS

Question

Я хочу, чтобы заполнить внутренний поток, который, естественно, не полный, с помощью оператора takeUntil, как это:

outerObservable 
    .mergeMap(() => { 
     innerObservable 
      .takeUntil(stopObservable$) 
    }); 

Это работает, внутренний поток завершает, как и ожидалось, но Я хотел бы, чтобы внешний поток возвращал одно последнее значение после сигнала останова. Даже после многих поисковых запросов я до сих пор не знаю, как это сделать.

EDIT:

Я написал оператор, который, кажется, сделать трюк, но оставив открытым вопрос, потому что я знаю, что есть лучший путь или что-то я полностью недоразумение.

function takeUntilThen(notifier, oneLastValue) { 
 
    return Rx.Observable.create(subscriber => { 
 
     var source = this; 
 
     notifier.subscribe(() => { 
 
      subscriber.next(oneLastValue); 
 
      subscriber.complete() 
 
     }); 
 
     return source.subscribe(value => { 
 
      subscriber.next(value); 
 
     }, 
 
     err => subscriber.error(err), 
 
     () => subscriber.complete()); 
 
    }); 
 
}

Answer 1

Похоже, вы хотите, чтобы мчаться между innerObservable и stopObservable и выигрывает Какой должна быть в состоянии вывести что-то.

Для этого используется оператор aptly named .race(), а не .takeUntil(). Смотрите пример в рецепте DOCS Redux-наблюдаемый на Отмена:

https://redux-observable.js.org/docs/recipes/Cancellation.html#cancel-and-do-something-else-eg-emit-a-different-action

const somethingEpic = action$ => 
    action$.ofType(SOMETHING) 
    .mergeMap(action => 
     innerObservable 
     .race(
      stopObservable 
      .take(1) 
     ) 
    ); 

Поскольку ваш пример псевдо-код, вот один, что более конкретно:

import { ajax } from 'rxjs/observable/dom/ajax'; 

const fetchUserEpic = action$ => 
    action$.ofType(FETCH_USER) 
    .mergeMap(action => 
     ajax.getJSON(`/api/users/${action.payload}`) 
     .map(response => fetchUserFulfilled(response)) 
     .race(
      action$.ofType(FETCH_USER_CANCELLED) 
      .map(() => incrementCounter()) 
      .take(1) 
     ) 
    );