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 #15986 from Ladicek/reactive-messaging-live-reload
Support live reload for SmallRye Reactive Messaging
- Loading branch information
Showing
11 changed files
with
352 additions
and
1 deletion.
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
104 changes: 104 additions & 0 deletions
104
...java/io/quarkus/smallrye/reactivemessaging/amqp/devmode/nohttp/AmqpDevModeNoHttpTest.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,104 @@ | ||
| package io.quarkus.smallrye.reactivemessaging.amqp.devmode.nohttp; | ||
|
|
||
| import static org.assertj.core.api.Assertions.assertThat; | ||
| import static org.awaitility.Awaitility.await; | ||
|
|
||
| import java.util.List; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.logging.LogRecord; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import org.apache.activemq.artemis.protocol.amqp.broker.ProtonProtocolManagerFactory; | ||
| import org.jboss.shrinkwrap.api.ShrinkWrap; | ||
| import org.jboss.shrinkwrap.api.spec.JavaArchive; | ||
| import org.junit.jupiter.api.AfterAll; | ||
| import org.junit.jupiter.api.BeforeAll; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.junit.jupiter.api.extension.RegisterExtension; | ||
|
|
||
| import io.quarkus.smallrye.reactivemessaging.amqp.AnonymousAmqpBroker; | ||
| import io.quarkus.test.QuarkusDevModeTest; | ||
|
|
||
| public class AmqpDevModeNoHttpTest { | ||
| @RegisterExtension | ||
| static QuarkusDevModeTest TEST = new QuarkusDevModeTest() | ||
| .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class) | ||
| .addClasses(Producer.class, Consumer.class, | ||
| AnonymousAmqpBroker.class, ProtonProtocolManagerFactory.class) | ||
| .addAsResource("broker.xml") | ||
| .addAsResource("application.properties")) | ||
| .setLogRecordPredicate(r -> r.getLoggerName().equals(Consumer.class.getName())); | ||
|
|
||
| @BeforeAll | ||
| public static void startBroker() { | ||
| AnonymousAmqpBroker.start(); | ||
| } | ||
|
|
||
| @AfterAll | ||
| public static void stopBroker() { | ||
| AnonymousAmqpBroker.stop(); | ||
| } | ||
|
|
||
| // For all tests below: we don't know exactly when the AMQP connector receives credits from the broker | ||
| // and then requests items from the producer, so we don't know what number will be sent to the queue first. | ||
| // What we do know, and hence test, is the relationship between consecutive numbers in the queue. | ||
| // See also https://github.com/smallrye/smallrye-reactive-messaging/issues/1125. | ||
|
|
||
| @Test | ||
| public void testProducerUpdate() { | ||
| await().atMost(1, TimeUnit.MINUTES).untilAsserted(() -> { | ||
| List<LogRecord> log = TEST.getLogRecords(); | ||
| assertThat(log).hasSizeGreaterThanOrEqualTo(5); | ||
|
|
||
| List<Long> nums = log.stream() | ||
| .map(it -> Long.parseLong(it.getMessage())) | ||
| .collect(Collectors.toList()); | ||
|
|
||
| long last = nums.get(nums.size() - 1); | ||
| assertThat(nums).containsSequence(last - 4, last - 3, last - 2, last - 1, last); | ||
| }); | ||
|
|
||
| TEST.modifySourceFile(Producer.class, s -> s.replace("* 1", "* 2")); | ||
|
|
||
| await().atMost(1, TimeUnit.MINUTES).untilAsserted(() -> { | ||
| List<LogRecord> log = TEST.getLogRecords(); | ||
| assertThat(log).hasSizeGreaterThanOrEqualTo(5); | ||
|
|
||
| List<Long> nums = log.stream() | ||
| .map(it -> Long.parseLong(it.getMessage())) | ||
| .collect(Collectors.toList()); | ||
|
|
||
| long last = nums.get(nums.size() - 1); | ||
| assertThat(nums).containsSequence(last - 8, last - 6, last - 4, last - 2, last); | ||
| }); | ||
| } | ||
|
|
||
| @Test | ||
| public void testConsumerUpdate() { | ||
| await().atMost(1, TimeUnit.MINUTES).untilAsserted(() -> { | ||
| List<LogRecord> log = TEST.getLogRecords(); | ||
| assertThat(log).hasSizeGreaterThanOrEqualTo(5); | ||
|
|
||
| List<Long> nums = log.stream() | ||
| .map(it -> Long.parseLong(it.getMessage())) | ||
| .collect(Collectors.toList()); | ||
|
|
||
| long last = nums.get(nums.size() - 1); | ||
| assertThat(nums).containsSequence(last - 4, last - 3, last - 2, last - 1, last); | ||
| }); | ||
|
|
||
| TEST.modifySourceFile(Consumer.class, s -> s.replace("* 1", "* 3")); | ||
|
|
||
| await().atMost(1, TimeUnit.MINUTES).untilAsserted(() -> { | ||
| List<LogRecord> log = TEST.getLogRecords(); | ||
| assertThat(log).hasSizeGreaterThanOrEqualTo(5); | ||
|
|
||
| List<Long> nums = log.stream() | ||
| .map(it -> Long.parseLong(it.getMessage())) | ||
| .collect(Collectors.toList()); | ||
|
|
||
| long last = nums.get(nums.size() - 1); | ||
| assertThat(nums).containsSequence(last - 12, last - 9, last - 6, last - 3, last); | ||
| }); | ||
| } | ||
| } |
16 changes: 16 additions & 0 deletions
16
...ent/src/test/java/io/quarkus/smallrye/reactivemessaging/amqp/devmode/nohttp/Consumer.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,16 @@ | ||
| package io.quarkus.smallrye.reactivemessaging.amqp.devmode.nohttp; | ||
|
|
||
| import javax.enterprise.context.ApplicationScoped; | ||
|
|
||
| import org.eclipse.microprofile.reactive.messaging.Incoming; | ||
| import org.jboss.logging.Logger; | ||
|
|
||
| @ApplicationScoped | ||
| public class Consumer { | ||
| private static final Logger log = Logger.getLogger(Consumer.class); | ||
|
|
||
| @Incoming("in") | ||
| public void consume(long content) { | ||
| log.info(content * 1); | ||
| } | ||
| } |
20 changes: 20 additions & 0 deletions
20
...ent/src/test/java/io/quarkus/smallrye/reactivemessaging/amqp/devmode/nohttp/Producer.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,20 @@ | ||
| package io.quarkus.smallrye.reactivemessaging.amqp.devmode.nohttp; | ||
|
|
||
| import java.time.Duration; | ||
|
|
||
| import javax.enterprise.context.ApplicationScoped; | ||
|
|
||
| import org.eclipse.microprofile.reactive.messaging.Outgoing; | ||
| import org.reactivestreams.Publisher; | ||
|
|
||
| import io.smallrye.mutiny.Multi; | ||
|
|
||
| @ApplicationScoped | ||
| public class Producer { | ||
| @Outgoing("source") | ||
| public Publisher<Long> generate() { | ||
| return Multi.createFrom().ticks().every(Duration.ofMillis(200)) | ||
| .onOverflow().drop() | ||
| .map(i -> i * 1); | ||
| } | ||
| } |
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
14 changes: 14 additions & 0 deletions
14
...io/quarkus/smallrye/reactivemessaging/runtime/devmode/DevModeSupportConnectorFactory.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,14 @@ | ||
| package io.quarkus.smallrye.reactivemessaging.runtime.devmode; | ||
|
|
||
| import java.lang.annotation.ElementType; | ||
| import java.lang.annotation.Retention; | ||
| import java.lang.annotation.RetentionPolicy; | ||
| import java.lang.annotation.Target; | ||
|
|
||
| import javax.interceptor.InterceptorBinding; | ||
|
|
||
| @InterceptorBinding | ||
| @Target(ElementType.TYPE) | ||
| @Retention(RetentionPolicy.RUNTIME) | ||
| public @interface DevModeSupportConnectorFactory { | ||
| } |
82 changes: 82 additions & 0 deletions
82
...smallrye/reactivemessaging/runtime/devmode/DevModeSupportConnectorFactoryInterceptor.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,82 @@ | ||
| package io.quarkus.smallrye.reactivemessaging.runtime.devmode; | ||
|
|
||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.function.Supplier; | ||
|
|
||
| import javax.annotation.Priority; | ||
| import javax.interceptor.AroundInvoke; | ||
| import javax.interceptor.Interceptor; | ||
| import javax.interceptor.InvocationContext; | ||
|
|
||
| import org.eclipse.microprofile.reactive.messaging.Message; | ||
| import org.eclipse.microprofile.reactive.streams.operators.PublisherBuilder; | ||
| import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams; | ||
| import org.eclipse.microprofile.reactive.streams.operators.SubscriberBuilder; | ||
| import org.reactivestreams.Subscriber; | ||
| import org.reactivestreams.Subscription; | ||
|
|
||
| @Interceptor | ||
| @DevModeSupportConnectorFactory | ||
| @Priority(Interceptor.Priority.PLATFORM_BEFORE + 10) | ||
| public class DevModeSupportConnectorFactoryInterceptor { | ||
| private static Supplier<CompletableFuture<Boolean>> onMessage; | ||
|
|
||
| static void register(Supplier<CompletableFuture<Boolean>> onMessage) { | ||
| DevModeSupportConnectorFactoryInterceptor.onMessage = onMessage; | ||
| } | ||
|
|
||
| @AroundInvoke | ||
| public Object intercept(InvocationContext ctx) throws Exception { | ||
| if (onMessage == null) { | ||
| return ctx.proceed(); | ||
| } | ||
|
|
||
| if (ctx.getMethod().getName().equals("getPublisherBuilder")) { | ||
| PublisherBuilder<Message<?>> result = (PublisherBuilder<Message<?>>) ctx.proceed(); | ||
| return result.flatMapCompletionStage(msg -> { | ||
| CompletableFuture<Message<?>> future = new CompletableFuture<>(); | ||
| onMessage.get().whenComplete((restarted, error) -> { | ||
| if (!restarted) { | ||
| // if restarted, a new stream is already running, | ||
| // no point in emitting an event to the old stream | ||
| future.complete(msg); | ||
| } | ||
| }); | ||
| return future; | ||
| }); | ||
| } | ||
|
|
||
| if (ctx.getMethod().getName().equals("getSubscriberBuilder")) { | ||
| SubscriberBuilder<Message<?>, Void> result = (SubscriberBuilder<Message<?>, Void>) ctx.proceed(); | ||
| return ReactiveStreams.fromSubscriber(new Subscriber<Message<?>>() { | ||
| private Subscriber<Message<?>> subscriber; | ||
|
|
||
| @Override | ||
| public void onSubscribe(Subscription s) { | ||
| subscriber = result.build(); | ||
| subscriber.onSubscribe(s); | ||
| } | ||
|
|
||
| @Override | ||
| public void onNext(Message<?> o) { | ||
| subscriber.onNext(o); | ||
| onMessage.get().join(); | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable t) { | ||
| subscriber.onError(t); | ||
| onMessage.get().join(); | ||
| } | ||
|
|
||
| @Override | ||
| public void onComplete() { | ||
| subscriber.onComplete(); | ||
| onMessage.get().join(); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| return ctx.proceed(); | ||
| } | ||
| } |
Oops, something went wrong.