Question about where I/O and subscribe are run

[hishamnajemrel]

This is all within the context of a Quarkus application.

Say I have a synchronous DB call:

public List getEnabledChecks(String orgId) {
return list("organizationReference.organizationId = ?1 and isEnabled = true", orgId).stream()
.map(CheckConfigurationEntity::getName)
.toList();
}

I make it async by doing:

public Uni<List> getEnabledChecksAsync(String orgId) {
var future = CompletableFuture.supplyAsync(() -> getEnabledChecks(orgId), managedExecutor);
return Uni.createFrom().completionStage(future);
}

@Inject
@ManagedExecutorConfig(propagated = ThreadContext.CDI)
ManagedExecutor managedExecutor;

Now, in a different CDI controlled class that is @ApplicationScoped, I consume Messages and depending on the message, I do some functionality: both checkConfigurationService.getEnabledChecksAsync and handleDefaultStatesForNewCar are async and run on a managedexecutor

private void processCarEvent(PortalEvent event) {
switch (event.eventType()) {
case CREATED -> {
checkConfigurationService
.getEnabledChecksAsync(event.organizationId())
.onItem().transformToUni(checks -> {
List<Uni> saveOperations = checks.stream()
.map(checkName ->
checkStateManager.handleDefaultStatesForNewCar(
event.organizationId(),
checkName,
CheckEnabledStateTransition.DISABLED_TO_ENABLED)
)
.toList();
return Uni.combine().all().unis(saveOperations).discardItems();
})
.emitOn(managedExecutor)
.subscribe().with(
result -> {
LOG.debug("Car default states added for CAR CREATED event");
},
failure -> {
// Log or handle error
LOG.error("Failed to process CAR CREATED event: {}", failure.getMessage());
}
);
}

  My question is, I know the I/O work is being done on the managed executor, where is the subscribe portion being done? If I am emitting on the managedExecutor, is it being done there? If I don't emit onto the managedExecutor, will the function coming to an end mean that there will be no more thread context for the work to be done or is the calling thread (if it is still alive) be where the work is done? I reckon I don't need the emitOn, please help me clarify my understanding of where I/O, subscribed.with() are being run.

[jponge]

See https://quarkus.io/guides/quarkus-reactive-architecture

(and also, please format your markdown, especially code blocks)