Как издать эпос только в том случае, если другой эпик закончен?

Question

У меня есть 2 объекта: получатель платежа и транзакция. У меня есть payeeCreateEpic, который фильтрует на started испускает обычную цепочку finished / failed. То же самое для transactionCreateEpic.

Для простоты предположим, что Получатель платежа имеет только UUID и Имя. Однако транзакция может иметь одну из 3-х возможностей: не привязана ни к одному получателю (полезная нагрузка имеет payee: null), присоединена к существующему получателю (в пользовательском интерфейсе у пользователя есть выпадающий список всех получателей, где он выбирает одну, полезная нагрузка, имеющая payee: payee-uuid ) или прикреплен к не существующему получателю (в пользовательском интерфейсе пользователь имеет возможность создать получателя, который не сохраняется в бэкэнде, но должен быть сохранен, если вся транзакция сохранена, в полезной нагрузке, которую я отправляю payee: payee-name).

Теперь, если транзакция создания отправлена ​​без получателя или существующего получателя, я просто перехожу на POST /transactions и жду его разрешения, чтобы выполнить действие finished или failed действия transactionCreateEpic.

Однако, если !isUuid(txPayload.payee) оценивается как true, в эпосе transactionCreateEpic я хочу инициировать payeeCreateEpic с именем получателя и ждать finished =>, затем создать транзакцию с идентификатором получателя из бэкэнд или failed => прервать транзакцию (и отобразить ошибку для пользователя).

Я не уверен, как это сделать, поскольку в эпосе транзакции я уже в фильтре для действия TRANSACTION_CREATE_STARTED и не уверен, как подписаться на другое действие, генерируемое payeeCreateEpic.

Код:

const createPayeeEpic = (actions$: Observable<Action>) =>
    actions$.pipe(
        filter(CreateAction.start.match),
        mergeMap((action) =>
            from(MoneyPinApiClient.getInstance().payee.create(CreateRequestAdapter(action.payload))).pipe(
                map((response) => CreateAction.success({
                    params: action.payload,
                    result: CreateResultAdapter(response.data)
                })),
                catchError((err) => of(
                    <any>CreateAction.failure({params: action.payload, error: err}),
                    <any>MoneyPinApiErrorAction(err)
                ))
            )
        )
    );
const createTransactionEpic = (actions$: Observable<Action>) =>
    actions$.pipe(
        filter(CreateAction.start.match),
        mergeMap((action) => {
            if(!isUuid(action.payload.payee) {
               **EMIT PayeeCreateAction.start({name: action.payload.payee})**
               **WAIT FOR PayeeCreateAction.success (or PayeeCreateAction.failure)**
               action.payload.payee = resultOf(PayeeCreateAction.success).id;
            }
            return from(MoneyPinApiClient.getInstance().transaction.create(CreateRequestAdapter(action.payload))).pipe(
                map((response) => CreateAction.success({
                        params: action.payload,
                        result: CreateResultAdapter(response.data)
                    })),
                catchError((err) => of(
                    <any>CreateAction.failure({params: action.payload, error: err}),
                    <any>MoneyPinApiErrorAction(err)
                ))
            )
        })
    );

Посмотрите на createTransactionEpic, вот где я с трудом могу инициировать createPayeeEpic + ожидание успешного сбоя.

Answer 1

Решение состоит в том, чтобы извлечь фактическое создание транзакции в некоторый фабричный метод:

const transactionCreateStreamFactory = (action: MoneyPinAction<CreateActionParams>) =>
from(MoneyPinApiClient.getInstance().transaction.create(CreateRequestAdapter(action.payload))).pipe(
    concatMap((response) => CreateAction.success({
        params: action.payload,
        result: CreateResultAdapter(response.data)
    })),
    catchError((err) => of(
        <any>CreateAction.failure({params: action.payload, error: err}),
        <any>MoneyPinApiErrorAction(err)
    ))
);

А затем эпопея создания транзакции на самом деле выглядит так:

const createTransactionEpic = (actions$: Observable<Action>) =>
actions$.pipe(
    filter(CreateAction.start.match),
    mergeMap((action) => {
        const {payee, payload} = action.payload;
        if (payee && payload.payee_id) {
            return merge(
                actions$.pipe(
                    filter(PayeeCreateAction.success.match),
                    filter((payeeResponse) => payeeResponse.payload.result.payee.id === payload.payee_id),
                    take(1),
                    mergeMap(() => transactionCreateStreamFactory(action)),
                ),
                actions$.pipe(
                    filter(PayeeCreateAction.failure.match),
                    filter((payeeResponse) => payeeResponse.payload.params.payload.id === payload.payee_id),
                    take(1),
                    map(() => CreateAction.failure({
                        params: action.payload,
                        error: new Error("failed to create payee")
                    }))
                )
            ).pipe(
                startWith(PayeeCreateAction.start({
                    notebook_id: action.payload.notebook_id,
                    payload: {id: payload.payee_id, name: payee}
                }))
            );
        } else {
            return transactionCreateStreamFactory(action);
        }
    })
);

Таким образом, если получатель платежа с идентификатором был задан во время создания, я бы запустил другой канал, который прослушивает действие payee.success или payee.failure, фильтрует соответствующее действие (в случае, если во время одновременного действия может появиться несколько действий этого типа) создать или что-то еще), прослушивает только один раз (take(1)) и объединяет это с фактическим эпосом создания транзакции, и весь поток начинается с действия payee.start, которое обрабатывается эпопеей получателя (и, в свою очередь, испускает успех / неисполнение действий получателя).

Answer 2

Ваша проблема похожа на синхронизацию потоков. Используйте один из шаблонов синхронизации потоков. Например. https://en.wikipedia.org/wiki/Monitor_(synchronization)