Fix connector archetype test

smallrye · Nov 13, 2023 · e8dae05 · e8dae05
1 parent 386fc06
commit e8dae05
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/.../main/resources/archetype-resources/src/main/java/__connectorPrefix__IncomingChannel.java b/.../main/resources/archetype-resources/src/main/java/__connectorPrefix__IncomingChannel.java
@@ -1,5 +1,6 @@
 package ${package};
 
+import java.util.Objects;
 import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -38,8 +39,9 @@ public class ${connectorPrefix}IncomingChannel {
         this.failureHandler = ${connectorPrefix}FailureHandler.create(this.client);
         this.tracingEnabled = cfg.getTracingEnabled();
         Multi<? extends Message<?>> receiveMulti = Multi.createBy().repeating()
-                .uni(() -> Uni.createFrom().completionStage(this.client.poll()))
+                .uni(() -> Uni.createFrom().completionStage(this.client == null ? null : this.client.poll()))
                 .until(__ -> closed.get())
+                .filter(Objects::nonNull)
                 .emitOn(context::runOnContext)
                 .map(consumed -> new ${connectorPrefix}Message<>(consumed, ackHandler, failureHandler));