Mutiny 0.15.0 integration

Brings integration to the new Multi / Uni logging operator. Uses the latest Vert.x client bindings aligned with Mutiny 0.15.0.
quarkusio · Apr 1, 2021 · 9c4dfd5 · 9c4dfd5
1 parent 542cbfa
commit 9c4dfd5
Show file tree

Hide file tree

Showing 6 changed files with 54 additions and 8 deletions.
diff --git a/bom/application/pom.xml b/bom/application/pom.xml
@@ -51,7 +51,7 @@
         <smallrye-jwt.version>2.4.4</smallrye-jwt.version>
         <smallrye-context-propagation.version>1.1.0</smallrye-context-propagation.version>
         <smallrye-reactive-streams-operators.version>1.0.13</smallrye-reactive-streams-operators.version>
-        <smallrye-reactive-utils.version>1.5.0</smallrye-reactive-utils.version>
+        <smallrye-reactive-utils.version>1.6.0</smallrye-reactive-utils.version>
         <smallrye-reactive-messaging.version>2.9.0</smallrye-reactive-messaging.version>
         <jakarta.activation.version>1.2.1</jakarta.activation.version>
         <jakarta.annotation-api.version>1.3.5</jakarta.annotation-api.version>
@@ -137,7 +137,7 @@
         <netty.version>4.1.49.Final</netty.version>
         <reactive-streams.version>1.0.3</reactive-streams.version>
         <jboss-logging.version>3.4.1.Final</jboss-logging.version>
-        <mutiny.version>0.14.0</mutiny.version>
+        <mutiny.version>0.15.0</mutiny.version>
         <kafka2.version>2.7.0</kafka2.version>
         <zookeeper.version>3.5.7</zookeeper.version>
         <!-- Scala is used by Kafka so we need to choose a compatible version -->

