RXJS错误处理并包装多个调用

如何解决RXJS错误处理并包装多个调用

是否有更好的方法在Rxjs中协调以下代码？ 并且从Javascript的角度Promise / Asyc / Await来看，考虑一些硬错误和软错误处理会更好。

调用对是硬依赖性，在错误中以及随后的错误处理程序检查中将其标记为true，然后将其重新引发。和软错误会添加到最终data.error中，以便根据软错误进行任何处理。

如何更好地做到这一点？

const { Observable } = require('rxjs');
const { map,filter,flatMap,concatMap } = require('rxjs/operators');

function testcallback(fail,callback) {
    if (fail == true) {
        setTimeout(function(){ callback(new Error({ message: "Operation failed" })); },100);       
    }
    else {
        setTimeout(function(){ callback(null,10); },100);
    }
}

function createData(param) {
    return Observable.bindNodeCallback(testcallback)(param);
}

function associateData(param) {
    return Observable.bindNodeCallback(testcallback)(param);
}

function removeData(param) {
    return Observable.bindNodeCallback(testcallback)(param);
}

function markAssociateSuccess(param) {
   return Observable.bindNodeCallback(testcallback)(param);
}

function removeOldData(param) {
    return Observable.bindNodeCallback(testcallback)(param);
}

function getAllData() {
    return Observable.of( {error:null,data:[333]});
}


function updateData(updateData) {
    let param = updateRequest.param;
    return createData(1 == param)
        .catch((err) => {
            err.errorStep = 'CREATE_DATA';
            err.hard = true;
            throw (err);
        }).flatMap(
            () => {
                return associateData(2 == param)
                    .catch((err) => {
                        err.errorStep = 'ASSOCIATE_DATA';
                        err.hard = true;
                        return removeData(3==param)
                            .map(() => {
                                throw (err);
                            });
                    });
            },(createDataResponse,associateDataResponse) => {
                return [createDataResponse,associateDataResponse];
            }
        )
        .flatMap(() => {
             if (updateData.markAssociationSuccess) {
                return markAssociateSuccess(4 == param);
            } else {
                return Observable.of({});
            }
        }).catch((err) => {
            if (err.hard) {
                throw (err);
            }
            err.errorStep = 'ASSOCIATE_SUCCESS';
            return Observable.of({ error: err });
        })
        .flatMap((data) => {

            if (data.error) {
                return Observable.of(data);
            }
            return removeOldData(5 == param);
        }).catch((err) => {
            if (err.hard) {
                throw (err);
            }
            err.errorStep = 'REMOVE_OLD_DATA';
            return Observable.of({ error: err });
        })
        .flatMap(
            function fetchData() {
                return getAllData();
            },function resultSelector(prevRes,{ error,data }) {
                if (error) {
                    return {};
                }
                if (prevRes.error) {
                    data.error = prevRes.error;
                }
                return data;
            }
        )
        .subscribe(
            function onNext(data) {
                console.log("Successful operation final data: " + data);
            },function onError(err) {
                console.log("Errored out" + JSON.stringify(err));
            },function onComplete() {
                console.log("Stream got completed");
            }
        );
}

const updateRequest = {
    param:1,markAssociationSuccess: true
}

updateData(updateRequest);