Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update SmallRye Reactive Messaging to 2.4.0 and Mutiny to 0.9.0 #12470

Merged
merged 7 commits into from
Oct 3, 2020
Merged

Update SmallRye Reactive Messaging to 2.4.0 and Mutiny to 0.9.0 #12470

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
624f82c
Update SmallRye Reactive Messaging to version 2.4.0
cescoffier Oct 2, 2020
79f6d82
Integrate the expected payload type extraction
cescoffier Sep 17, 2020
3598333
Move all emitter and channel tests in a dedicated package
cescoffier Sep 21, 2020
62d0df3
Add support for mutiny emitter
cescoffier Sep 21, 2020
363ac6a
Implement converter support
cescoffier Sep 22, 2020
6d1b4d7
Integrate the update of the MutinyEmitter API
cescoffier Sep 25, 2020
ab8ba26
Update Mutiny to 0.9.0 (performance fix)
cescoffier Oct 2, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions bom/application/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@
<smallrye-context-propagation.version>1.0.13</smallrye-context-propagation.version>
<smallrye-reactive-streams-operators.version>1.0.13</smallrye-reactive-streams-operators.version>
<smallrye-converter-api.version>1.1.0</smallrye-converter-api.version>
<smallrye-reactive-messaging.version>2.3.0</smallrye-reactive-messaging.version>
<smallrye-reactive-messaging.version>2.4.0</smallrye-reactive-messaging.version>
<swagger-ui.version>3.34.0</swagger-ui.version>
<jakarta.activation.version>1.2.1</jakarta.activation.version>
<jakarta.annotation-api.version>1.3.5</jakarta.annotation-api.version>
Expand Down Expand Up @@ -130,7 +130,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.8.1</mutiny.version>
<mutiny.version>0.9.0</mutiny.version>
<axle-client.version>1.2.1</axle-client.version>
<mutiny-client.version>1.2.1</mutiny-client.version>
<kafka2.version>2.5.0</kafka2.version>
Expand Down
17 changes: 13 additions & 4 deletions ...ment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/EmitterBuildItem.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,14 @@ public final class EmitterBuildItem extends MultiBuildItem {
* Creates a new instance of {@link EmitterBuildItem} setting the overflow strategy.
*
* @param name the name of the stream
* @param isMutinyEmitter if the emitter is a {@link io.smallrye.reactive.messaging.MutinyEmitter}
* @param overflow the overflow strategy
* @param bufferSize the buffer size, if overflow is set to {@code BUFFER}
* @return the new {@link EmitterBuildItem}
*/
static EmitterBuildItem of(String name, String overflow, int bufferSize, boolean hasBroadcast, int awaitSubscribers) {
return new EmitterBuildItem(name, overflow, bufferSize, hasBroadcast, awaitSubscribers);
static EmitterBuildItem of(String name, boolean isMutinyEmitter, String overflow, int bufferSize, boolean hasBroadcast,
int awaitSubscribers) {
return new EmitterBuildItem(name, isMutinyEmitter, overflow, bufferSize, hasBroadcast, awaitSubscribers);
}

/**
Expand All @@ -39,22 +41,29 @@ static EmitterBuildItem of(String name, String overflow, int bufferSize, boolean
*/
private final boolean hasBroadcast;

/**
* Whether the emitter is a {@link io.smallrye.reactive.messaging.MutinyEmitter} or a regular (non-mutiny) emitter.
*/
private final boolean isMutinyEmitter;

/**
* If the emitter uses the {@link io.smallrye.reactive.messaging.annotations.Broadcast} annotation, indicates the
* number of subscribers to be expected before subscribing upstream.
*/
private final int awaitSubscribers;

public EmitterBuildItem(String name, String overflow, int bufferSize, boolean hasBroadcast, int awaitSubscribers) {
public EmitterBuildItem(String name, boolean isMutinyEmitter, String overflow, int bufferSize, boolean hasBroadcast,
int awaitSubscribers) {
this.name = name;
this.overflow = overflow;
this.isMutinyEmitter = isMutinyEmitter;
this.bufferSize = bufferSize;
this.hasBroadcast = hasBroadcast;
this.awaitSubscribers = hasBroadcast ? awaitSubscribers : -1;
}

public EmitterConfiguration getEmitterConfig() {
return new EmitterConfiguration(name, OnOverflowLiteral.create(overflow, bufferSize),
return new EmitterConfiguration(name, isMutinyEmitter, OnOverflowLiteral.create(overflow, bufferSize),
hasBroadcast ? new BroadcastLiteral(awaitSubscribers) : null);
}

Expand Down
118 changes: 86 additions & 32 deletions ...va/io/quarkus/smallrye/reactivemessaging/deployment/QuarkusMediatorConfigurationUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,9 @@
package io.quarkus.smallrye.reactivemessaging.deployment;

import static io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames.ACKNOWLEDGMENT;
import static io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames.BLOCKING;
import static io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames.BROADCAST;
import static io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames.INCOMING;
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.*;

import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import org.eclipse.microprofile.reactive.messaging.Acknowledgment;
Expand All @@ -26,12 +22,12 @@
import io.smallrye.reactive.messaging.annotations.Blocking;
import io.smallrye.reactive.messaging.annotations.Merge;

final class QuarkusMediatorConfigurationUtil {
public final class QuarkusMediatorConfigurationUtil {

private QuarkusMediatorConfigurationUtil() {
}

static QuarkusMediatorConfiguration create(MethodInfo methodInfo, BeanInfo bean,
public static QuarkusMediatorConfiguration create(MethodInfo methodInfo, BeanInfo bean,
String generatedInvokerName, RecorderContext recorderContext, ClassLoader cl) {

Class<?> returnTypeClass = load(methodInfo.returnType().name().toString(), cl);
Expand Down Expand Up @@ -81,41 +77,49 @@ static QuarkusMediatorConfiguration create(MethodInfo methodInfo, BeanInfo bean,
acknowledgment);
configuration.setProduction(validationOutput.getProduction());
configuration.setConsumption(validationOutput.getConsumption());
if (validationOutput.getUseBuilderTypes() != null) {
configuration.setIngestedPayloadType(validationOutput.getIngestedPayloadType());
if (validationOutput.getUseBuilderTypes()) {
configuration.setUseBuilderTypes(validationOutput.getUseBuilderTypes());
} else {
configuration.setUseBuilderTypes(false);
}

if (acknowledgment == null) {
acknowledgment = mediatorConfigurationSupport.processDefaultAcknowledgement(shape,
validationOutput.getConsumption());
validationOutput.getConsumption(), validationOutput.getProduction());
configuration.setAcknowledgment(acknowledgment);
}

configuration.setMerge(mediatorConfigurationSupport.processMerge(incomingValues, () -> {
AnnotationInstance instance = methodInfo.annotation(MERGE);
if (instance != null) {
AnnotationValue value = instance.value();
if (value == null) {
return Merge.Mode.MERGE; // the default value of @Merge
configuration.setMerge(mediatorConfigurationSupport.processMerge(incomingValues, new Supplier<Merge.Mode>() {
@Override
public Merge.Mode get() {
AnnotationInstance instance = methodInfo.annotation(MERGE);
if (instance != null) {
AnnotationValue value = instance.value();
if (value == null) {
return Merge.Mode.MERGE; // the default value of @Merge
}
return Merge.Mode.valueOf(value.asEnum());
}
return Merge.Mode.valueOf(value.asEnum());
return null;
}
return null;
}));

configuration.setBroadcastValue(mediatorConfigurationSupport.processBroadcast(outgoingValue, () -> {
AnnotationInstance instance = methodInfo.annotation(BROADCAST);
if (instance != null) {
AnnotationValue value = instance.value();
if (value == null) {
return 0; // the default value of @Broadcast
}
return value.asInt();
}
return null;
}));
configuration.setBroadcastValue(mediatorConfigurationSupport.processBroadcast(outgoingValue,
new Supplier<Integer>() {
@Override
public Integer get() {
AnnotationInstance instance = methodInfo.annotation(BROADCAST);
if (instance != null) {
AnnotationValue value = instance.value();
if (value == null) {
return 0; // the default value of @Broadcast
}
return value.asInt();
}
return null;
}
}));

AnnotationInstance blockingAnnotation = methodInfo.annotation(BLOCKING);
if (blockingAnnotation != null) {
Expand Down Expand Up @@ -192,7 +196,7 @@ private static String fullMethodName(MethodInfo methodInfo) {
return methodInfo.declaringClass() + "#" + methodInfo.name();
}

private static class ReturnTypeGenericTypeAssignable extends JandexGenericTypeAssignable {
public static class ReturnTypeGenericTypeAssignable extends JandexGenericTypeAssignable {

public ReturnTypeGenericTypeAssignable(MethodInfo method, ClassLoader classLoader) {
super(method.returnType(), classLoader);
Expand Down Expand Up @@ -223,18 +227,68 @@ public Result check(Class<?> target, int index) {
return Result.InvalidIndex;
}
}

@Override
public java.lang.reflect.Type getType(int index) {
Type t = extract(type, index);
if (t != null) {
return load(t.name().toString(), classLoader);
}
return null;
}

private Type extract(Type type, int index) {
if (type.kind() != Type.Kind.PARAMETERIZED_TYPE) {
return null;
} else {
List<Type> arguments = type.asParameterizedType().arguments();
if (arguments.size() >= index + 1) {
Type result = arguments.get(index);
if (result.kind() == Type.Kind.WILDCARD_TYPE) {
return null;
}
return result;
} else {
return null;
}
}
}

@Override
public java.lang.reflect.Type getType(int index, int subIndex) {
Type generic = extract(type, index);
if (generic != null) {
Type t = extract(generic, subIndex);
if (t != null) {
return load(t.name().toString(), classLoader);
}
return null;
} else {
return null;
}
}
}

private static class AlwaysInvalidIndexGenericTypeAssignable
public static class AlwaysInvalidIndexGenericTypeAssignable
implements MediatorConfigurationSupport.GenericTypeAssignable {

@Override
public Result check(Class<?> target, int index) {
return Result.InvalidIndex;
}

@Override
public java.lang.reflect.Type getType(int index) {
return null;
}

@Override
public java.lang.reflect.Type getType(int index, int subIndex) {
return null;
}
}

private static class MethodParamGenericTypeAssignable extends JandexGenericTypeAssignable {
public static class MethodParamGenericTypeAssignable extends JandexGenericTypeAssignable {

public MethodParamGenericTypeAssignable(MethodInfo method, int paramIndex, ClassLoader classLoader) {
super(getGenericParameterType(method, paramIndex), classLoader);
Expand Down
2 changes: 2 additions & 0 deletions ...main/java/io/quarkus/smallrye/reactivemessaging/deployment/ReactiveMessagingDotNames.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import org.eclipse.microprofile.reactive.messaging.Outgoing;
import org.jboss.jandex.DotName;

import io.smallrye.reactive.messaging.MutinyEmitter;
import io.smallrye.reactive.messaging.annotations.Blocking;
import io.smallrye.reactive.messaging.annotations.Broadcast;
import io.smallrye.reactive.messaging.annotations.Channel;
Expand All @@ -21,6 +22,7 @@ public final class ReactiveMessagingDotNames {
static final DotName CHANNEL = DotName.createSimple(org.eclipse.microprofile.reactive.messaging.Channel.class.getName());
static final DotName LEGACY_CHANNEL = DotName.createSimple(Channel.class.getName());
static final DotName EMITTER = DotName.createSimple(org.eclipse.microprofile.reactive.messaging.Emitter.class.getName());
static final DotName MUTINY_EMITTER = DotName.createSimple(MutinyEmitter.class.getName());
static final DotName LEGACY_EMITTER = DotName.createSimple(Emitter.class.getName());
static final DotName ON_OVERFLOW = DotName
.createSimple(org.eclipse.microprofile.reactive.messaging.OnOverflow.class.getName());
Expand Down
80 changes: 35 additions & 45 deletions .../io/quarkus/smallrye/reactivemessaging/deployment/SmallRyeReactiveMessagingProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Optional;
Expand Down Expand Up @@ -71,9 +72,6 @@
import io.smallrye.reactive.messaging.health.SmallRyeReactiveMessagingLivenessCheck;
import io.smallrye.reactive.messaging.health.SmallRyeReactiveMessagingReadinessCheck;

/**
*
*/
public class SmallRyeReactiveMessagingProcessor {

private static final Logger LOGGER = Logger
Expand Down Expand Up @@ -168,22 +166,12 @@ void validateBeanDeployment(

for (InjectionPointInfo injectionPoint : validationPhase.getContext()
.get(BuildExtension.Key.INJECTION_POINTS)) {
Optional<AnnotationInstance> broadcast = getAnnotation(annotationStore, injectionPoint,
ReactiveMessagingDotNames.BROADCAST);

Optional<AnnotationInstance> broadcast = annotationStore.getAnnotations(injectionPoint.getTarget())
.stream()
.filter(ai -> ReactiveMessagingDotNames.BROADCAST.equals(ai.name()))
.filter(ai -> {
if (ai.target().kind() == AnnotationTarget.Kind.METHOD_PARAMETER && injectionPoint
.isParam()) {
return ai.target().asMethodParameter().position() == injectionPoint.getPosition();
}
return true;
})
.findAny();

// New emitter from the spec.
if (injectionPoint.getRequiredType().name().equals(
ReactiveMessagingDotNames.EMITTER)) {
// New emitter from the spec, or Mutiny emitter
if (injectionPoint.getRequiredType().name().equals(ReactiveMessagingDotNames.EMITTER)
|| injectionPoint.getRequiredType().name().equals(ReactiveMessagingDotNames.MUTINY_EMITTER)) {
AnnotationInstance instance = injectionPoint
.getRequiredQualifier(ReactiveMessagingDotNames.CHANNEL);
if (instance == null) {
Expand All @@ -193,19 +181,10 @@ void validateBeanDeployment(
.getTargetInfo()));
} else {
String channelName = instance.value().asString();
Optional<AnnotationInstance> overflow = annotationStore.getAnnotations(injectionPoint.getTarget())
.stream()
.filter(ai -> ReactiveMessagingDotNames.ON_OVERFLOW
.equals(ai.name()))
.filter(ai -> {
if (ai.target().kind() == AnnotationTarget.Kind.METHOD_PARAMETER && injectionPoint
.isParam()) {
return ai.target().asMethodParameter().position() == injectionPoint.getPosition();
}
return true;
})
.findAny();
createEmitter(emitters, injectionPoint, channelName, overflow, broadcast);
Optional<AnnotationInstance> overflow = getAnnotation(annotationStore, injectionPoint,
ReactiveMessagingDotNames.ON_OVERFLOW);
createEmitter(emitters,
injectionPoint, channelName, overflow, broadcast);
}
}

Expand All @@ -221,27 +200,37 @@ void validateBeanDeployment(
.getTargetInfo()));
} else {
String channelName = instance.value().asString();
Optional<AnnotationInstance> overflow = annotationStore.getAnnotations(injectionPoint.getTarget())
.stream()
.filter(ai -> ReactiveMessagingDotNames.LEGACY_ON_OVERFLOW
.equals(ai.name()))
.filter(ai -> {
if (ai.target().kind() == AnnotationTarget.Kind.METHOD_PARAMETER && injectionPoint
.isParam()) {
return ai.target().asMethodParameter().position() == injectionPoint.getPosition();
}
return true;
})
.findAny();
Optional<AnnotationInstance> overflow = getAnnotation(annotationStore, injectionPoint,
ReactiveMessagingDotNames.LEGACY_ON_OVERFLOW);

createEmitter(emitters, injectionPoint, channelName, overflow, broadcast);
}
}
}
}

private Optional<AnnotationInstance> getAnnotation(AnnotationStore annotationStore, InjectionPointInfo injectionPoint,
DotName onOverflowAnnotation) {
Collection<AnnotationInstance> annotations = annotationStore.getAnnotations(injectionPoint.getTarget());
for (AnnotationInstance annotation : annotations) {
if (onOverflowAnnotation.equals(annotation.name())) {
// For method parameter we must check the position
if (annotation.target().kind() == AnnotationTarget.Kind.METHOD_PARAMETER
&& injectionPoint.isParam()
&& annotation.target().asMethodParameter().position() == injectionPoint.getPosition()) {
return Optional.of(annotation);
} else if (annotation.target().kind() != AnnotationTarget.Kind.METHOD_PARAMETER) {
// For other kind, no need to check anything else
return Optional.of(annotation);
}
}
}
return Optional.empty();
}

@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
private void createEmitter(BuildProducer<EmitterBuildItem> emitters, InjectionPointInfo injectionPoint,
private void createEmitter(BuildProducer<EmitterBuildItem> emitters,
InjectionPointInfo injectionPoint,
String channelName,
Optional<AnnotationInstance> overflow,
Optional<AnnotationInstance> broadcast) {
Expand All @@ -265,8 +254,9 @@ private void createEmitter(BuildProducer<EmitterBuildItem> emitters, InjectionPo
strategy = annotation.value().asString();
}

boolean isMutinyEmitter = injectionPoint.getRequiredType().name().equals(ReactiveMessagingDotNames.MUTINY_EMITTER);
emitters.produce(
EmitterBuildItem.of(channelName, strategy, bufferSize, hasBroadcast, awaitSubscribers));
EmitterBuildItem.of(channelName, isMutinyEmitter, strategy, bufferSize, hasBroadcast, awaitSubscribers));
}

@BuildStep
Expand Down
Loading