Skip to content

feat: add support for RabbitMQ mandatory property #145

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

Merged
merged 9 commits into from
Jul 28, 2025
Merged

feat: add support for RabbitMQ mandatory property #145

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
d7b6585
feat: add mandatory property
lugomez29 Jul 25, 2025
874a845
break circular dependency
juancgalvis Jul 25, 2025
03c5ce4
feat: implement UnroutableMessageProcessor and refactor message handling
lugomez29 Jul 25, 2025
28ab2dc
docs: add mandatory property documentation
lugomez29 Jul 25, 2025
51ec18a
docs: mandatory property documentation is translated into English
lugomez29 Jul 25, 2025
f50adda
docs: improve formatting and readability of mandatory property docume…
lugomez29 Jul 25, 2025
1c2fc5d
build: update dependencies
lugomez29 Jul 28, 2025
16aa2f5
build: update version in gradle.properties
lugomez29 Jul 28, 2025
ec1e40b
Update main.gradle
juancgalvis Jul 28, 2025
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
1 change: 1 addition & 0 deletions async/async-commons-api/src/main/java/org/reactivecommons/async/api/HandlerRegistry.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@

@Getter
@NoArgsConstructor(access = AccessLevel.PACKAGE)
@SuppressWarnings({"unchecked", "unused"})
public final class HandlerRegistry {
public static final String DEFAULT_DOMAIN = "app";
private final RegisteredDomainHandlers<RegisteredEventListener<?, ?>> domainEventListeners =
Expand Down
9 changes: 1 addition & 8 deletions ...main/java/org/reactivecommons/async/api/handlers/registered/RegisteredCommandHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,6 @@
package org.reactivecommons.async.api.handlers.registered;

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.reactivecommons.async.api.handlers.CommandHandler;

@RequiredArgsConstructor
@Getter
public class RegisteredCommandHandler<T,D> {
private final String path;
private final CommandHandler<D> handler;
private final Class<T> inputClass;
public record RegisteredCommandHandler<T, D>(String path, CommandHandler<D> handler, Class<T> inputClass) {
}
9 changes: 1 addition & 8 deletions .../main/java/org/reactivecommons/async/api/handlers/registered/RegisteredEventListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,6 @@
package org.reactivecommons.async.api.handlers.registered;

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.reactivecommons.async.api.handlers.EventHandler;

@RequiredArgsConstructor
@Getter
public class RegisteredEventListener<T, D> {
private final String path;
private final EventHandler<D> handler;
private final Class<T> inputClass;
public record RegisteredEventListener<T, D>(String path, EventHandler<D> handler, Class<T> inputClass) {
}
9 changes: 1 addition & 8 deletions ...c/main/java/org/reactivecommons/async/api/handlers/registered/RegisteredQueryHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,6 @@
package org.reactivecommons.async.api.handlers.registered;

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.reactivecommons.async.api.handlers.QueryHandlerDelegate;

@RequiredArgsConstructor
@Getter
public class RegisteredQueryHandler<T, C> {
private final String path;
private final QueryHandlerDelegate<T, C> handler;
private final Class<C> queryClass;
public record RegisteredQueryHandler<T, C>(String path, QueryHandlerDelegate<T, C> handler, Class<C> queryClass) {
}
80 changes: 41 additions & 39 deletions async/async-commons-api/src/test/java/org/reactivecommons/async/api/HandlerRegistryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,8 @@ void shouldListenDomainEvent() {

assertThat(registry.getDomainEventListeners().get(domain))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass,
RegisteredEventListener::getHandler
.extracting(RegisteredEventListener::path, RegisteredEventListener::inputClass,
RegisteredEventListener::handler
)
.containsExactly(name, SomeDataClass.class, eventHandler)).hasSize(1);
}
Expand All @@ -53,8 +53,8 @@ void shouldListenDomainCloudEvent() {

assertThat(registry.getDomainEventListeners().get(domain))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass,
RegisteredEventListener::getHandler
.extracting(RegisteredEventListener::path, RegisteredEventListener::inputClass,
RegisteredEventListener::handler
)
.containsExactly(name, CloudEvent.class, eventHandler)).hasSize(1);
}
Expand All @@ -67,8 +67,8 @@ void shouldListenDomainRawEvent() {

assertThat(registry.getDomainEventListeners().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass,
RegisteredEventListener::getHandler
.extracting(RegisteredEventListener::path, RegisteredEventListener::inputClass,
RegisteredEventListener::handler
)
.containsExactly(name, RawMessage.class, eventHandler)).hasSize(1);
}
Expand All @@ -81,8 +81,8 @@ void shouldListenEvent() {

assertThat(registry.getDomainEventListeners().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass,
RegisteredEventListener::getHandler
.extracting(RegisteredEventListener::path, RegisteredEventListener::inputClass,
RegisteredEventListener::handler
)
.containsExactly(name, SomeDataClass.class, eventHandler)).hasSize(1);
}
Expand All @@ -95,8 +95,8 @@ void shouldListenCloudEvent() {

assertThat(registry.getDomainEventListeners().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass,
RegisteredEventListener::getHandler
.extracting(RegisteredEventListener::path, RegisteredEventListener::inputClass,
RegisteredEventListener::handler
)
.containsExactly(name, CloudEvent.class, eventHandler)).hasSize(1);
}
Expand Down Expand Up @@ -124,14 +124,14 @@ void shouldRegisterPatternEventHandler() {
void shouldRegisterNotificationEventListener() {
registry.listenNotificationEvent(name, message -> Mono.empty(), SomeDataClass.class);
assertThat(registry.getEventNotificationListener().get(DEFAULT_DOMAIN))
.anySatisfy(listener -> assertThat(listener.getPath()).isEqualTo(name));
.anySatisfy(listener -> assertThat(listener.path()).isEqualTo(name));
}

@Test
void shouldRegisterNotificationCloudEventListener() {
registry.listenNotificationCloudEvent(name, message -> Mono.empty());
assertThat(registry.getEventNotificationListener().get(DEFAULT_DOMAIN))
.anySatisfy(listener -> assertThat(listener.getPath()).isEqualTo(name));
.anySatisfy(listener -> assertThat(listener.path()).isEqualTo(name));
}

@Test
Expand All @@ -142,8 +142,8 @@ void shouldRegisterNotificationRawEventListener() {

assertThat(registry.getEventNotificationListener().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass,
RegisteredEventListener::getHandler
.extracting(RegisteredEventListener::path, RegisteredEventListener::inputClass,
RegisteredEventListener::handler
)
.containsExactly(nameRawNotification, RawMessage.class, eventHandler)).hasSize(1);
}
Expand All @@ -156,8 +156,8 @@ void listenEvent() {

assertThat(registry.getDomainEventListeners().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass,
RegisteredEventListener::getHandler
.extracting(RegisteredEventListener::path, RegisteredEventListener::inputClass,
RegisteredEventListener::handler
)
.containsExactly(name, SomeDataClass.class, handler)).hasSize(1);
}
Expand All @@ -170,8 +170,8 @@ void shouldListenDynamicEvent() {

assertThat(registry.getDynamicEventHandlers().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass,
RegisteredEventListener::getHandler
.extracting(RegisteredEventListener::path, RegisteredEventListener::inputClass,
RegisteredEventListener::handler
)
.containsExactly(name, SomeDataClass.class, eventHandler)).hasSize(1);
}
Expand All @@ -184,8 +184,8 @@ void shouldListenDynamicCloudEvent() {

assertThat(registry.getDynamicEventHandlers().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass,
RegisteredEventListener::getHandler
.extracting(RegisteredEventListener::path, RegisteredEventListener::inputClass,
RegisteredEventListener::handler
)
.containsExactly(name, CloudEvent.class, eventHandler)).hasSize(1);
}
Expand All @@ -198,8 +198,8 @@ void handleDomainCommand() {

assertThat(registry.getCommandHandlers().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredCommandHandler::getPath, RegisteredCommandHandler::getInputClass,
RegisteredCommandHandler::getHandler
.extracting(RegisteredCommandHandler::path, RegisteredCommandHandler::inputClass,
RegisteredCommandHandler::handler
)
.containsExactly(name, SomeDataClass.class, handler)).hasSize(1);
}
Expand All @@ -212,8 +212,8 @@ void handleCloudEventCommand() {

assertThat(registry.getCommandHandlers().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredCommandHandler::getPath, RegisteredCommandHandler::getInputClass,
RegisteredCommandHandler::getHandler
.extracting(RegisteredCommandHandler::path, RegisteredCommandHandler::inputClass,
RegisteredCommandHandler::handler
)
.containsExactly(name, CloudEvent.class, cloudCommandHandler)).hasSize(1);
}
Expand All @@ -226,8 +226,8 @@ void handleRawCommand() {

assertThat(registry.getCommandHandlers().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredCommandHandler::getPath, RegisteredCommandHandler::getInputClass,
RegisteredCommandHandler::getHandler
.extracting(RegisteredCommandHandler::path, RegisteredCommandHandler::inputClass,
RegisteredCommandHandler::handler
)
.containsExactly(nameRaw, RawMessage.class, eventHandler)).hasSize(1);
}
Expand All @@ -240,7 +240,7 @@ void shouldServerCloudEventQuery() {

assertThat(registry.getHandlers().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredQueryHandler::getPath, RegisteredQueryHandler::getQueryClass)
.extracting(RegisteredQueryHandler::path, RegisteredQueryHandler::queryClass)
.containsExactly(name, CloudEvent.class)).hasSize(1);
}

Expand All @@ -250,7 +250,7 @@ void handleCommandWithLambda() {

assertThat(registry.getCommandHandlers().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredCommandHandler::getPath, RegisteredCommandHandler::getInputClass)
.extracting(RegisteredCommandHandler::path, RegisteredCommandHandler::inputClass)
.containsExactly(name, SomeDataClass.class)).hasSize(1);
}

Expand All @@ -260,7 +260,7 @@ void serveQueryWithLambda() {
registry.serveQuery(name, message -> Mono.empty(), SomeDataClass.class);
assertThat(registry.getHandlers().get(DEFAULT_DOMAIN))
.anySatisfy(registered -> assertThat(registered)
.extracting(RegisteredQueryHandler::getPath, RegisteredQueryHandler::getQueryClass)
.extracting(RegisteredQueryHandler::path, RegisteredQueryHandler::queryClass)
.containsExactly(name, SomeDataClass.class)).hasSize(1);
}

Expand All @@ -269,9 +269,9 @@ void serveQueryWithTypeInference() {
QueryHandler<SomeDataClass, SomeDataClass> handler = new SomeQueryHandler();
registry.serveQuery(name, handler, SomeDataClass.class);
assertThat(registry.getHandlers().get(DEFAULT_DOMAIN)).anySatisfy(registered -> {
assertThat(registered).extracting(RegisteredQueryHandler::getPath, RegisteredQueryHandler::getQueryClass)
assertThat(registered).extracting(RegisteredQueryHandler::path, RegisteredQueryHandler::queryClass)
.containsExactly(name, SomeDataClass.class);
assertThat(registered).extracting(RegisteredQueryHandler::getHandler)
assertThat(registered).extracting(RegisteredQueryHandler::handler)
.isInstanceOf(QueryHandlerDelegate.class);
}).hasSize(1);
}
Expand All @@ -280,19 +280,21 @@ void serveQueryWithTypeInference() {
void serveQueryDelegate() {
QueryHandlerDelegate<Void, SomeDataClass> handler = new SomeQueryHandlerDelegate();
registry.serveQuery(name, handler, SomeDataClass.class);
assertThat(registry.getHandlers().get(DEFAULT_DOMAIN)).anySatisfy(registered -> {
assertThat(registered).extracting(RegisteredQueryHandler::getPath, RegisteredQueryHandler::getQueryClass)
.containsExactly(name, SomeDataClass.class);
}).hasSize(1);
assertThat(registry.getHandlers().get(DEFAULT_DOMAIN)).anySatisfy(registered ->
assertThat(registered)
.extracting(RegisteredQueryHandler::path, RegisteredQueryHandler::queryClass)
.containsExactly(name, SomeDataClass.class))
.hasSize(1);
}

@Test
void serveQueryDelegateWithLambda() {
registry.serveQuery(name, (from, message) -> Mono.empty(), SomeDataClass.class);
assertThat(registry.getHandlers().get(DEFAULT_DOMAIN)).anySatisfy(registered -> {
assertThat(registered).extracting(RegisteredQueryHandler::getPath, RegisteredQueryHandler::getQueryClass)
.containsExactly(name, SomeDataClass.class);
}).hasSize(1);
assertThat(registry.getHandlers().get(DEFAULT_DOMAIN)).anySatisfy(registered ->
assertThat(registered)
.extracting(RegisteredQueryHandler::path, RegisteredQueryHandler::queryClass)
.containsExactly(name, SomeDataClass.class))
.hasSize(1);
}

private static class SomeQueryHandlerDelegate implements QueryHandlerDelegate<Void, SomeDataClass> {
Expand Down
2 changes: 1 addition & 1 deletion async/async-commons/async-commons.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ dependencies {
compileOnly 'io.projectreactor:reactor-core'
api 'com.fasterxml.jackson.core:jackson-databind'
api 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310'
implementation 'commons-io:commons-io:2.18.0'
implementation 'commons-io:commons-io:2.20.0'
implementation 'io.cloudevents:cloudevents-json-jackson:4.0.1'

testImplementation 'io.projectreactor:reactor-test'
Expand Down
10 changes: 4 additions & 6 deletions async/async-commons/src/main/java/org/reactivecommons/async/commons/HandlerResolver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

@Log
@RequiredArgsConstructor
@SuppressWarnings("unchecked")
public class HandlerResolver {

private final Map<String, RegisteredQueryHandler<?, ?>> queryHandlers;
Expand All @@ -37,19 +38,16 @@ public boolean hasQueryHandlers() {
return !queryHandlers.isEmpty();
}

@SuppressWarnings("unchecked")
public <T, M> RegisteredQueryHandler<T, M> getQueryHandler(String path) {
return (RegisteredQueryHandler<T, M>) queryHandlers
.computeIfAbsent(path, getMatchHandler(queryHandlers));
}

@SuppressWarnings("unchecked")
public <T, D> RegisteredCommandHandler<T, D> getCommandHandler(String path) {
return (RegisteredCommandHandler<T, D>) commandHandlers
.computeIfAbsent(path, getMatchHandler(commandHandlers));
}

@SuppressWarnings("unchecked")
public <T, D> RegisteredEventListener<T, D> getEventListener(String path) {
if (eventListeners.containsKey(path)) {
return (RegisteredEventListener<T, D>) eventListeners.get(path);
Expand Down Expand Up @@ -81,15 +79,15 @@ public List<String> getNotificationNames() {
}

public void addEventListener(RegisteredEventListener<?, ?> listener) {
eventListeners.put(listener.getPath(), listener);
eventListeners.put(listener.path(), listener);
}

public void addQueryHandler(RegisteredQueryHandler<?, ?> handler) {
if (handler.getPath().contains("*") || handler.getPath().contains("#")) {
if (handler.path().contains("*") || handler.path().contains("#")) {
throw new RuntimeException("avoid * or # in dynamic handlers, make sure you have no conflicts with cached" +
" patterns");
}
queryHandlers.put(handler.getPath(), handler);
queryHandlers.put(handler.path(), handler);
}

private <T> Function<String, T> getMatchHandler(Map<String, T> handlers) {
Expand Down
10 changes: 5 additions & 5 deletions ...rc/main/java/org/reactivecommons/async/commons/utils/resolver/HandlerResolverBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ public static HandlerResolver buildResolver(String domain, Map<String, HandlerRe
.getOrDefault(domain, List.of())
.stream())
.collect(ConcurrentHashMap::new, (map, handler)
-> map.put(handler.getPath(), handler), ConcurrentHashMap::putAll
-> map.put(handler.path(), handler), ConcurrentHashMap::putAll
);

final ConcurrentMap<String, RegisteredCommandHandler<?, ?>> commandHandlers = registries
Expand All @@ -39,7 +39,7 @@ public static HandlerResolver buildResolver(String domain, Map<String, HandlerRe
.getOrDefault(domain, List.of())
.stream())
.collect(ConcurrentHashMap::new, (map, handler)
-> map.put(handler.getPath(), handler), ConcurrentHashMap::putAll
-> map.put(handler.path(), handler), ConcurrentHashMap::putAll
);

final ConcurrentMap<String, RegisteredEventListener<?, ?>> eventNotificationListener = registries
Expand All @@ -49,7 +49,7 @@ public static HandlerResolver buildResolver(String domain, Map<String, HandlerRe
.getOrDefault(domain, List.of())
.stream())
.collect(ConcurrentHashMap::new, (map, handler)
-> map.put(handler.getPath(), handler), ConcurrentHashMap::putAll
-> map.put(handler.path(), handler), ConcurrentHashMap::putAll
);

final ConcurrentMap<String, RegisteredEventListener<?, ?>> eventsToBind = getEventsToBind(domain,
Expand Down Expand Up @@ -81,7 +81,7 @@ public <T, D> RegisteredCommandHandler<T, D> getCommandHandler(String path) {
.stream(),
getDynamics(domain, r)))
.collect(ConcurrentHashMap::new, (map, handler)
-> map.put(handler.getPath(), handler), ConcurrentHashMap::putAll
-> map.put(handler.path(), handler), ConcurrentHashMap::putAll
);
}

Expand All @@ -100,7 +100,7 @@ public <T, D> RegisteredCommandHandler<T, D> getCommandHandler(String path) {
return Stream.empty();
})
.collect(ConcurrentHashMap::new, (map, handler)
-> map.put(handler.getPath(), handler), ConcurrentHashMap::putAll
-> map.put(handler.path(), handler), ConcurrentHashMap::putAll
);
}
}
2 changes: 1 addition & 1 deletion ...java/org/reactivecommons/async/commons/utils/matcher/KeyMatcherPerformanceManualTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ class KeyMatcherPerformanceManualTest {


@BeforeEach
public void init() {
void init() {
ClassLoader classLoader = getClass().getClassLoader();
File file = new File(Objects.requireNonNull(classLoader.getResource("candidateNamesForMatching.txt")).getFile());
try {
Expand Down
Loading