-
Notifications
You must be signed in to change notification settings - Fork 2.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #37899 from ozangunalp/smallrye_rm_4.13.0
Bump Smallrye Reactive Messaging version from 4.12.0 to 4.13.0
- Loading branch information
Showing
18 changed files
with
634 additions
and
22 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
33 changes: 33 additions & 0 deletions
33
...ent/src/main/java/io/quarkus/micrometer/deployment/binder/ReactiveMessagingProcessor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| package io.quarkus.micrometer.deployment.binder; | ||
|
|
||
| import java.util.function.BooleanSupplier; | ||
|
|
||
| import io.quarkus.arc.deployment.AdditionalBeanBuildItem; | ||
| import io.quarkus.deployment.annotations.BuildStep; | ||
| import io.quarkus.micrometer.runtime.MicrometerRecorder; | ||
| import io.quarkus.micrometer.runtime.config.MicrometerConfig; | ||
|
|
||
| public class ReactiveMessagingProcessor { | ||
|
|
||
| static final String MESSAGE_OBSERVATION_COLLECTOR = "io.smallrye.reactive.messaging.observation.MessageObservationCollector"; | ||
| static final String METRICS_BEAN_CLASS = "io.quarkus.micrometer.runtime.binder.reactivemessaging.MicrometerObservationCollector"; | ||
| static final Class<?> MESSAGE_OBSERVATION_COLLECTOR_CLASS = MicrometerRecorder | ||
| .getClassForName(MESSAGE_OBSERVATION_COLLECTOR); | ||
|
|
||
| static class ReactiveMessagingSupportEnabled implements BooleanSupplier { | ||
| MicrometerConfig mConfig; | ||
|
|
||
| public boolean getAsBoolean() { | ||
| return MESSAGE_OBSERVATION_COLLECTOR_CLASS != null && | ||
| mConfig.checkBinderEnabledWithDefault(mConfig.binder.messaging); | ||
| } | ||
| } | ||
|
|
||
| @BuildStep(onlyIf = ReactiveMessagingSupportEnabled.class) | ||
| AdditionalBeanBuildItem createCDIEventConsumer() { | ||
| return AdditionalBeanBuildItem.builder() | ||
| .addBeanClass(METRICS_BEAN_CLASS) | ||
| .setUnremovable() | ||
| .build(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
77 changes: 77 additions & 0 deletions
77
...o/quarkus/micrometer/runtime/binder/reactivemessaging/MicrometerObservationCollector.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| package io.quarkus.micrometer.runtime.binder.reactivemessaging; | ||
|
|
||
| import jakarta.enterprise.context.ApplicationScoped; | ||
| import jakarta.inject.Inject; | ||
|
|
||
| import org.eclipse.microprofile.config.inject.ConfigProperty; | ||
| import org.eclipse.microprofile.reactive.messaging.Message; | ||
|
|
||
| import io.micrometer.core.instrument.Counter; | ||
| import io.micrometer.core.instrument.Metrics; | ||
| import io.micrometer.core.instrument.Tag; | ||
| import io.micrometer.core.instrument.Tags; | ||
| import io.micrometer.core.instrument.Timer; | ||
| import io.smallrye.reactive.messaging.observation.DefaultMessageObservation; | ||
| import io.smallrye.reactive.messaging.observation.MessageObservation; | ||
| import io.smallrye.reactive.messaging.observation.MessageObservationCollector; | ||
| import io.smallrye.reactive.messaging.observation.ObservationContext; | ||
|
|
||
| @ApplicationScoped | ||
| public class MicrometerObservationCollector | ||
| implements MessageObservationCollector<MicrometerObservationCollector.MicrometerContext> { | ||
|
|
||
| @Inject | ||
| @ConfigProperty(name = "quarkus.messaging.observation.micrometer.enabled", defaultValue = "true") | ||
| boolean enabled; | ||
|
|
||
| @Override | ||
| public MicrometerContext initObservation(String channel, boolean incoming, boolean emitter) { | ||
| if (enabled) { | ||
| return new MicrometerContext(channel); | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public MessageObservation onNewMessage(String channel, Message<?> message, MicrometerContext ctx) { | ||
| ctx.msgCount.increment(); | ||
| return new DefaultMessageObservation(channel); | ||
| } | ||
|
|
||
| public static class MicrometerContext implements ObservationContext { | ||
| final Counter msgCount; | ||
| final Timer duration; | ||
| final Counter acks; | ||
| final Counter nacks; | ||
|
|
||
| public MicrometerContext(String channel) { | ||
| Tags tags = Tags.of(Tag.of("channel", channel)); | ||
| this.msgCount = Counter.builder("quarkus.messaging.message.count") | ||
| .description("The number of messages observed") | ||
| .tags(tags) | ||
| .register(Metrics.globalRegistry); | ||
| this.duration = Timer.builder("quarkus.messaging.message.duration") | ||
| .description("The duration of the message processing") | ||
| .tags(tags) | ||
| .register(Metrics.globalRegistry); | ||
| this.acks = Counter.builder("quarkus.messaging.message.acks") | ||
| .description("The number of messages processed successfully") | ||
| .tags(tags) | ||
| .register(Metrics.globalRegistry); | ||
| this.nacks = Counter.builder("quarkus.messaging.message.failures") | ||
| .description("The number of messages processed with failures") | ||
| .tags(tags) | ||
| .register(Metrics.globalRegistry); | ||
| } | ||
|
|
||
| @Override | ||
| public void complete(MessageObservation observation) { | ||
| if (observation.getReason() == null) { | ||
| acks.increment(); | ||
| } else { | ||
| nacks.increment(); | ||
| } | ||
| duration.record(observation.getCompletionDuration()); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
35 changes: 35 additions & 0 deletions
35
...time/src/main/java/io/quarkus/micrometer/runtime/config/ReactiveMessagingConfigGroup.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| package io.quarkus.micrometer.runtime.config; | ||
|
|
||
| import java.util.Optional; | ||
|
|
||
| import io.quarkus.runtime.annotations.ConfigGroup; | ||
| import io.quarkus.runtime.annotations.ConfigItem; | ||
|
|
||
| /** | ||
| * Build / static runtime config for Reactive Messaging Binders | ||
| */ | ||
| @ConfigGroup | ||
| public class ReactiveMessagingConfigGroup implements MicrometerConfig.CapabilityEnabled { | ||
| /** | ||
| * Kafka metrics support. | ||
| * <p> | ||
| * Support for Reactive Messaging metrics will be enabled if Micrometer support is enabled, | ||
| * MessageObservationCollector interface is on the classpath | ||
| * and either this value is true, or this value is unset and | ||
| * {@code quarkus.micrometer.binder-enabled-default} is true. | ||
| */ | ||
| @ConfigItem | ||
| public Optional<Boolean> enabled; | ||
|
|
||
| @Override | ||
| public Optional<Boolean> getEnabled() { | ||
| return enabled; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return this.getClass().getSimpleName() | ||
| + "{enabled=" + enabled | ||
| + '}'; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.