diff --git a/extensions/mutiny/deployment/src/main/java/io/quarkus/mutiny/deployment/MutinyProcessor.java b/extensions/mutiny/deployment/src/main/java/io/quarkus/mutiny/deployment/MutinyProcessor.java
@@ -19,14 +19,16 @@ public FeatureBuildItem registerFeature() {
 
     @BuildStep
     @Record(ExecutionTime.RUNTIME_INIT)
-    public void initExecutor(ExecutorBuildItem executorBuildItem, MutinyInfrastructure recorder) {
+    public void runtimeInit(ExecutorBuildItem executorBuildItem, MutinyInfrastructure recorder) {
         ExecutorService executor = executorBuildItem.getExecutorProxy();
         recorder.configureMutinyInfrastructure(executor);
     }
 
     @BuildStep
     @Record(ExecutionTime.STATIC_INIT)
-    public void configureDroppedExceptionHandlerAndThreadBlockingChecker(MutinyInfrastructure recorder) {
-        recorder.configureDroppedExceptionHandlerAndThreadBlockingChecker();
+    public void buildTimeInit(MutinyInfrastructure recorder) {
+        recorder.configureDroppedExceptionHandler();
+        recorder.configureThreadBlockingChecker();
+        recorder.configureOperatorLogger();
     }
 }
diff --git a/extensions/mutiny/deployment/src/test/java/io/quarkus/mutiny/deployment/test/MutinyTest.java b/extensions/mutiny/deployment/src/test/java/io/quarkus/mutiny/deployment/test/MutinyTest.java
@@ -103,6 +103,24 @@ public void testForbiddenUniBlocking() throws InterruptedException {
         Assertions.assertTrue(exception.getMessage().contains("The current thread cannot be blocked"));
     }
 
+    @Test
+    public void testOperatorLogging() {
+        TestingJulHandler julHandler = new TestingJulHandler();
+        Logger logger = LogManager.getLogManager().getLogger("io.quarkus.mutiny.runtime.MutinyInfrastructure");
+        logger.addHandler(julHandler);
+
+        AtomicReference<String> value = new AtomicReference<>();
+        bean.loggingOperator().subscribe().with(value::set);
+
+        Assertions.assertEquals("HELLO", value.get());
+
+        Assertions.assertEquals(2, julHandler.logRecords.size());
+        LogRecord logRecord = julHandler.logRecords.get(0);
+        Assertions.assertEquals("check.0 | onSubscribe()", logRecord.getMessage());
+        logRecord = julHandler.logRecords.get(1);
+        Assertions.assertEquals("check.0 | onItem(HELLO)", logRecord.getMessage());
+    }
+
     @ApplicationScoped
     public static class BeanUsingMutiny {
 
@@ -123,6 +141,12 @@ public Uni<String> droppedException() {
                     })
                     .onCancellation().call(() -> Uni.createFrom().failure(new IOException("boom")));
         }
+
+        public Uni<String> loggingOperator() {
+            return Uni.createFrom().item("hello")
+                    .onItem().transform(String::toUpperCase)
+                    .log("check");
+        }
     }
 
     private static class TestingJulHandler extends Handler {

diff --git a/extensions/mutiny/runtime/src/main/java/io/quarkus/mutiny/runtime/MutinyInfrastructure.java b/extensions/mutiny/runtime/src/main/java/io/quarkus/mutiny/runtime/MutinyInfrastructure.java
@@ -18,15 +18,17 @@ public void configureMutinyInfrastructure(ExecutorService exec) {
         Infrastructure.setDefaultExecutor(exec);
     }
 
-    public void configureDroppedExceptionHandlerAndThreadBlockingChecker() {
+    public void configureDroppedExceptionHandler() {
         Logger logger = Logger.getLogger(MutinyInfrastructure.class);
         Infrastructure.setDroppedExceptionHandler(new Consumer<Throwable>() {
             @Override
             public void accept(Throwable throwable) {
                 logger.error("Mutiny had to drop the following exception", throwable);
             }
         });
+    }
 
+    public void configureThreadBlockingChecker() {
         Infrastructure.setCanCallerThreadBeBlockedSupplier(new BooleanSupplier() {
             @Override
             public boolean getAsBoolean() {
@@ -42,4 +44,22 @@ public boolean getAsBoolean() {
             }
         });
     }
+
+    public void configureOperatorLogger() {
+        Logger logger = Logger.getLogger(MutinyInfrastructure.class);
+        Infrastructure.setOperatorLogger(new Infrastructure.OperatorLogger() {
+            @Override
+            public void log(String identifier, String event, Object value, Throwable failure) {
+                String log = identifier + " | ";
+                if (failure != null) {
+                    log = log + event + "(" + failure.getClass() + "(" + failure.getMessage() + "))";
+                } else if (value != null) {
+                    log = log + event + "(" + value + ")";
+                } else {
+                    log = log + event + "()";
+                }
+                logger.info(log);
+            }
+        });
+    }
 }
diff --git a/independent-projects/qute/pom.xml b/independent-projects/qute/pom.xml
@@ -40,7 +40,7 @@
         <version.jboss-logging>3.3.2.Final</version.jboss-logging>
         <version.surefire.plugin>3.0.0-M5</version.surefire.plugin>
         <version.nexus-staging-maven-plugin>1.6.8</version.nexus-staging-maven-plugin>
-        <version.smallrye-mutiny>0.14.0</version.smallrye-mutiny>
+        <version.smallrye-mutiny>0.15.0</version.smallrye-mutiny>
     </properties>
 
     <modules>

diff --git a/independent-projects/resteasy-reactive/pom.xml b/independent-projects/resteasy-reactive/pom.xml
@@ -49,7 +49,7 @@
         <nexus-staging-maven-plugin.version>1.6.8</nexus-staging-maven-plugin.version>
         <jboss-jaxrs-api_2.1_spec.version>2.0.1.Final</jboss-jaxrs-api_2.1_spec.version>
         <jakarta.json.version>1.1.6</jakarta.json.version>
-        <mutiny.version>0.14.0</mutiny.version>
+        <mutiny.version>0.15.0</mutiny.version>
         <smallrye-common-annotation.version>1.5.0</smallrye-common-annotation.version>
         <vertx.version>3.9.6</vertx.version>
         <rest-assured.version>4.3.3</rest-assured.version>