Перезапустите сагу о MassTransit

В моем приложении пользователь может запускать обновления для определенного ресурса. Когда это происходит, отправляется событие UpdateInitiated.

Сага MassTransit запускается на основе этого события и запускает серию других событий, пока не сочтет себя выполненным и не завершится.

Пока идет обновление, пользователи могут вносить изменения в ресурс и запрашивать новое обновление. Я пытаюсь понять, как я могу прервать текущую сагу и создать новый экземпляр, когда новое событие UpdateInitiated отправляется для того же ресурса.

Я думаю о двух подходах

  1. Корреляция по ResourceId. Таким образом, первоначальная сага также получит второе событие UpdateInitiated. Но тогда я не знаю, как сбросить сагу (и перестать прислушиваться к событиям, связанным с предыдущим обновлением).

  2. Используйте CorrelatedBy<Guid> (как я сейчас делаю). Это создаст новую сагу для нового события обновления. Но тогда я не знаю, как отменить начальную сагу.

Есть ли способ прослушать произвольные события в саге? Например: прослушивать все UpdatedInitiated события, даже если они не коррелируют, и отменить текущую сагу, если есть событие с совпадающим ResourceId.


masstransit automatonymous
person Joel    schedule 06.05.2021    source источник

Ответы (1)


arrow_upward
0
arrow_downward

Я понял, что можно зарегистрироваться на события, где вы коррелируете с чем-то другим, кроме CorrelationId. В моем случае использования я хочу прервать текущую сагу, если есть новое событие, запускающее новую сагу, для того же ResourceId.

Я сделал вот что:

// Listen to update initiated events for the same ResourceId
Event(() => UpdateInitiated , e => e
    .CorrelateBy((instance, context) =>
        instance.CurrentState != 2 &&
        instance.CorrelationId != context.Message.CorrelationId &&
        instance.ResourceId == context.Message.ResourceId
    )
    .SelectId(x => NewId.NextGuid()));

// Finalize this saga if there is a newer one for the same ResourceId
DuringAny(
    When(UpdateInitiated)
        .Then(context => Log.Information("Saga aborted due to newer update initiated by {correlationId}", context.Data.CorrelationId))
        .Finalize());
person Joel    schedule 18.05.2021