Создание пользовательского канала RXJS, который принимает наблюдаемое и ожидает, пока это наблюдаемое удовлетворяет условию - PullRequest
Новая
       49

Создание пользовательского канала RXJS, который принимает наблюдаемое и ожидает, пока это наблюдаемое удовлетворяет условию

0 голосов
/ 24 сентября 2019

У меня есть следующий код 

export function featureReady(feature: BaseFeatureService) {
  return pipe(
    zip(feature.loading$),
    filter(([inputObject, loading]) => !loading),
    map(([inputObject, loading]) => {
      return inputObject;
    })
  );
}

И я хочу использовать его следующим образом. 

observable$.pipe(
   featureReady(this.propertyFeatureService)
);

zip теперь устарел вместо статического zip, которыйозначает, что мое вышеупомянутое решение перестанет работать, есть ли оператор RXJS, который я могу заменить либо zip оператором, либо все решение в целом?

Я также был бы рад любому решению, которое принимает "Observable" для ожиданиясостояние, так как я бы не отказался передать feature.loading$

Спасибо.

Ответы [ 2 ]

0 голосов
/ 24 сентября 2019

лучше использовать конвейерный оператор skipUntil Rxjs, который пропускает значение до тех пор, пока внутренняя наблюдаемая не удовлетворяет условию.

В приведенном ниже коде источник Observable будет ждать, пока функция fun вернет пользовательский канал, который будет ждать, покаожидаемое состояние загрузки ложно

const { interval, timer,of,concat } = rxjs;
const {  skipUntil,delay,filter,ta } = rxjs.operators;


// Write TypeScript code!
const appDiv = document.getElementById('app');
appDiv.innerHTML = `<h1>Wait until the loading status is false</h1>`;

//custom pipe function which will  until loading
const fun = (wait1) =>  skipUntil(wait1.pipe(filter(status => !status['loading'])));

console.clear()
const status = [{loading:true},{loading:false}]
//emit every 1s
const source = interval(1000);

// inner obserbale which will emit the status after certain interval
const wait1 = concat(of(status[0]).pipe(delay(1000)),of(status[1]).pipe(delay(3000))).pipe(delay(5000));
//skip emitted values from source until inner observable emits false after the loading status turned to false
const example = source.pipe(fun(wait1));

const subscribe = example.subscribe(val => appDiv.innerHTML += 'interval ' + val + '<br/>');
const subscribe1 = wait1.subscribe(val => appDiv.innerHTML += 'inner Observable status ' + val.loading + '<br/>');
<script src="https://unpkg.com/rxjs/bundles/rxjs.umd.min.js"></script>

<div id="app"></div>

Вы можете найти решение stackBlitz .

0 голосов
/ 24 сентября 2019

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

Теперь он выглядит так: 

export function featureReady(feature: BaseFeatureService) {
  return pipe(
    withLatestFrom(feature.loading$),
    filter(([inputObject, loading]) => !loading),
    map(([inputObject, loading]) => {
      return inputObject;
    })
  );
}

Я оставлю вопрос, если у кого-то есть лучшее решение.

...