fix comments

quarkusio · Jul 6, 2023 · 3a53c7c · 3a53c7c
1 parent 0badc7c
commit 3a53c7c
Showing 1 changed file with 4 additions and 4 deletions.
diff --git a/...rc/main/java/io/quarkus/smallrye/reactivemessaging/runtime/QuarkusWorkerPoolRegistry.java b/...rc/main/java/io/quarkus/smallrye/reactivemessaging/runtime/QuarkusWorkerPoolRegistry.java
@@ -84,8 +84,6 @@ public void run() {
                         }
                     };
                 } catch (InvocationTargetException | IllegalAccessException | NoSuchMethodException e) {
-                    System.err.println(e);
-                    //quite ugly but works
                     logger.warnf("You weren't able to create an executor that spawns virtual threads, the default" +
                             " blocking executor will be used, please check that your JDK is compatible with " +
                             "virtual threads");
@@ -114,8 +112,9 @@ public <T> Uni<T> executeWork(Context currentContext, Uni<T> uni, String workerN
                 return currentContext.executeBlocking(Uni.createFrom().deferred(() -> uni), ordered);
             }
             return executionHolder.vertx().executeBlocking(uni, ordered);
-        } else if (virtualThreadConcurrency.containsKey(workerName)) {
-            Semaphore semaphore = virtualThreadConcurrency.get(workerName);
+        }
+        Semaphore semaphore = virtualThreadConcurrency.get(workerName);
+        if (semaphore != null) {
             boolean acquired = semaphore.tryAcquire();
             if (acquired) {
                 return runOnVirtualThread(currentContext, uni, semaphore);
@@ -151,6 +150,7 @@ public <T> Uni<T> executeWork(Context currentContext, Uni<T> uni, String workerN
 
     private <T> Uni<T> runOnVirtualThread(Context currentContext, Uni<T> uni, Semaphore semaphore) {
         return uni.runSubscriptionOn(command -> VIRTUAL_EXECUTOR_SUPPLIER.get().execute(command))
+                .onCancellation().invoke(semaphore::release)
                 .onItemOrFailure().transformToUni((item, failure) -> {
                     semaphore.release();
                     return Uni.createFrom().emitter(emitter -> {