How to consume an event after transaction is over using Quarkus Event Bus?

user_6148340 0
Viewed 234

I'm using Quarkus Event Bus that's built on eclipse vert.x I have a transactional method in class A that invokes EventBus.send() which triggers a method annotated with @ConsumeEvent in class B

Is there an option to consume the event only after the transaction is complete?

I know there is an option to do that using Java EE Events https://javaee.github.io/tutorial/cdi-adv005.html

Is it possible to do so using Quarkus Event Bus?

event-bus vert.x quarkus
2 Answers

There is no such an option at moment as far as I know. Maybe we can change the event sending code slightly:

@Inject
private Mutiny.SessionFactory sessionFactory;

@Inject
private EventBus eventBus;

//@ReactiveTranactional
public Uni<Void> create(final Fruit fruit) {
    return sessionFactory.withTransaction((session, tx) -> session.persist(fruit))
        .invoke(() -> eventBus.send("test_topic", "some test message")));
}

 @ConsumeEvent("test_topic")
public Uni<Void> listEntities(String testMessage) {
    log.info("Printing all entities via: " + testMessage);
    repository.findAll().stream().map(TestEntity::toString).forEach(log::info);
}
user_23111450

I used another aproach sucessfully with CDI Events and Observers.

Obs. Quarkus 2.11.2.Final

https://quarkus.io/guides/cdi#events-and-observers

https://docs.jboss.org/weld/reference/latest/en-US/html/events.html#_transactional_observers

Ex:

  • Producer
import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.event.Event;
import javax.enterprise.inject.Any;
import javax.inject.Inject;
import javax.transaction.Transactional;

@ApplicationScoped
public class MatchService implements Serializable {

  @Inject @Any Event<Match> matchEvent;

  @Transactional
  public void createAsync(final CreateMatchDto dto) {

    try {
      final Match match = this.create(dto);
      this.matchEvent.fire(match);
    } catch (final Exception e) {
      e.printStackTrace();
    }
  }
}
  • Consumer
import static javax.enterprise.event.TransactionPhase.AFTER_SUCCESS;
import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.event.Observes;
import javax.inject.Inject;
import javax.transaction.Transactional;

@ApplicationScoped
public class AnalyzerMatchService implements Serializable {

  @Transactional(Transactional.TxType.REQUIRES_NEW)
  public void analyzeMatch(@Observes(during = AFTER_SUCCESS) final Match match) {

    try {

      System.out.println( match.getId() + " - " + match.toString());
    }
   }
}

Remember to use @Transactional(Transactional.TxType.REQUIRES_NEW) in the consumer method if you need database operations.

user_20359990
Related