RxJS - Skip functions if some condition is met

Issue

In an Angular Web-Application I am working on, I need to do a lot of things using Observables.
Usually Observables are used for sequences of asynchronous events, however, for me, it is usually only 1 result (coming from a http-request).
I often need to do some complex mapping to get the desired result out of the http-requests. However, this mapping is only needed, if some condition is met, else I could skip the whole mapping part and return a default value.
For example:
I want to get the sum of the sales of a specific article in a given city.
Following steps are required to get the result:

1. Get all clients, which live in that given city
   city.getClients() : Observable<Array<Client>> (http GET -> Observable<Array<Client>>)
2. Determine the sales of the given article for every client
   (flatMap -> Observable<Array<ClientSale>>)
3. Get the total amount of sales of the given article for every client.
   (map -> Observable<Array<number>>)
4. Sum it up
   (map -> Observable<number>)

Now it is possible, that there are no clients living in that given city.
In that case, steps 2-4 could be skipped and the onNext-callback could be called with the value 0.

Note, that this is just a simple example, to explain what I want to do. The real use-cases require more steps and I would need to do a

if (meetsCondition(x))
    return map(x);
else
    return defaultValue;

in every step.

The filter-function is almost what I am looking for. However it does not call the onNext for the filtered values.

Is there a way to achieve what I am looking for?

EDIT:
I'll try to add some more detail to the question:
I have a lot of lazy-loaded data in my application. So if you request a city, it won't load the related clients until you request them by calling getClients.
Therefore most of the getters return an Observable of the data instead of the data itself.

Now I often need to calculate things, by combining values of different sources (client, article etc.).
Often this calculation could be stoped in the middle, as no valid result could be determined.
This could be the case, if a value is empty, but also if a filtered list contains more then one entry.

So the perfect solution for me would be, if I am able to do something like this:

client.getOrders()
   .skipIf(value => !meetsCondition(value), defaultValue)
   .map(value => mapValue(value))        // Probably not called
   .map(value => mapValueAgain(value))   // Probably not called
   .subscribe(value => {
        if (value == defaultValue)
            console.log("Skip");
        else
            console.log(value)
   });

Solution

I constructed your case like the following. Right?

function getClients(city:string) {
    return Observable.of(city == 'Seoul' ? ['c1', 'c2'] : []).delay(100);
}
function getSales(client:string, article:string) {
    let sales:any = {
        'c1': {
            'item1': 10,
            'item2': 20
        },
        'c2': {
            'item1': 30,
            'item2': 40
        }
    };

    return Observable.of(sales[client][article]).delay(100);
}

let city = 'Seoul';
let article = 'item2';

getClients(city)
    .mergeMap(clients => {
        console.log('>> mergeMap');
        let salesObs = clients.map<Observable<number>>(client => getSales(client, article));
        return Observable.merge(salesObs).mergeAll();
    })
    .reduce((sum, sales) => {
        console.log('>> reduce');
        return sum + sales;
    }, 0)
    .subscribe(value => console.log(`sum = ${value}`));

No problems though clients is empty in this case. Change city to another value and try.

I added some codes to work in general conditions using 'map' and 'catch'. If it meets some skip condition, 'mergeMap' and 'reduce' is not excuted.

getClients(city)
    .map(clients => {
        // some skip condition
        if (clients.length == 0)
            throw 0 /* default value */;

        return clients;
    })
    .mergeMap(clients => {
        console.log('>> mergeMap');
        let salesObs = clients.map<Observable<number>>(client => getSales(client, article));
        return Observable.merge(salesObs).mergeAll();
    })
    .reduce((sum, sales) => {
        console.log('>> reduce');
        return sum + sales;
    }, 0)
    .catch((err, caught) => {
        console.log('>> catch');
        // default value
        return Observable.of(err);
    })
    .subscribe(value => console.log(`sum = ${value}`));

Answered By - thatseeyou