Управляйте подтверждением с помощью Mutiny при преобразовании сообщения в Multi <Message>

Question

Я пытаюсь преобразовать одно входное сообщение в несколько сообщений. У меня есть метод со следующей подписью:

@Incoming("CH_IN")
@Outgoing("CH_OUT")
Multi<Message<B>> process(Message<A> in) {

}

Класс A похож на:

class A {
    private List<String> ids

    // getters and setters
}

Для каждого id в A я хотел бы создать экземпляр B. Как я могу это сделать и справиться с подтверждением сообщения in?

Это упрощение того, что у меня есть, но я не уверен, что это правильный способ сделать это :

@Incoming("CH_IN")
@Outgoing("CH_OUT")
Multi<Message<B>> process(Message<A> in) {
    List<Message<B>> out = new ArrayList<>();

    // Code to iterate ids and create instances of Message<B>
    // In certain cases the out list will be empty

    return Multi.createFrom().iterable(out).on().completion(in::ack);
}

Это правильный способ подтверждения входящего сообщения?

Answer 1

Это упрощение того, что у меня есть, но я не уверен, что это правильный способ сделать это

Это будет работать до тех пор, пока вам не понадобится любые блокирующие вызовы в вашем прокомментированном разделе обработки, но это не идеально ИМХО:

• Стилистически смешивание императивного и реактивного кода в одном методе не очень аккуратно
• Функционально этот подход не позволит вам вызывать какой-либо блокирующий код в вашей обработке, если вам это нужно (сейчас или в будущем).

Вместо этого я бы рекомендовал что-то вроде:

return Multi.createFrom().item(in)
        .flatMap(m -> Uni.createFrom().item(m).repeat().atMost(5)) //Replace with your actual processing
        .on().completion(in::ack);

..., который сохраняет весь метод как одну реактивную цепочку и дает вам возможность flatMap() другим издателям, если вам нужно (например, если для вашей обработки требуется веб-вызов, вызов базы данных или какой-либо другой блокирующий ввод-вывод. )

Это правильный способ подтверждения входящего сообщения?

Если вы хотите, чтобы подтверждение происходило только по завершении обработки, да.

Единственный недостаток в том, что если catastrophi c сбой произошел на полпути обработки, возможно, некоторые из них могут быть обработаны до подтверждения in. Единственное другое разумное поведение - использовать вместо него on().subscribed(), что просто создает противоположную проблему (in будет подтверждено, но некоторые или все сообщения могут не быть отправлены на ваш исходящий канал.) Какой из них предпочтительнее, зависит от вашего использования -корпус.