From cadd5a487532759b8a948adab6a14f7a9668c3b9 Mon Sep 17 00:00:00 2001 From: Ozan Gunalp Date: Thu, 21 Sep 2023 14:13:28 +0200 Subject: [PATCH] Bump Smallrye RM from 4.9.0 to 4.10.1 --- bom/application/pom.xml | 2 +- docs/src/main/asciidoc/kafka.adoc | 62 ++++++++++++++++++- .../kafka/deployment/DotNames.java | 1 + ...allRyeReactiveMessagingKafkaProcessor.java | 15 ++++- .../deployment/DefaultSerdeConfigTest.java | 41 ++++++++++++ .../QuarkusMediatorConfigurationUtil.java | 19 ++++-- .../deployment/ReactiveMessagingDotNames.java | 2 + .../SmallRyeReactiveMessagingProcessor.java | 3 + .../deployment/WiringProcessor.java | 24 ++++++- .../runtime/QuarkusMediatorConfiguration.java | 28 +++++++-- .../main/java/io/quarkus/it/kafka/Fruit.java | 21 ++++++- .../io/quarkus/it/kafka/FruitProducer.java | 9 +-- .../quarkus/it/kafka/FruitSplitProducer.java | 23 +++++++ .../io/quarkus/it/kafka/PricesProducer.java | 21 ------- .../it/kafka/PricesRepeatableProducers.java | 22 +++++++ .../src/main/resources/application.properties | 6 +- .../quarkus/it/kafka/KafkaConnectorTest.java | 4 +- 17 files changed, 261 insertions(+), 42 deletions(-) create mode 100644 integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/FruitSplitProducer.java delete mode 100644 integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/PricesProducer.java create mode 100644 integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/PricesRepeatableProducers.java diff --git a/bom/application/pom.xml b/bom/application/pom.xml index f2e0e016d6614..b072d8b3533c9 100644 --- a/bom/application/pom.xml +++ b/bom/application/pom.xml @@ -63,7 +63,7 @@ 1.0.13 3.0.1 3.6.0 - 4.9.0 + 4.10.1 2.3.1 2.1.2 2.1.1 diff --git a/docs/src/main/asciidoc/kafka.adoc b/docs/src/main/asciidoc/kafka.adoc index 25e142198b021..a76c98778f8a5 100644 --- a/docs/src/main/asciidoc/kafka.adoc +++ b/docs/src/main/asciidoc/kafka.adoc @@ -1245,6 +1245,66 @@ Reciprocally, multiple producers on the same channel can be merged by setting `m On the `@Incoming` methods, you can control how multiple channels are merged using the `@Merge` annotation. ==== +Repeating the `@Outgoing` annotation on outbound or processing methods allows another way of dispatching messages to multiple outgoing channels: + +[source, java] +---- +import java.util.Random; + +import jakarta.enterprise.context.ApplicationScoped; + +import org.eclipse.microprofile.reactive.messaging.Outgoing; + +@ApplicationScoped +public class MultipleProducers { + + private final Random random = new Random(); + + @Outgoing("generated") + @Outgoing("generated-2") + double priceBroadcast() { + return random.nextDouble(); + } + +} +---- + +In the previous example generated price will be broadcast to both outbound channels. +The following example selectively sends messages to multiple outgoing channels using the `Targeted` container object, +containing key as channel name and value as message payload. + +[source, java] +---- +import jakarta.enterprise.context.ApplicationScoped; + +import org.eclipse.microprofile.reactive.messaging.Incoming; +import org.eclipse.microprofile.reactive.messaging.Outgoing; + +import io.smallrye.reactive.messaging.Targeted; + +@ApplicationScoped +public class TargetedProducers { + + @Incoming("in") + @Outgoing("out1") + @Outgoing("out2") + @Outgoing("out3") + public Targeted process(double price) { + Targeted targeted = Targeted.of("out1", "Price: " + price, + "out2", "Quote: " + price); + if (price > 90.0) { + return targeted.with("out3", price); + } + return targeted; + } + +} +---- + +Note that <> doesn't work for signatures using the `Targeted`. + +For more details on using multiple outgoings, please refer to the http://smallrye.io/smallrye-reactive-messaging/4.10.0/concepts/outgoings/[SmallRye Reactive Messaging documentation]. + === Kafka Transactions Kafka transactions enable atomic writes to multiple Kafka topics and partitions. @@ -2261,7 +2321,7 @@ See the xref:quarkus-reactive-architecture.adoc[Quarkus Reactive Architecture do == Channel Decorators -SmallRye Reactive Messaging supports decorating incoming and outgoing channels for implementing cross-cutting concerns such as monitoring, tracing or message interception. For more information on implementing decorators and message interceptors see the http://smallrye.io/smallrye-reactive-messaging/3.19.1/concepts/decorators/[SmallRye Reactive Messaging documentation]. +SmallRye Reactive Messaging supports decorating incoming and outgoing channels for implementing cross-cutting concerns such as monitoring, tracing or message interception. For more information on implementing decorators and message interceptors see the http://smallrye.io/smallrye-reactive-messaging/latest/concepts/decorators/[SmallRye Reactive Messaging documentation]. [[kafka-configuration]] == Configuration Reference diff --git a/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DotNames.java b/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DotNames.java index 66366a7095eb7..0a065e4bcedf1 100644 --- a/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DotNames.java +++ b/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DotNames.java @@ -28,6 +28,7 @@ final class DotNames { static final DotName PROCESSOR = DotName.createSimple(org.reactivestreams.Processor.class.getName()); static final DotName PROCESSOR_BUILDER = DotName.createSimple(org.eclipse.microprofile.reactive.streams.operators.ProcessorBuilder.class.getName()); static final DotName MULTI = DotName.createSimple(io.smallrye.mutiny.Multi.class.getName()); + static final DotName MULTI_SPLITTER = DotName.createSimple(io.smallrye.mutiny.operators.multi.split.MultiSplitter.class.getName()); static final DotName AVRO_GENERATED = DotName.createSimple("org.apache.avro.specific.AvroGenerated"); static final DotName AVRO_GENERIC_RECORD = DotName.createSimple("org.apache.avro.generic.GenericRecord"); diff --git a/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/SmallRyeReactiveMessagingKafkaProcessor.java b/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/SmallRyeReactiveMessagingKafkaProcessor.java index 1981a3ad43ff8..825c9550e3ff4 100644 --- a/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/SmallRyeReactiveMessagingKafkaProcessor.java +++ b/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/SmallRyeReactiveMessagingKafkaProcessor.java @@ -181,9 +181,11 @@ public void defaultChannelConfiguration( if (launchMode.getLaunchMode().isDevOrTest()) { if (!runtimeConfig.enableGracefulShutdownInDevAndTestMode) { List incomings = discoveryState.findRepeatableAnnotationsOnMethods(DotNames.INCOMING); + List outgoings = discoveryState.findRepeatableAnnotationsOnMethods(DotNames.OUTGOING); List channels = discoveryState.findAnnotationsOnInjectionPoints(DotNames.CHANNEL); List annotations = new ArrayList<>(); annotations.addAll(incomings); + annotations.addAll(outgoings); annotations.addAll(channels); for (AnnotationInstance annotation : annotations) { String channelName = annotation.value().asString(); @@ -221,7 +223,7 @@ void discoverDefaultSerdeConfig(DefaultSerdeDiscoveryState discovery, alreadyGeneratedDeserializers); } - for (AnnotationInstance annotation : discovery.findAnnotationsOnMethods(DotNames.OUTGOING)) { + for (AnnotationInstance annotation : discovery.findRepeatableAnnotationsOnMethods(DotNames.OUTGOING)) { String channelName = annotation.value().asString(); if (!discovery.isKafkaConnector(channelsManagedByConnectors, false, channelName)) { continue; @@ -428,6 +430,7 @@ private Type getOutgoingTypeFromMethod(MethodInfo method) { if ((isPublisher(returnType) && parametersCount == 0) || (isPublisherBuilder(returnType) && parametersCount == 0) || (isMulti(returnType) && parametersCount == 0) + || (isMultiSplitter(returnType) && parametersCount == 0) || (isCompletionStage(returnType) && parametersCount == 0) || (isUni(returnType) && parametersCount == 0)) { outgoingType = returnType.asParameterizedType().arguments().get(0); @@ -443,7 +446,8 @@ private Type getOutgoingTypeFromMethod(MethodInfo method) { || (isUni(returnType) && parametersCount == 1) || (isPublisher(returnType) && parametersCount == 1) || (isPublisherBuilder(returnType) && parametersCount == 1) - || (isMulti(returnType) && parametersCount == 1)) { + || (isMulti(returnType) && parametersCount == 1) + || (isMultiSplitter(returnType) && parametersCount == 1)) { outgoingType = returnType.asParameterizedType().arguments().get(0); } else if ((isProcessor(returnType) && parametersCount == 0) || (isProcessorBuilder(returnType) && parametersCount == 0)) { @@ -556,6 +560,13 @@ private static boolean isMulti(Type type) { && type.asParameterizedType().arguments().size() == 1; } + private static boolean isMultiSplitter(Type type) { + // raw type MultiSplitter is wrong, must be MultiSplitter + return DotNames.MULTI_SPLITTER.equals(type.name()) + && type.kind() == Type.Kind.PARAMETERIZED_TYPE + && type.asParameterizedType().arguments().size() == 2; + } + private static boolean isSubscriber(Type type) { // raw type Subscriber is wrong, must be Subscriber return DotNames.SUBSCRIBER.equals(type.name()) diff --git a/extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest.java b/extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest.java index 70a6d866a7019..c1ca6b2e4eb93 100644 --- a/extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest.java +++ b/extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest.java @@ -48,6 +48,7 @@ import io.smallrye.config.common.MapBackedConfigSource; import io.smallrye.mutiny.Multi; import io.smallrye.mutiny.Uni; +import io.smallrye.mutiny.operators.multi.split.MultiSplitter; import io.smallrye.reactive.messaging.MutinyEmitter; import io.smallrye.reactive.messaging.kafka.KafkaRecord; import io.smallrye.reactive.messaging.kafka.KafkaRecordBatch; @@ -66,6 +67,7 @@ private static void doTest(Config customConfig, Tuple[] expectations, Class.. List> classes = new ArrayList<>(Arrays.asList(classesToIndex)); classes.add(Incoming.class); + classes.add(Outgoing.class); DefaultSerdeDiscoveryState discovery = new DefaultSerdeDiscoveryState(index(classes)) { @Override Config getConfig() { @@ -2727,6 +2729,45 @@ void method2(JsonObject msg) { } + @Test + void repeatableOutgoings() { + Tuple[] expectations = { + tuple("mp.messaging.outgoing.channel1.value.serializer", "org.apache.kafka.common.serialization.StringSerializer"), + tuple("mp.messaging.outgoing.channel2.value.serializer", "org.apache.kafka.common.serialization.StringSerializer"), + tuple("mp.messaging.outgoing.channel3.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"), + tuple("mp.messaging.outgoing.channel4.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"), + tuple("mp.messaging.outgoing.channel5.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"), + tuple("mp.messaging.outgoing.channel6.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"), + }; + doTest(expectations, RepeatableOutgoingsChannels.class); + } + + private static class RepeatableOutgoingsChannels { + + @Outgoing("channel1") + @Outgoing("channel2") + String method1() { + return null; + } + + @Outgoing("channel3") + @Outgoing("channel4") + JsonObject method2() { + return null; + } + + enum T { + + } + + @Outgoing("channel5") + @Outgoing("channel6") + MultiSplitter method3() { + return null; + } + + } + @Test void channelNameContainingDot() { Tuple[] expectations = { diff --git a/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/QuarkusMediatorConfigurationUtil.java b/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/QuarkusMediatorConfigurationUtil.java index e7b2e6c856759..53834d1195f70 100644 --- a/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/QuarkusMediatorConfigurationUtil.java +++ b/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/QuarkusMediatorConfigurationUtil.java @@ -9,6 +9,7 @@ import static io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames.KOTLIN_UNIT; import static io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames.MERGE; import static io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames.OUTGOING; +import static io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames.OUTGOINGS; import static io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames.RUN_ON_VIRTUAL_THREAD; import static io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames.SMALLRYE_BLOCKING; import static io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames.TRANSACTIONAL; @@ -116,10 +117,12 @@ public static QuarkusMediatorConfiguration create(MethodInfo methodInfo, boolean incomingValues.addAll(getIncomingValues(methodInfo)); configuration.setIncomings(incomingValues); - String outgoingValue = getValue(methodInfo, OUTGOING); - configuration.setOutgoing(outgoingValue); + // We need to extract the value of @Outgoing and @Outgoings (which contains an array of @Outgoing) + List outgoingValues = new ArrayList<>(getValues(methodInfo, OUTGOING)); + outgoingValues.addAll(getOutgoingValues(methodInfo)); + configuration.setOutgoings(outgoingValues); - Shape shape = mediatorConfigurationSupport.determineShape(incomingValues, outgoingValue); + Shape shape = mediatorConfigurationSupport.determineShape(incomingValues, outgoingValues); configuration.setShape(shape); Acknowledgment.Strategy acknowledgment = mediatorConfigurationSupport .processSuppliedAcknowledgement(incomingValues, @@ -161,7 +164,7 @@ public Merge.Mode get() { } })); - configuration.setBroadcastValue(mediatorConfigurationSupport.processBroadcast(outgoingValue, + configuration.setBroadcastValue(mediatorConfigurationSupport.processBroadcast(outgoingValues, new Supplier() { @Override public Integer get() { @@ -176,6 +179,7 @@ public Integer get() { return null; } })); + configuration.setHasTargetedOutput(mediatorConfigurationSupport.processTargetedOutput()); AnnotationInstance blockingAnnotation = methodInfo.annotation(BLOCKING); AnnotationInstance smallryeBlockingAnnotation = methodInfo.annotation(SMALLRYE_BLOCKING); @@ -328,6 +332,13 @@ private static List getIncomingValues(MethodInfo methodInfo) { .collect(Collectors.toList()); } + private static List getOutgoingValues(MethodInfo methodInfo) { + return methodInfo.annotations().stream().filter(ai -> ai.name().equals(OUTGOINGS)) + .flatMap(outgoings -> Arrays.stream(outgoings.value().asNestedArray())) + .map(outgoing -> outgoing.value().asString()) + .collect(Collectors.toList()); + } + private static String fullMethodName(MethodInfo methodInfo) { return methodInfo.declaringClass() + "#" + methodInfo.name(); } diff --git a/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/ReactiveMessagingDotNames.java b/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/ReactiveMessagingDotNames.java index 384dcd3aee6ba..e78713e715f9b 100644 --- a/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/ReactiveMessagingDotNames.java +++ b/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/ReactiveMessagingDotNames.java @@ -23,6 +23,7 @@ import io.smallrye.reactive.messaging.annotations.Incomings; import io.smallrye.reactive.messaging.annotations.Merge; import io.smallrye.reactive.messaging.annotations.OnOverflow; +import io.smallrye.reactive.messaging.annotations.Outgoings; import io.smallrye.reactive.messaging.connector.InboundConnector; import io.smallrye.reactive.messaging.connector.OutboundConnector; import io.smallrye.reactive.messaging.keyed.KeyValueExtractor; @@ -38,6 +39,7 @@ public final class ReactiveMessagingDotNames { static final DotName INCOMING = DotName.createSimple(Incoming.class.getName()); static final DotName INCOMINGS = DotName.createSimple(Incomings.class.getName()); static final DotName OUTGOING = DotName.createSimple(Outgoing.class.getName()); + static final DotName OUTGOINGS = DotName.createSimple(Outgoings.class.getName()); public static final DotName CONNECTOR = DotName.createSimple(Connector.class.getName()); static final DotName CONNECTOR_ATTRIBUTES = DotName.createSimple(ConnectorAttributes.class.getName()); diff --git a/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/SmallRyeReactiveMessagingProcessor.java b/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/SmallRyeReactiveMessagingProcessor.java index d2ebac1e6fa28..aa82b90d8ac8c 100644 --- a/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/SmallRyeReactiveMessagingProcessor.java +++ b/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/SmallRyeReactiveMessagingProcessor.java @@ -159,6 +159,9 @@ public List removalExclusions() { new UnremovableBeanBuildItem( new BeanClassAnnotationExclusion( ReactiveMessagingDotNames.OUTGOING)), + new UnremovableBeanBuildItem( + new BeanClassAnnotationExclusion( + ReactiveMessagingDotNames.OUTGOINGS)), new UnremovableBeanBuildItem( new BeanClassAnnotationExclusion( ReactiveMessagingDotNames.MESSAGE_CONVERTER)), diff --git a/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/WiringProcessor.java b/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/WiringProcessor.java index 61a31a4f8f5e2..41fbd927a99a6 100644 --- a/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/WiringProcessor.java +++ b/extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/WiringProcessor.java @@ -107,15 +107,19 @@ void extractComponents(BeanDiscoveryFinishedBuildItem beanDiscoveryFinished, ReactiveMessagingDotNames.INCOMINGS); AnnotationInstance outgoing = transformedAnnotations.getAnnotation(method, ReactiveMessagingDotNames.OUTGOING); + AnnotationInstance outgoings = transformedAnnotations.getAnnotation(method, + ReactiveMessagingDotNames.OUTGOINGS); AnnotationInstance blocking = transformedAnnotations.getAnnotation(method, BLOCKING); - if (incoming != null || incomings != null || outgoing != null) { + if (incoming != null || incomings != null || outgoing != null || outgoings != null) { handleMethodAnnotatedWithIncoming(appChannels, validationErrors, configDescriptionBuildItemBuildProducer, method, incoming); handleMethodAnnotationWithIncomings(appChannels, validationErrors, configDescriptionBuildItemBuildProducer, method, incomings); handleMethodAnnotationWithOutgoing(appChannels, validationErrors, configDescriptionBuildItemBuildProducer, method, outgoing); + handleMethodAnnotationWithOutgoings(appChannels, validationErrors, configDescriptionBuildItemBuildProducer, + method, outgoings); if (WiringHelper.isSynthetic(method)) { continue; @@ -218,6 +222,24 @@ private void handleMethodAnnotationWithOutgoing(BuildProducer } } + private void handleMethodAnnotationWithOutgoings(BuildProducer appChannels, + BuildProducer validationErrors, + BuildProducer configDescriptionBuildItemBuildProducer, + MethodInfo method, AnnotationInstance outgoings) { + if (outgoings != null) { + for (AnnotationInstance instance : outgoings.value().asNestedArray()) { + if (instance.value().asString().isEmpty()) { + validationErrors.produce(new ValidationPhaseBuildItem.ValidationErrorBuildItem( + new DeploymentException("Empty @Outgoing annotation on method " + method))); + } + configDescriptionBuildItemBuildProducer.produce(new ConfigDescriptionBuildItem( + "mp.messaging.outgoing." + instance.value().asString() + ".connector", null, + "The connector to use", null, null, ConfigPhase.BUILD_TIME)); + produceOutgoingChannel(appChannels, instance.value().asString()); + } + } + } + private void handleMethodAnnotationWithIncomings(BuildProducer appChannels, BuildProducer validationErrors, BuildProducer configDescriptionBuildItemBuildProducer, diff --git a/extensions/smallrye-reactive-messaging/runtime/src/main/java/io/quarkus/smallrye/reactivemessaging/runtime/QuarkusMediatorConfiguration.java b/extensions/smallrye-reactive-messaging/runtime/src/main/java/io/quarkus/smallrye/reactivemessaging/runtime/QuarkusMediatorConfiguration.java index 605cede36f44e..f1fa6442420e3 100644 --- a/extensions/smallrye-reactive-messaging/runtime/src/main/java/io/quarkus/smallrye/reactivemessaging/runtime/QuarkusMediatorConfiguration.java +++ b/extensions/smallrye-reactive-messaging/runtime/src/main/java/io/quarkus/smallrye/reactivemessaging/runtime/QuarkusMediatorConfiguration.java @@ -31,7 +31,7 @@ public class QuarkusMediatorConfiguration implements MediatorConfiguration { private List incomings = new ArrayList<>(); - private String outgoing; + private List outgoings = new ArrayList<>(); private Acknowledgment.Strategy acknowledgment; @@ -63,6 +63,8 @@ public class QuarkusMediatorConfiguration implements MediatorConfiguration { private Type valueType; private Class keyed; + private boolean hasTargetedOutput = false; + public String getBeanId() { return beanId; } @@ -117,11 +119,16 @@ public void setIncomings(List incomings) { @Override public String getOutgoing() { - return outgoing; + return outgoings.get(0); + } + + @Override + public List getOutgoings() { + return outgoings; } - public void setOutgoing(String outgoing) { - this.outgoing = outgoing; + public void setOutgoings(List outgoings) { + this.outgoings = outgoings; } @Override @@ -310,4 +317,17 @@ public void setValueType(Type valueType) { public void setKeyed(Class keyed) { this.keyed = keyed; } + + @Override + public boolean hasTargetedOutput() { + return hasTargetedOutput; + } + + public boolean isHasTargetedOutput() { + return hasTargetedOutput; + } + + public void setHasTargetedOutput(boolean hasTargetedOutput) { + this.hasTargetedOutput = hasTargetedOutput; + } } diff --git a/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/Fruit.java b/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/Fruit.java index afe51f1d0551f..2eca3e8ac02f1 100644 --- a/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/Fruit.java +++ b/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/Fruit.java @@ -3,12 +3,31 @@ public class Fruit { public String name; + public Fruits type; - public Fruit(String name) { + public Fruit(String name, Fruits type) { this.name = name; + this.type = type; } public Fruit() { // Jackson will uses this constructor } + + public enum Fruits { + BERRY, + CITRUS, + POME, + STONE, + TROPICAL; + + @Override + public String toString() { + return "fruit-" + name().toLowerCase(); + } + + Fruit create(String name) { + return new Fruit(name, this); + } + } } diff --git a/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/FruitProducer.java b/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/FruitProducer.java index 251f621733494..8b9e5af30bd43 100644 --- a/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/FruitProducer.java +++ b/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/FruitProducer.java @@ -11,9 +11,10 @@ public class FruitProducer { @Outgoing("fruits-out") public Multi generateFruits() { return Multi.createFrom().items( - new Fruit("apple"), - new Fruit("banana"), - new Fruit("peach"), - new Fruit("orange")); + Fruit.Fruits.BERRY.create("strawberry"), + Fruit.Fruits.POME.create("apple"), + Fruit.Fruits.TROPICAL.create("banana"), + Fruit.Fruits.STONE.create("peach"), + Fruit.Fruits.CITRUS.create("orange")); } } diff --git a/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/FruitSplitProducer.java b/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/FruitSplitProducer.java new file mode 100644 index 0000000000000..e51f3089c680d --- /dev/null +++ b/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/FruitSplitProducer.java @@ -0,0 +1,23 @@ +package io.quarkus.it.kafka; + +import jakarta.enterprise.context.ApplicationScoped; + +import org.eclipse.microprofile.reactive.messaging.Incoming; +import org.eclipse.microprofile.reactive.messaging.Outgoing; + +import io.smallrye.mutiny.Multi; +import io.smallrye.mutiny.operators.multi.split.MultiSplitter; + +@ApplicationScoped +public class FruitSplitProducer { + + @Incoming("fruits-out") + @Outgoing("fruit-berry") + @Outgoing("fruit-citrus") + @Outgoing("fruit-pome") + @Outgoing("fruit-stone") + @Outgoing("fruit-tropical") + MultiSplitter produce(Multi fruits) { + return fruits.split(Fruit.Fruits.class, s -> s.type); + } +} diff --git a/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/PricesProducer.java b/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/PricesProducer.java deleted file mode 100644 index cd47f37c2a158..0000000000000 --- a/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/PricesProducer.java +++ /dev/null @@ -1,21 +0,0 @@ -package io.quarkus.it.kafka; - -import jakarta.enterprise.context.ApplicationScoped; - -import org.eclipse.microprofile.reactive.messaging.Outgoing; - -import io.smallrye.mutiny.Multi; - -@ApplicationScoped -public class PricesProducer { - - @Outgoing("prices-out") - public Multi generatePrices() { - return Multi.createFrom().items(1.2, 2.2, 3.4); - } - - @Outgoing("prices-out2") - public Multi generatePrices2() { - return Multi.createFrom().items(4.5, 5.6, 6.7); - } -} diff --git a/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/PricesRepeatableProducers.java b/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/PricesRepeatableProducers.java new file mode 100644 index 0000000000000..da21d8a40a01c --- /dev/null +++ b/integration-tests/reactive-messaging-kafka/src/main/java/io/quarkus/it/kafka/PricesRepeatableProducers.java @@ -0,0 +1,22 @@ +package io.quarkus.it.kafka; + +import jakarta.enterprise.context.ApplicationScoped; + +import org.eclipse.microprofile.reactive.messaging.Outgoing; + +import io.smallrye.mutiny.Multi; +import io.smallrye.reactive.messaging.Targeted; + +@ApplicationScoped +public class PricesRepeatableProducers { + + @Outgoing("prices-out") + @Outgoing("prices-out2") + public Multi produce() { + return Multi.createFrom().items( + Targeted.of("prices-out", 1.2, "prices-out2", 4.5), + Targeted.of("prices-out", 2.2, "prices-out2", 5.6), + Targeted.of("prices-out", 3.4, "prices-out2", 6.7)); + } + +} diff --git a/integration-tests/reactive-messaging-kafka/src/main/resources/application.properties b/integration-tests/reactive-messaging-kafka/src/main/resources/application.properties index b80f4bdaff62a..89117fa6769b0 100644 --- a/integration-tests/reactive-messaging-kafka/src/main/resources/application.properties +++ b/integration-tests/reactive-messaging-kafka/src/main/resources/application.properties @@ -18,8 +18,8 @@ mp.messaging.incoming.people-in.checkpoint.state-type=io.quarkus.it.kafka.KafkaR mp.messaging.incoming.people-in.checkpoint.quarkus-redis.client-name=my-redis mp.messaging.incoming.people-in.group.id=people-checkpoint -mp.messaging.outgoing.fruits-out.topic=fruits -mp.messaging.incoming.fruits-in.topic=fruits +mp.messaging.incoming.fruits-in.topic=fruit-.* +mp.messaging.incoming.fruits-in.pattern=true mp.messaging.outgoing.pets-out.topic=pets mp.messaging.incoming.pets-in.topic=pets @@ -27,7 +27,9 @@ mp.messaging.incoming.pets-in.topic=pets quarkus.redis.my-redis.hosts=${quarkus.redis.hosts} mp.messaging.outgoing.prices-out.topic=prices +mp.messaging.outgoing.prices-out.value.serializer=org.apache.kafka.common.serialization.DoubleSerializer mp.messaging.outgoing.prices-out2.topic=prices2 +mp.messaging.outgoing.prices-out2.value.serializer=org.apache.kafka.common.serialization.DoubleSerializer mp.messaging.incoming.prices-in.topic=prices mp.messaging.incoming.prices-in2.topic=prices2 diff --git a/integration-tests/reactive-messaging-kafka/src/test/java/io/quarkus/it/kafka/KafkaConnectorTest.java b/integration-tests/reactive-messaging-kafka/src/test/java/io/quarkus/it/kafka/KafkaConnectorTest.java index 3527c4a9ecf11..82bd9d316defb 100644 --- a/integration-tests/reactive-messaging-kafka/src/test/java/io/quarkus/it/kafka/KafkaConnectorTest.java +++ b/integration-tests/reactive-messaging-kafka/src/test/java/io/quarkus/it/kafka/KafkaConnectorTest.java @@ -24,6 +24,8 @@ public class KafkaConnectorTest { protected static final TypeRef> TYPE_REF = new TypeRef>() { }; + protected static final TypeRef> FRUIT_TYPE_REF = new TypeRef>() { + }; protected static final TypeRef> PEOPLE_STATE_REF = new TypeRef>() { }; @@ -48,7 +50,7 @@ public void testPets() { @Test public void testFruits() { - await().untilAsserted(() -> Assertions.assertEquals(get("/kafka/fruits").as(TYPE_REF).size(), 4)); + await().untilAsserted(() -> Assertions.assertEquals(get("/kafka/fruits").as(FRUIT_TYPE_REF).size(), 5)); } @Test