Using RxJs to apply a `filter` to an Array of Observable<T>, and emitting the value as soon as it meets a certain condition

Question

Using RxJs to apply a `filter` to an Array of Observable<T>, and emitting the value as soon as it meets a certain condition

Currently working on an application built with Angular2 in TypeScript, utilizing rxjs 5.

Edit: Just to clarify, I am still relatively new to the Rx library and looking for the best practices. I did try to find some answers in the documentation before seeking help on SO.

Here's what I have:

class Result { constructor(public inError: boolean) { } }

const checks: Array<() => Observable<Result>> = [...];

This array consists of functions, each returning an observable that holds a Result object.

What I need:

I would like to 'map' this array into an Array<Observable<Result>>, by executing each function sequentially ...
... but I want to stop mapping as soon as the first Result.inError is true!

I've been trying various combinations of reduce, takeWith, contains, etc... The deferred nature of Observables is causing confusion.

Any assistance would be appreciated greatly!

typescript angular rxjs

Answer 1

Answer №1

To handle synchronous operations with your observables, you can easily follow this approach:

const results = [];
for (let i = 0; i < checks.length; i +=1) {
    checks[i]().subscribe(result => if (result.inError) {
        break;
    } else {
        results.push(checks[i]());
    });
}
// Results observables can be accessed here.

For asynchronous observers:

const results = [];
function callObservable(index) {
    checks[index]().subscribe(result => if (result.inError) {
            // Results observables accessible here.
        } else {
            results.push(checks[i]());
            callObservable(index + 1);
        })
}
callObservable(0);

However, these methods may not provide any significant benefits. Your observables are already invoked before they are added to the results array, potentially leading to unexpected values upon subsequent calls.

Answer 2

To handle synchronous operations with your observables, you can easily follow this approach:

const results = [];
for (let i = 0; i < checks.length; i +=1) {
    checks[i]().subscribe(result => if (result.inError) {
        break;
    } else {
        results.push(checks[i]());
    });
}
// Results observables can be accessed here.

For asynchronous observers:

const results = [];
function callObservable(index) {
    checks[index]().subscribe(result => if (result.inError) {
            // Results observables accessible here.
        } else {
            results.push(checks[i]());
            callObservable(index + 1);
        })
}
callObservable(0);

However, these methods may not provide any significant benefits. Your observables are already invoked before they are added to the results array, potentially leading to unexpected values upon subsequent calls.

Answer 3

Answer №2

After extensive research and investigation, a solution has been uncovered:

// Introducing a class that will be encapsulated in the observables
class Result { constructor(public isError: boolean) { } }

// Array of Observables that will emit instances of the Result class
// Thunk is used for lazy instantiation of Result classes only when necessary
const oResults : Array<() => Observable<Result>> = [
  Rx.Observable.of(() => new Result(false)),
  Rx.Observable.of(() => new Result(false)),
  Rx.Observable.of(() => new Result(true)),
  Rx.Observable.of(() => new Result(false))
];

// Custom implementation of an INCLUSIVE 'takeWhile' function (discovered on SO)
const takeWhileInclusive(source, predicate){
    return source.publish(co => co.takeWhile(predicate)
        .merge(co.skipWhile(predicate).take(1)));
}

// Filtering out results as anticipated
const res = takeWhileInclusive(
    Rx.Observable.merge(oResults).map( x => x()), 
    x => !x.isError
);

// Subscribing to the resulting stream to receive the first set of false Results
// followed by the first true result
res.subscribe(next => console.info("Next result", next));

Answer 4

After extensive research and investigation, a solution has been uncovered:

// Introducing a class that will be encapsulated in the observables
class Result { constructor(public isError: boolean) { } }

// Array of Observables that will emit instances of the Result class
// Thunk is used for lazy instantiation of Result classes only when necessary
const oResults : Array<() => Observable<Result>> = [
  Rx.Observable.of(() => new Result(false)),
  Rx.Observable.of(() => new Result(false)),
  Rx.Observable.of(() => new Result(true)),
  Rx.Observable.of(() => new Result(false))
];

// Custom implementation of an INCLUSIVE 'takeWhile' function (discovered on SO)
const takeWhileInclusive(source, predicate){
    return source.publish(co => co.takeWhile(predicate)
        .merge(co.skipWhile(predicate).take(1)));
}

// Filtering out results as anticipated
const res = takeWhileInclusive(
    Rx.Observable.merge(oResults).map( x => x()), 
    x => !x.isError
);

// Subscribing to the resulting stream to receive the first set of false Results
// followed by the first true result
res.subscribe(next => console.info("Next result", next));

Using RxJs to apply a `filter` to an Array of Observable<T>, and emitting the value as soon as it meets a certain condition

Answer №1

Answer №2

Similar questions

Retrieve the object property based on an array of indices

Exploring the weather API integration with Angular 4

Experiencing CORS problem in Ionic 3 when accessing API on device

The attribute 'modify, adjust, define' is not found in the 'Observable<{}>' type

Disallow negative numbers but allow decimals in HTML input

Monitoring Object Changes in Angular 4

Design a Dynamic Navigation Bar with Angular Material to Enhance User Experience

Ways to update column B in ag-Grid following a modification in column A

Unraveling HTML in TypeScript with Angular 2

Guide on incorporating Kendo UI typings into a TypeScript module

TS - deduce the specific type of a key value without receiving a union type

Stop users from logging in simultaneously on multiple systems

Exploring the Power of TailwindCss in Storybook 6 for Angular Development

Clearing dropdown values when navigating back on a page in Angular 6: A step-by-step guide

Typescript: The property 'userId' is not found within the 'unknown' type

The issue of Undefined TypeError arises when using Angular HttpInterceptor and injecting any service

The concealed [hidden] attribute in Angular2 triggers the display of the element after a brief delay

Issue with bidirectional binding on angular material slide toggle not functioning as anticipated (Angular 4)

What causes padding/margin when using overflow: hidden?

Updating a Record in Angularfire2