From 01a3f56aa7bc398d91dec127457579a97f5479c5 Mon Sep 17 00:00:00 2001
From: Juan C Galvis <8420868+juancgalvis@users.noreply.github.com>
Date: Tue, 11 Aug 2026 15:02:31 -0500
Subject: [PATCH 1/2] fix: allow programmatic features
---
.../EnableReactiveCommonsDynamic.java | 50 ++++
.../config/ReactiveCommonsDomainFeatures.java | 19 ++
.../config/ReactiveCommonsDynamicConfig.java | 269 +++++++++++++++++
.../config/ReactiveCommonsFeatures.java | 52 ++++
.../ReactiveCommonsDynamicConfigTest.java | 273 ++++++++++++++++++
5 files changed, 663 insertions(+)
create mode 100644 starters/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/annotations/EnableReactiveCommonsDynamic.java
create mode 100644 starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDomainFeatures.java
create mode 100644 starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfig.java
create mode 100644 starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsFeatures.java
create mode 100644 starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfigTest.java
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/annotations/EnableReactiveCommonsDynamic.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/annotations/EnableReactiveCommonsDynamic.java
new file mode 100644
index 00000000..5b5e521a
--- /dev/null
+++ b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/annotations/EnableReactiveCommonsDynamic.java
@@ -0,0 +1,50 @@
+package org.reactivecommons.async.impl.config.annotations;
+
+import org.reactivecommons.async.starter.config.ReactiveCommonsDynamicConfig;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Enables runtime-configurable reactive-commons features driven by a
+ * {@link org.reactivecommons.async.starter.config.ReactiveCommonsFeatures} bean.
+ *
+ * Place this annotation on any {@code @Configuration} class (typically your
+ * {@code @SpringBootApplication}) and declare a
+ * {@link org.reactivecommons.async.starter.config.ReactiveCommonsFeatures} bean
+ * to control which features are activated at runtime:
+ *
+ *
{@code
+ * @EnableReactiveCommonsDynamic
+ * @SpringBootApplication
+ * public class MyApplication { ... }
+ *
+ * @Bean
+ * public ReactiveCommonsFeatures reactiveCommonsFeatures() {
+ * boolean needsEvents = someService.needsEventListening();
+ * return ReactiveCommonsFeatures.builder()
+ * .listenEvents(needsEvents)
+ * .sendEvents(true)
+ * .listenCommands(true)
+ * .sendCommands(true)
+ * .build();
+ * }
+ * }
+ *
+ * This is a dynamic alternative to combining the static annotations
+ * ({@code @EnableEventListeners}, {@code @EnableCommandListeners},
+ * {@code @EnableDomainEventBus}, etc.). Both approaches coexist — no changes
+ * are needed to existing code that uses the static annotations.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.TYPE})
+@Documented
+@Import(ReactiveCommonsDynamicConfig.class)
+@Configuration
+public @interface EnableReactiveCommonsDynamic {
+}
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDomainFeatures.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDomainFeatures.java
new file mode 100644
index 00000000..ec88cbf1
--- /dev/null
+++ b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDomainFeatures.java
@@ -0,0 +1,19 @@
+package org.reactivecommons.async.starter.config;
+
+import java.util.HashMap;
+
+
+public class ReactiveCommonsDomainFeatures extends HashMap {
+
+ public ReactiveCommonsDomainFeatures() {
+ super();
+ }
+
+ public ReactiveCommonsFeatures ofDomain(String key) {
+ return this.get(key);
+ }
+
+ public ReactiveCommonsFeatures withDomain(String key) {
+ return this.computeIfAbsent(key, k -> new ReactiveCommonsFeatures());
+ }
+}
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfig.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfig.java
new file mode 100644
index 00000000..f82b19c7
--- /dev/null
+++ b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfig.java
@@ -0,0 +1,269 @@
+package org.reactivecommons.async.starter.config;
+
+import lombok.extern.log4j.Log4j2;
+import org.reactivecommons.api.domain.DomainEventBus;
+import org.reactivecommons.async.api.DirectAsyncGateway;
+import org.reactivecommons.async.starter.senders.GenericDirectAsyncGateway;
+import org.reactivecommons.async.starter.senders.GenericDomainEventBus;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import reactor.core.publisher.Mono;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Dynamic alternative to the static {@code @Enable*} annotations.
+ *
+ * All beans in this configuration are only created when a {@link ReactiveCommonsFeatures} bean
+ * is present in the application context. Each feature is then conditionally activated based
+ * on the flags set in that bean.
+ *
+ * Use {@code @EnableReactiveCommonsDynamic} (or {@code @Import(ReactiveCommonsDynamicConfig.class)})
+ * together with a {@code @Bean} of type {@link ReactiveCommonsFeatures} in your application.
+ */
+@Log4j2
+@Configuration
+@Import({ReactiveCommonsConfig.class, ReactiveCommonsListenersConfig.class})
+public class ReactiveCommonsDynamicConfig {
+
+ // -------------------------------------------------------------------------
+ // Listener activation — side-effect beans that start broker listeners
+ // when the corresponding feature flag is true.
+ // -------------------------------------------------------------------------
+
+ @Bean
+ @SuppressWarnings("rawtypes")
+ public Object dynamicEventListenerActivator(ConnectionManager manager,
+ DomainHandlers handlers,
+ ReactiveCommonsDomainFeatures features) {
+ manager.forDomain((domain, provider) -> {
+ if (features.ofDomain(domain).isListenEvents()) {
+ log.info("ReactiveCommons: activating event listeners for domain '{}'", domain);
+ provider.listenDomainEvents(handlers.get(domain));
+ }
+ });
+ return new Object();
+ }
+
+ @Bean
+ @SuppressWarnings("rawtypes")
+ public Object dynamicNotificationListenerActivator(ConnectionManager manager,
+ DomainHandlers handlers,
+ ReactiveCommonsDomainFeatures features) {
+ manager.forDomain((domain, provider) -> {
+ if (features.ofDomain(domain).isListenNotificationEvents()) {
+ log.info("ReactiveCommons: activating notification event listeners for domain '{}'", domain);
+ provider.listenNotificationEvents(handlers.get(domain));
+ }
+ });
+ return new Object();
+ }
+
+ @Bean
+ @SuppressWarnings("rawtypes")
+ public Object dynamicCommandListenerActivator(ConnectionManager manager,
+ DomainHandlers handlers,
+ ReactiveCommonsDomainFeatures features) {
+ manager.forDomain((domain, provider) -> {
+ if (features.ofDomain(domain).isListenCommands()) {
+ log.info("ReactiveCommons: activating command listeners for domain '{}'", domain);
+ provider.listenCommands(handlers.get(domain));
+ }
+ });
+ return new Object();
+ }
+
+ @Bean
+ @SuppressWarnings("rawtypes")
+ public Object dynamicQueryListenerActivator(ConnectionManager manager,
+ DomainHandlers handlers,
+ ReactiveCommonsDomainFeatures features) {
+ manager.forDomain((domain, provider) -> {
+ if (features.ofDomain(domain).isListenQueries()) {
+ log.info("ReactiveCommons: activating query listeners for domain '{}'", domain);
+ provider.listenQueries(handlers.get(domain));
+ }
+ });
+ return new Object();
+ }
+
+ @Bean
+ @SuppressWarnings("rawtypes")
+ public Object dynamicQueueListenerActivator(ConnectionManager manager,
+ DomainHandlers handlers,
+ ReactiveCommonsDomainFeatures features) {
+ manager.forDomain((domain, provider) -> {
+ if (features.ofDomain(domain).isListenQueues()) {
+ log.info("ReactiveCommons: activating queue listeners for domain '{}'", domain);
+ provider.listenQueues(handlers.get(domain));
+ }
+ });
+ return new Object();
+ }
+
+ // -------------------------------------------------------------------------
+ // Sender beans — always created when ReactiveCommonsFeatures is present.
+ // If the corresponding flag is false, calls return a Mono.error to signal
+ // misconfiguration clearly at use-time rather than at startup.
+ // -------------------------------------------------------------------------
+
+ @Bean
+ @SuppressWarnings("rawtypes")
+ public DomainEventBus dynamicDomainEventBus(ConnectionManager manager,
+ ReactiveCommonsDomainFeatures features) {
+ ConcurrentMap buses = new ConcurrentHashMap<>();
+ manager.forDomain((domain, provider) -> {
+ if (features.ofDomain(domain).isSendEvents()) {
+ log.info("ReactiveCommons: activating DomainEventBus for domain '{}'", domain);
+ buses.put(domain, provider.getDomainBus());
+ }
+ });
+ if (buses.isEmpty()) {
+ return new DisabledDomainEventBus();
+ }
+ return new GenericDomainEventBus(buses);
+ }
+
+ @Bean
+ @SuppressWarnings("rawtypes")
+ public DirectAsyncGateway dynamicDirectAsyncGateway(ConnectionManager manager,
+ ReactiveCommonsDomainFeatures features) {
+ ConcurrentMap gateways = new ConcurrentHashMap<>();
+ manager.forDomain((domain, provider) -> {
+ if (features.ofDomain(domain).isSendCommands()) {
+ log.info("ReactiveCommons: activating DirectAsyncGateway for domain '{}'", domain);
+ gateways.put(domain, provider.getDirectAsyncGateway());
+ }
+ });
+ if (gateways.isEmpty()) {
+ return new DisabledDirectAsyncGateway();
+ }
+ return new GenericDirectAsyncGateway(gateways);
+ }
+
+ // -------------------------------------------------------------------------
+ // Disabled no-op implementations used when a sender feature flag is false.
+ // These prevent startup failures caused by missing bean definitions while
+ // surfacing a clear error at the point where the disabled feature is used.
+ // -------------------------------------------------------------------------
+
+ private static final String SEND_EVENTS_DISABLED =
+ "sendEvents feature is disabled in ReactiveCommonsFeatures. " +
+ "Set sendEvents=true to publish domain events.";
+
+ private static final String SEND_COMMANDS_DISABLED =
+ "sendCommands feature is disabled in ReactiveCommonsFeatures. " +
+ "Set sendCommands=true to send commands or queries.";
+
+ private static class DisabledDomainEventBus implements DomainEventBus {
+ @Override
+ public org.reactivestreams.Publisher emit(org.reactivecommons.api.domain.DomainEvent event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+
+ @Override
+ public org.reactivestreams.Publisher emit(String domain,
+ org.reactivecommons.api.domain.DomainEvent event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+
+ @Override
+ public org.reactivestreams.Publisher emit(io.cloudevents.CloudEvent event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+
+ @Override
+ public org.reactivestreams.Publisher emit(String domain, io.cloudevents.CloudEvent event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+
+ @Override
+ public org.reactivestreams.Publisher emit(org.reactivecommons.api.domain.RawMessage event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+
+ @Override
+ public org.reactivestreams.Publisher emit(String domain,
+ org.reactivecommons.api.domain.RawMessage event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+ }
+
+ private static class DisabledDirectAsyncGateway implements DirectAsyncGateway {
+ @Override
+ public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName,
+ long delayMillis) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName,
+ String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName,
+ long delayMillis, String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName, long delayMillis) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName, String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName, long delayMillis,
+ String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono requestReply(org.reactivecommons.async.api.AsyncQuery query, String targetName,
+ Class type) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono requestReply(org.reactivecommons.async.api.AsyncQuery query, String targetName,
+ Class type, String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono requestReply(io.cloudevents.CloudEvent query,
+ String targetName, Class type) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono requestReply(io.cloudevents.CloudEvent query,
+ String targetName, Class type,
+ String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono reply(T response, org.reactivecommons.async.api.From from) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+ }
+}
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsFeatures.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsFeatures.java
new file mode 100644
index 00000000..c2c5f1c4
--- /dev/null
+++ b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsFeatures.java
@@ -0,0 +1,52 @@
+package org.reactivecommons.async.starter.config;
+
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+public class ReactiveCommonsFeatures {
+
+ /**
+ * Equivalent to {@code @EnableEventListeners}.
+ * Start listening to domain events from the broker.
+ */
+ private boolean listenEvents;
+
+ /**
+ * Equivalent to {@code @EnableNotificationListener}.
+ * Start listening to notification events from the broker.
+ */
+ private boolean listenNotificationEvents;
+
+ /**
+ * Equivalent to {@code @EnableCommandListeners}.
+ * Start listening to commands from the broker.
+ */
+ private boolean listenCommands;
+
+ /**
+ * Equivalent to {@code @EnableQueryListeners}.
+ * Start listening to queries from the broker.
+ */
+ private boolean listenQueries;
+
+ /**
+ * Equivalent to {@code @EnableQueueListeners}.
+ * Start listening to queues from the broker.
+ */
+ private boolean listenQueues;
+
+ /**
+ * Equivalent to {@code @EnableDomainEventBus}.
+ * Exposes a {@link org.reactivecommons.api.domain.DomainEventBus} bean for publishing events.
+ */
+ private boolean sendEvents;
+
+ /**
+ * Equivalent to {@code @EnableDirectAsyncGateway}.
+ * Exposes a {@link org.reactivecommons.async.api.DirectAsyncGateway} bean for sending
+ * commands and request-reply queries.
+ */
+ private boolean sendCommands;
+}
diff --git a/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfigTest.java b/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfigTest.java
new file mode 100644
index 00000000..a3da605f
--- /dev/null
+++ b/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfigTest.java
@@ -0,0 +1,273 @@
+package org.reactivecommons.async.starter.config;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.reactivecommons.api.domain.DomainEventBus;
+import org.reactivecommons.async.api.DirectAsyncGateway;
+import org.reactivecommons.async.commons.HandlerResolver;
+import org.reactivecommons.async.starter.broker.BrokerProvider;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
+import static reactor.test.StepVerifier.create;
+
+@ExtendWith(MockitoExtension.class)
+class ReactiveCommonsDynamicConfigTest {
+
+ @Mock
+ private BrokerProvider provider;
+ @Mock
+ private HandlerResolver resolver;
+ @Mock
+ private DomainEventBus domainEventBus;
+ @Mock
+ private DirectAsyncGateway directAsyncGateway;
+
+ private ConnectionManager manager;
+ private DomainHandlers handlers;
+ private ReactiveCommonsDynamicConfig config;
+
+ @BeforeEach
+ void setUp() {
+ config = new ReactiveCommonsDynamicConfig();
+ manager = new ConnectionManager();
+ manager.addDomain(DEFAULT_DOMAIN, provider);
+ handlers = new DomainHandlers();
+ handlers.add(DEFAULT_DOMAIN, resolver);
+ }
+
+ // -------------------------------------------------------------------------
+ // Helper: build a ReactiveCommonsDomainFeatures for DEFAULT_DOMAIN
+ // -------------------------------------------------------------------------
+
+ private ReactiveCommonsDomainFeatures domainFeatures(boolean listenEvents,
+ boolean listenNotifications,
+ boolean listenCommands,
+ boolean listenQueries,
+ boolean sendEvents,
+ boolean sendCommands) {
+ ReactiveCommonsDomainFeatures features = new ReactiveCommonsDomainFeatures();
+ ReactiveCommonsFeatures f = features.withDomain(DEFAULT_DOMAIN);
+ f.setListenEvents(listenEvents);
+ f.setListenNotificationEvents(listenNotifications);
+ f.setListenCommands(listenCommands);
+ f.setListenQueries(listenQueries);
+ f.setSendEvents(sendEvents);
+ f.setSendCommands(sendCommands);
+ return features;
+ }
+
+ private ReactiveCommonsDomainFeatures allFalse() {
+ return domainFeatures(false, false, false, false, false, false);
+ }
+
+ // -------------------------------------------------------------------------
+ // Listener activation tests
+ // -------------------------------------------------------------------------
+
+ @Test
+ void shouldActivateEventListenersWhenFlagIsTrue() {
+ ReactiveCommonsDomainFeatures features = domainFeatures(true, false, false, false, false, false);
+ config.dynamicEventListenerActivator(manager, handlers, features);
+ verify(provider).listenDomainEvents(resolver);
+ }
+
+ @Test
+ void shouldNotActivateEventListenersWhenFlagIsFalse() {
+ config.dynamicEventListenerActivator(manager, handlers, allFalse());
+ verify(provider, never()).listenDomainEvents(resolver);
+ }
+
+ @Test
+ void shouldActivateNotificationListenersWhenFlagIsTrue() {
+ ReactiveCommonsDomainFeatures features = domainFeatures(false, true, false, false, false, false);
+ config.dynamicNotificationListenerActivator(manager, handlers, features);
+ verify(provider).listenNotificationEvents(resolver);
+ }
+
+ @Test
+ void shouldNotActivateNotificationListenersWhenFlagIsFalse() {
+ config.dynamicNotificationListenerActivator(manager, handlers, allFalse());
+ verify(provider, never()).listenNotificationEvents(resolver);
+ }
+
+ @Test
+ void shouldActivateCommandListenersWhenFlagIsTrue() {
+ ReactiveCommonsDomainFeatures features = domainFeatures(false, false, true, false, false, false);
+ config.dynamicCommandListenerActivator(manager, handlers, features);
+ verify(provider).listenCommands(resolver);
+ }
+
+ @Test
+ void shouldNotActivateCommandListenersWhenFlagIsFalse() {
+ config.dynamicCommandListenerActivator(manager, handlers, allFalse());
+ verify(provider, never()).listenCommands(resolver);
+ }
+
+ @Test
+ void shouldActivateQueryListenersWhenFlagIsTrue() {
+ ReactiveCommonsDomainFeatures features = domainFeatures(false, false, false, true, false, false);
+ config.dynamicQueryListenerActivator(manager, handlers, features);
+ verify(provider).listenQueries(resolver);
+ }
+
+ @Test
+ void shouldNotActivateQueryListenersWhenFlagIsFalse() {
+ config.dynamicQueryListenerActivator(manager, handlers, allFalse());
+ verify(provider, never()).listenQueries(resolver);
+ }
+
+ @Test
+ void shouldActivateAllListenersWhenAllFlagsAreTrue() {
+ ReactiveCommonsDomainFeatures features = domainFeatures(true, true, true, true, false, false);
+
+ config.dynamicEventListenerActivator(manager, handlers, features);
+ config.dynamicNotificationListenerActivator(manager, handlers, features);
+ config.dynamicCommandListenerActivator(manager, handlers, features);
+ config.dynamicQueryListenerActivator(manager, handlers, features);
+
+ verify(provider).listenDomainEvents(resolver);
+ verify(provider).listenNotificationEvents(resolver);
+ verify(provider).listenCommands(resolver);
+ verify(provider).listenQueries(resolver);
+ }
+
+ @Test
+ void shouldActivateNoListenersWhenAllFlagsAreFalse() {
+ ReactiveCommonsDomainFeatures features = allFalse();
+
+ config.dynamicEventListenerActivator(manager, handlers, features);
+ config.dynamicNotificationListenerActivator(manager, handlers, features);
+ config.dynamicCommandListenerActivator(manager, handlers, features);
+ config.dynamicQueryListenerActivator(manager, handlers, features);
+
+ verify(provider, never()).listenDomainEvents(resolver);
+ verify(provider, never()).listenNotificationEvents(resolver);
+ verify(provider, never()).listenCommands(resolver);
+ verify(provider, never()).listenQueries(resolver);
+ }
+
+ @Test
+ void shouldActivateListenersOnMultipleDomainsWhenFlagIsTrue() {
+ manager.addDomain("other-domain", provider);
+ handlers.add("other-domain", resolver);
+
+ ReactiveCommonsDomainFeatures features = new ReactiveCommonsDomainFeatures();
+ features.withDomain(DEFAULT_DOMAIN).setListenEvents(true);
+ features.withDomain("other-domain").setListenEvents(true);
+
+ config.dynamicEventListenerActivator(manager, handlers, features);
+
+ verify(provider, times(2)).listenDomainEvents(resolver);
+ }
+
+ @Test
+ void shouldOnlyActivateListenerForConfiguredDomainWhenOtherDomainFlagIsFalse() {
+ manager.addDomain("other-domain", provider);
+ handlers.add("other-domain", resolver);
+
+ ReactiveCommonsDomainFeatures features = new ReactiveCommonsDomainFeatures();
+ features.withDomain(DEFAULT_DOMAIN).setListenEvents(true);
+ features.withDomain("other-domain"); // listenEvents=false by default
+
+ config.dynamicEventListenerActivator(manager, handlers, features);
+
+ verify(provider, times(1)).listenDomainEvents(resolver);
+ }
+
+ // -------------------------------------------------------------------------
+ // Sender bean tests
+ // -------------------------------------------------------------------------
+
+ @Test
+ void shouldCreateRealDomainEventBusWhenSendEventsIsTrue() {
+ when(provider.getDomainBus()).thenReturn(domainEventBus);
+ ReactiveCommonsDomainFeatures features = domainFeatures(false, false, false, false, true, false);
+
+ DomainEventBus bus = config.dynamicDomainEventBus(manager, features);
+
+ assertThat(bus).isNotNull();
+ verify(provider).getDomainBus();
+ }
+
+ @Test
+ void shouldReturnDisabledDomainEventBusWhenSendEventsIsFalse() {
+ ReactiveCommonsDomainFeatures features = allFalse();
+
+ DomainEventBus bus = config.dynamicDomainEventBus(manager, features);
+
+ assertThat(bus).isNotNull();
+ verify(provider, never()).getDomainBus();
+ }
+
+ @Test
+ void disabledDomainEventBusShouldEmitErrorOnUse() {
+ ReactiveCommonsDomainFeatures features = allFalse();
+ DomainEventBus bus = config.dynamicDomainEventBus(manager, features);
+
+ create(bus.emit(new org.reactivecommons.api.domain.DomainEvent<>("test", "1", "data")))
+ .expectErrorSatisfies(ex -> assertThat(ex)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("sendEvents"))
+ .verify();
+ }
+
+ @Test
+ void shouldCreateRealDirectAsyncGatewayWhenSendCommandsIsTrue() {
+ when(provider.getDirectAsyncGateway()).thenReturn(directAsyncGateway);
+ ReactiveCommonsDomainFeatures features = domainFeatures(false, false, false, false, false, true);
+
+ DirectAsyncGateway gateway = config.dynamicDirectAsyncGateway(manager, features);
+
+ assertThat(gateway).isNotNull();
+ verify(provider).getDirectAsyncGateway();
+ }
+
+ @Test
+ void shouldCreateDirectAsyncGatewayEvenWhenSendCommandsIsFalse() {
+ // The gateway is always registered in the map regardless of sendCommands flag.
+ // Conditional logging only; the actual put is unconditional.
+ ReactiveCommonsDomainFeatures features = allFalse();
+
+ DirectAsyncGateway gateway = config.dynamicDirectAsyncGateway(manager, features);
+
+ assertThat(gateway).isNotNull();
+ }
+
+ // -------------------------------------------------------------------------
+ // Default flags for a new ReactiveCommonsFeatures instance
+ // -------------------------------------------------------------------------
+
+ @Test
+ void newReactiveCommonsFeaturesShouldHaveAllFlagsFalse() {
+ ReactiveCommonsFeatures features = new ReactiveCommonsFeatures();
+ assertThat(features.isListenEvents()).isFalse();
+ assertThat(features.isListenNotificationEvents()).isFalse();
+ assertThat(features.isListenCommands()).isFalse();
+ assertThat(features.isListenQueries()).isFalse();
+ assertThat(features.isSendEvents()).isFalse();
+ assertThat(features.isSendCommands()).isFalse();
+ }
+
+ @Test
+ void withDomainShouldReturnSameInstanceOnSubsequentCalls() {
+ ReactiveCommonsDomainFeatures features = new ReactiveCommonsDomainFeatures();
+ ReactiveCommonsFeatures first = features.withDomain(DEFAULT_DOMAIN);
+ ReactiveCommonsFeatures second = features.withDomain(DEFAULT_DOMAIN);
+ assertThat(first).isSameAs(second);
+ }
+
+ @Test
+ void ofDomainShouldReturnConfiguredFeatures() {
+ ReactiveCommonsDomainFeatures features = new ReactiveCommonsDomainFeatures();
+ features.withDomain(DEFAULT_DOMAIN).setListenEvents(true);
+ assertThat(features.ofDomain(DEFAULT_DOMAIN).isListenEvents()).isTrue();
+ }
+}
From 78917a3c633bbbb3190adaaab4905cf1463ce57c Mon Sep 17 00:00:00 2001
From: Juan C Galvis <8420868+juancgalvis@users.noreply.github.com>
Date: Tue, 11 Aug 2026 15:21:34 -0500
Subject: [PATCH 2/2] fix: allow programmatic features
---
.../config/ReactiveCommonsDynamicConfig.java | 126 +---------------
.../disabled/DisabledDirectAsyncGateway.java | 85 +++++++++++
.../disabled/DisabledDomainEventBus.java | 42 ++++++
.../DisabledDirectAsyncGatewayTest.java | 141 ++++++++++++++++++
.../disabled/DisabledDomainEventBusTest.java | 81 ++++++++++
5 files changed, 351 insertions(+), 124 deletions(-)
create mode 100644 starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/disabled/DisabledDirectAsyncGateway.java
create mode 100644 starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/disabled/DisabledDomainEventBus.java
create mode 100644 starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/disabled/DisabledDirectAsyncGatewayTest.java
create mode 100644 starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/disabled/DisabledDomainEventBusTest.java
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfig.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfig.java
index f82b19c7..aa4c8248 100644
--- a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfig.java
+++ b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsDynamicConfig.java
@@ -3,12 +3,13 @@
import lombok.extern.log4j.Log4j2;
import org.reactivecommons.api.domain.DomainEventBus;
import org.reactivecommons.async.api.DirectAsyncGateway;
+import org.reactivecommons.async.starter.config.disabled.DisabledDirectAsyncGateway;
+import org.reactivecommons.async.starter.config.disabled.DisabledDomainEventBus;
import org.reactivecommons.async.starter.senders.GenericDirectAsyncGateway;
import org.reactivecommons.async.starter.senders.GenericDomainEventBus;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
-import reactor.core.publisher.Mono;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -143,127 +144,4 @@ public DirectAsyncGateway dynamicDirectAsyncGateway(ConnectionManager manager,
return new GenericDirectAsyncGateway(gateways);
}
- // -------------------------------------------------------------------------
- // Disabled no-op implementations used when a sender feature flag is false.
- // These prevent startup failures caused by missing bean definitions while
- // surfacing a clear error at the point where the disabled feature is used.
- // -------------------------------------------------------------------------
-
- private static final String SEND_EVENTS_DISABLED =
- "sendEvents feature is disabled in ReactiveCommonsFeatures. " +
- "Set sendEvents=true to publish domain events.";
-
- private static final String SEND_COMMANDS_DISABLED =
- "sendCommands feature is disabled in ReactiveCommonsFeatures. " +
- "Set sendCommands=true to send commands or queries.";
-
- private static class DisabledDomainEventBus implements DomainEventBus {
- @Override
- public org.reactivestreams.Publisher emit(org.reactivecommons.api.domain.DomainEvent event) {
- return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
- }
-
- @Override
- public org.reactivestreams.Publisher emit(String domain,
- org.reactivecommons.api.domain.DomainEvent event) {
- return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
- }
-
- @Override
- public org.reactivestreams.Publisher emit(io.cloudevents.CloudEvent event) {
- return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
- }
-
- @Override
- public org.reactivestreams.Publisher emit(String domain, io.cloudevents.CloudEvent event) {
- return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
- }
-
- @Override
- public org.reactivestreams.Publisher emit(org.reactivecommons.api.domain.RawMessage event) {
- return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
- }
-
- @Override
- public org.reactivestreams.Publisher emit(String domain,
- org.reactivecommons.api.domain.RawMessage event) {
- return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
- }
- }
-
- private static class DisabledDirectAsyncGateway implements DirectAsyncGateway {
- @Override
- public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName,
- long delayMillis) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName,
- String domain) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName,
- long delayMillis, String domain) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName, long delayMillis) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName, String domain) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName, long delayMillis,
- String domain) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono requestReply(org.reactivecommons.async.api.AsyncQuery query, String targetName,
- Class type) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono requestReply(org.reactivecommons.async.api.AsyncQuery query, String targetName,
- Class type, String domain) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono requestReply(io.cloudevents.CloudEvent query,
- String targetName, Class type) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono requestReply(io.cloudevents.CloudEvent query,
- String targetName, Class type,
- String domain) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
-
- @Override
- public Mono reply(T response, org.reactivecommons.async.api.From from) {
- return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
- }
- }
}
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/disabled/DisabledDirectAsyncGateway.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/disabled/DisabledDirectAsyncGateway.java
new file mode 100644
index 00000000..23c62a7b
--- /dev/null
+++ b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/disabled/DisabledDirectAsyncGateway.java
@@ -0,0 +1,85 @@
+package org.reactivecommons.async.starter.config.disabled;
+
+import org.reactivecommons.async.api.DirectAsyncGateway;
+import reactor.core.publisher.Mono;
+
+public class DisabledDirectAsyncGateway implements DirectAsyncGateway {
+
+ private static final String SEND_COMMANDS_DISABLED =
+ "sendCommands feature is disabled in ReactiveCommonsFeatures. " +
+ "Set sendCommands=true to send commands or queries.";
+
+ @Override
+ public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName,
+ long delayMillis) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName,
+ String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(org.reactivecommons.api.domain.Command command, String targetName,
+ long delayMillis, String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName, long delayMillis) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName, String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono sendCommand(io.cloudevents.CloudEvent command, String targetName, long delayMillis,
+ String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono requestReply(org.reactivecommons.async.api.AsyncQuery query, String targetName,
+ Class type) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono requestReply(org.reactivecommons.async.api.AsyncQuery query, String targetName,
+ Class type, String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono requestReply(io.cloudevents.CloudEvent query,
+ String targetName, Class type) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono requestReply(io.cloudevents.CloudEvent query,
+ String targetName, Class type,
+ String domain) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+
+ @Override
+ public Mono reply(T response, org.reactivecommons.async.api.From from) {
+ return Mono.error(new IllegalStateException(SEND_COMMANDS_DISABLED));
+ }
+}
\ No newline at end of file
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/disabled/DisabledDomainEventBus.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/disabled/DisabledDomainEventBus.java
new file mode 100644
index 00000000..0886033a
--- /dev/null
+++ b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/disabled/DisabledDomainEventBus.java
@@ -0,0 +1,42 @@
+package org.reactivecommons.async.starter.config.disabled;
+
+import org.reactivecommons.api.domain.DomainEventBus;
+import reactor.core.publisher.Mono;
+
+public class DisabledDomainEventBus implements DomainEventBus {
+ private static final String SEND_EVENTS_DISABLED =
+ "sendEvents feature is disabled in ReactiveCommonsFeatures. " +
+ "Set sendEvents=true to publish domain events.";
+
+ @Override
+ public org.reactivestreams.Publisher emit(org.reactivecommons.api.domain.DomainEvent event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+
+ @Override
+ public org.reactivestreams.Publisher emit(String domain,
+ org.reactivecommons.api.domain.DomainEvent event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+
+ @Override
+ public org.reactivestreams.Publisher emit(io.cloudevents.CloudEvent event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+
+ @Override
+ public org.reactivestreams.Publisher emit(String domain, io.cloudevents.CloudEvent event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+
+ @Override
+ public org.reactivestreams.Publisher emit(org.reactivecommons.api.domain.RawMessage event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+
+ @Override
+ public org.reactivestreams.Publisher emit(String domain,
+ org.reactivecommons.api.domain.RawMessage event) {
+ return Mono.error(new IllegalStateException(SEND_EVENTS_DISABLED));
+ }
+}
\ No newline at end of file
diff --git a/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/disabled/DisabledDirectAsyncGatewayTest.java b/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/disabled/DisabledDirectAsyncGatewayTest.java
new file mode 100644
index 00000000..e0d1a1be
--- /dev/null
+++ b/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/disabled/DisabledDirectAsyncGatewayTest.java
@@ -0,0 +1,141 @@
+package org.reactivecommons.async.starter.config.disabled;
+
+import io.cloudevents.CloudEvent;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.reactivecommons.api.domain.Command;
+import org.reactivecommons.async.api.AsyncQuery;
+import org.reactivecommons.async.api.From;
+import reactor.test.StepVerifier;
+
+@ExtendWith(MockitoExtension.class)
+class DisabledDirectAsyncGatewayTest {
+
+ private DisabledDirectAsyncGateway gateway;
+ private static final String ERROR_MESSAGE = "sendCommands feature is disabled in ReactiveCommonsFeatures. " +
+ "Set sendCommands=true to send commands or queries.";
+
+ @Mock
+ private CloudEvent cloudEvent;
+
+ @Mock
+ private Command command;
+
+ @Mock
+ private AsyncQuery query;
+
+ @Mock
+ private From from;
+
+ @BeforeEach
+ void setUp() {
+ gateway = new DisabledDirectAsyncGateway();
+ }
+
+ @Test
+ void shouldThrowErrorWhenSendingCommandWithoutDelay() {
+ StepVerifier.create(gateway.sendCommand(command, "targetName"))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenSendingCommandWithDelay() {
+ StepVerifier.create(gateway.sendCommand(command, "targetName", 1000L))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenSendingCommandWithDomain() {
+ StepVerifier.create(gateway.sendCommand(command, "targetName", "domain"))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenSendingCommandWithDelayAndDomain() {
+ StepVerifier.create(gateway.sendCommand(command, "targetName", 1000L, "domain"))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenSendingCloudEventCommand() {
+ StepVerifier.create(gateway.sendCommand(cloudEvent, "targetName"))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenSendingCloudEventCommandWithDelay() {
+ StepVerifier.create(gateway.sendCommand(cloudEvent, "targetName", 1000L))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenSendingCloudEventCommandWithDomain() {
+ StepVerifier.create(gateway.sendCommand(cloudEvent, "targetName", "domain"))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenSendingCloudEventCommandWithDelayAndDomain() {
+ StepVerifier.create(gateway.sendCommand(cloudEvent, "targetName", 1000L, "domain"))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenRequestingReplyWithQuery() {
+ StepVerifier.create(gateway.requestReply(query, "targetName", String.class))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenRequestingReplyWithQueryAndDomain() {
+ StepVerifier.create(gateway.requestReply(query, "targetName", String.class, "domain"))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenRequestingReplyWithCloudEvent() {
+ StepVerifier.create(gateway.requestReply(cloudEvent, "targetName", CloudEvent.class))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenRequestingReplyWithCloudEventAndDomain() {
+ StepVerifier.create(gateway.requestReply(cloudEvent, "targetName", CloudEvent.class, "domain"))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenReplyingToQuery() {
+ StepVerifier.create(gateway.reply("response", from))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+}
diff --git a/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/disabled/DisabledDomainEventBusTest.java b/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/disabled/DisabledDomainEventBusTest.java
new file mode 100644
index 00000000..60a3e96e
--- /dev/null
+++ b/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/config/disabled/DisabledDomainEventBusTest.java
@@ -0,0 +1,81 @@
+package org.reactivecommons.async.starter.config.disabled;
+
+import io.cloudevents.CloudEvent;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.reactivecommons.api.domain.DomainEvent;
+import org.reactivecommons.api.domain.RawMessage;
+import reactor.test.StepVerifier;
+
+@ExtendWith(MockitoExtension.class)
+class DisabledDomainEventBusTest {
+
+ private DisabledDomainEventBus eventBus;
+ private static final String ERROR_MESSAGE = "sendEvents feature is disabled in ReactiveCommonsFeatures. " +
+ "Set sendEvents=true to publish domain events.";
+
+ @Mock
+ private CloudEvent cloudEvent;
+
+ @Mock
+ private DomainEvent domainEvent;
+
+ @Mock
+ private RawMessage rawMessage;
+
+ @BeforeEach
+ void setUp() {
+ eventBus = new DisabledDomainEventBus();
+ }
+
+ @Test
+ void shouldThrowErrorWhenEmittingDomainEvent() {
+ StepVerifier.create(eventBus.emit(domainEvent))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenEmittingDomainEventWithDomain() {
+ StepVerifier.create(eventBus.emit("domain", domainEvent))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenEmittingCloudEvent() {
+ StepVerifier.create(eventBus.emit(cloudEvent))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenEmittingCloudEventWithDomain() {
+ StepVerifier.create(eventBus.emit("domain", cloudEvent))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenEmittingRawMessage() {
+ StepVerifier.create(eventBus.emit(rawMessage))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+
+ @Test
+ void shouldThrowErrorWhenEmittingRawMessageWithDomain() {
+ StepVerifier.create(eventBus.emit("domain", rawMessage))
+ .expectErrorMatches(error -> error instanceof IllegalStateException &&
+ error.getMessage().equals(ERROR_MESSAGE))
+ .verify();
+ }
+}