I'am new to reactive programming and try to persist my data into database from my service executed in a worker thread. The code is very simple, but nothing happen, nothing flushed.

The repository class :

@ApplicationScoped
public class GraviteeIndexRepository implements PanacheRepository<GraviteeIndex> {

    @Inject
    Mutiny.SessionFactory sf;

    public Uni<Void> persistAll(List<GraviteeIndex> graviteeIndices) {
       return  sf.withTransaction(session -> {

            Uni<Void> ret = session.persistAll(graviteeIndices);
            session.flush();
            return ret;
        });
    }
}

The service class:

@ApplicationScoped
public class GraviteeIndexService {

    private final GraviteeIndexRepository graviteeIndexRepository;

    public GraviteeIndexService(GraviteeIndexRepository graviteeIndexRepository) {
        this.graviteeIndexRepository = graviteeIndexRepository;
    }

    public void saveAll(List<GraviteeIndex> graviteeIndices) {
        Uni.createFrom().item(graviteeIndexRepository.persistAll(graviteeIndices)).await();

    }
}

The method saveAll() in service class just block until persist is done in repository class I presume.

My entity class if needed (I use lombok helpers) :

@Entity
@Table(name = "gravitee_request")
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class GraviteeIndex {

    @Id
    @Column(name = "_id")
    private String uuid;

    @Column(name = "_index")
    private String index;
}

Note : When I was in imperative mode (librairies not reactive in pom.xml and no Uni class used at all, the persist was ok. But for reactive code, I needed to change all, librairies and code)

thank you

I'am new to reactive programming and try to persist my data into database from my service executed in a worker thread. The code is very simple, but nothing happen, nothing flushed.

The repository class :

@ApplicationScoped
public class GraviteeIndexRepository implements PanacheRepository<GraviteeIndex> {

    @Inject
    Mutiny.SessionFactory sf;

    public Uni<Void> persistAll(List<GraviteeIndex> graviteeIndices) {
       return  sf.withTransaction(session -> {

            Uni<Void> ret = session.persistAll(graviteeIndices);
            session.flush();
            return ret;
        });
    }
}

The service class:

@ApplicationScoped
public class GraviteeIndexService {

    private final GraviteeIndexRepository graviteeIndexRepository;

    public GraviteeIndexService(GraviteeIndexRepository graviteeIndexRepository) {
        this.graviteeIndexRepository = graviteeIndexRepository;
    }

    public void saveAll(List<GraviteeIndex> graviteeIndices) {
        Uni.createFrom().item(graviteeIndexRepository.persistAll(graviteeIndices)).await();

    }
}

The method saveAll() in service class just block until persist is done in repository class I presume.

My entity class if needed (I use lombok helpers) :

@Entity
@Table(name = "gravitee_request")
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class GraviteeIndex {

    @Id
    @Column(name = "_id")
    private String uuid;

    @Column(name = "_index")
    private String index;
}

Note : When I was in imperative mode (librairies not reactive in pom.xml and no Uni class used at all, the persist was ok. But for reactive code, I needed to change all, librairies and code)

thank you

• quarkus
• reactive
• quarkus-panache

1 Answer 1

I think the simple answer is that you are calling .await and not .await().indefinitely(). It should look more like this:

public void saveAll(List<GraviteeIndex> graviteeIndices) {
    graviteeIndexRepository.persistAll(graviteeIndices).await().indefinitely();
}

It's usually better to add a time out though, or it might wait forever. You can do it with .atMost (instead of .indefinitely).

But, if you want to use reactive code with Quarkus, it's better to let Quarkus handle it by returning a reactive type. It should look like this:

public Uni<Void> saveAll(List<GraviteeIndex> graviteeIndices) {
    return graviteeIndexRepository.persistAll(graviteeIndices);
}

Quarkus now will recognize the reactive type and handle the result accordingly.

Also, reactive operations need to be chained. In your example, the flush operation is not going to be executed because it's not subscribed and the returned value is ignored. The correct code should be:

public Uni<Void> persistAll(List<GraviteeIndex> graviteeIndices) {
       return sf.withTransaction(session -> session
                    .persistAll(graviteeIndices)
                    // Now the flush runs after the persist and it's chained correctly
                    .call(session::flush)
       );
}

But, Hibernate will flush the operation at commit time. You can achieve the same result with:

public Uni<Void> persistAll(List<GraviteeIndex> graviteeIndices) {
    return sf.withTransaction(session -> session.persistAll(graviteeIndices));
}

This is a link to the Mutiny documentation: https://smallrye.io/smallrye-mutiny/latest/guides/rx/

It contains many more example on how to handle `Uni`s