Skip to content

Commit 98a9c39

Browse files
jfgomezvelezjfgomezvelez
committed
solucion de conflictos por merge, se realiza separacion de listener de notificaciones
2 parents 24638d6 + 1df461a commit 98a9c39

File tree

15 files changed

+378
-21
lines changed

15 files changed

+378
-21
lines changed

‎README.md‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,31 @@ Don't forget to add the implementation dependency to the main spring boot module
128128
}
129129
```
130130

131+
### Domain Event-Listener
132+
Reactive commons has four types of listeners implemented, available to be registered in the application via the **HandlerRegistry**, each of them is designed to tackle
133+
common requirements for listeners in event based applications and abstracts the behavior of event flow in every situation (Varying for example in retrying strategy, dead letter events, sources and so on).
134+
135+
The available event listeners are:
136+
- Domain Event Listener
137+
- Query Event Listener
138+
- Command Listener
139+
- Notification Listener
140+
141+
Example Code:
142+
```java
143+
public HandlerRegistry notificationEvents() {
144+
return HandlerRegistry.register()
145+
.listenNotificationEvent("gatewayRouteAdded", message -> {
146+
System.out.println("Refreshing instance");
147+
return Mono.empty();
148+
},GatewayEvent.class);
149+
}
150+
```
151+
152+
The above code shows how to handle a notification event (Notification event: an event that should be handled by
153+
every running instance of a microservice, e.g: notify to every instance that a configuration setting has changed
154+
and has to do a hot reload from a persistent source of that data).
155+
131156
### Request-Reply
132157
Example Code:
133158

‎async/async-commons-api/src/main/java/org/reactivecommons/async/api/HandlerRegistry.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ public static HandlerRegistry register(){
2525
private final List<RegisteredQueryHandler<?, ?>> handlers = new CopyOnWriteArrayList<>();
2626
private final List<RegisteredEventListener> eventListeners = new CopyOnWriteArrayList<>();
2727
private final List<RegisteredCommandHandler> commandHandlers = new CopyOnWriteArrayList<>();
28+
private final List<RegisteredEventListener> eventNotificationListener = new CopyOnWriteArrayList<>();
2829

2930
public <T> HandlerRegistry listenEvent(String eventName, EventHandler<T> fn, Class<T> eventClass){
3031
eventListeners.add(new RegisteredEventListener<>(eventName, fn, eventClass));
@@ -36,6 +37,11 @@ public <T> HandlerRegistry listenEvent(String eventName, EventHandler<T> handler
3637
return this;
3738
}
3839

40+
public <T> HandlerRegistry listenNotificationEvent(String eventName, EventHandler<T> fn, Class<T> eventClass){
41+
eventNotificationListener.add(new RegisteredEventListener<>(eventName, fn, eventClass));
42+
return this;
43+
}
44+
3945
public <T> HandlerRegistry handleCommand(String commandName, CommandHandler<T> fn, Class<T> commandClass){
4046
commandHandlers.add(new RegisteredCommandHandler<>(commandName, fn, commandClass));
4147
return this;

‎async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/MessageListenersConfig.java‎

Lines changed: 114 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,39 @@
11
package org.reactivecommons.async.impl.config;
22

33
import lombok.RequiredArgsConstructor;
4+
import org.reactivecommons.async.api.DefaultCommandHandler;
5+
import org.reactivecommons.async.api.DefaultQueryHandler;
6+
import org.reactivecommons.async.api.DynamicRegistry;
7+
import org.reactivecommons.async.api.HandlerRegistry;
8+
import org.reactivecommons.async.api.handlers.registered.RegisteredCommandHandler;
9+
import org.reactivecommons.async.api.handlers.registered.RegisteredEventListener;
10+
import org.reactivecommons.async.api.handlers.registered.RegisteredQueryHandler;
411
import org.reactivecommons.async.impl.DiscardNotifier;
12+
import org.reactivecommons.async.impl.DynamicRegistryImp;
513
import org.reactivecommons.async.impl.HandlerResolver;
614
import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
15+
import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
716
import org.reactivecommons.async.impl.config.props.AsyncProps;
817
import org.reactivecommons.async.impl.converters.MessageConverter;
18+
import org.reactivecommons.async.impl.listeners.ApplicationCommandListener;
919
import org.reactivecommons.async.impl.listeners.ApplicationEventListener;
20+
import org.reactivecommons.async.impl.listeners.ApplicationNotificationListener;
21+
import org.reactivecommons.async.impl.listeners.ApplicationQueryListener;
1022
import org.springframework.beans.factory.annotation.Value;
23+
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
24+
import org.springframework.context.ApplicationContext;
1125
import org.springframework.context.annotation.Bean;
1226
import org.springframework.context.annotation.Configuration;
1327
import org.springframework.context.annotation.Import;
28+
import reactor.core.publisher.Mono;
29+
30+
import java.util.Map;
31+
import java.util.concurrent.ConcurrentHashMap;
32+
import java.util.concurrent.ConcurrentMap;
1433

1534
@Configuration
1635
@RequiredArgsConstructor
36+
@Deprecated
1737
@Import(RabbitMqConfig.class)
1838
public class MessageListenersConfig {
1939

@@ -28,11 +48,103 @@ public ApplicationEventListener eventListener(HandlerResolver resolver, MessageC
2848
ReactiveMessageListener receiver, DiscardNotifier discardNotifier) {
2949
final ApplicationEventListener listener = new ApplicationEventListener(receiver,
3050
appName + ".subsEvents", resolver, asyncProps.getDomain().getEvents().getExchange(),
31-
messageConverter, asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(), asyncProps.getRetryDelay(),asyncProps.getDomain().getEvents().getMaxLengthBytes(),
32-
discardNotifier);
51+
messageConverter, asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(), asyncProps.getRetryDelay(),
52+
asyncProps.getDomain().getEvents().getMaxLengthBytes(), discardNotifier);
53+
listener.startListener();
54+
return listener;
55+
}
3356

57+
@Bean
58+
public ApplicationNotificationListener eventNotificationListener(HandlerResolver resolver, MessageConverter messageConverter,
59+
ReactiveMessageListener receiver, DiscardNotifier discardNotifier) {
60+
final ApplicationNotificationListener listener = new ApplicationNotificationListener(
61+
receiver,
62+
asyncProps.getDomain().getEvents().getExchange(),
63+
asyncProps.getNotificationProps().getQueueName(appName),
64+
resolver,
65+
messageConverter,
66+
discardNotifier);
3467
listener.startListener();
68+
return listener;
69+
}
3570

71+
@Bean //TODO: move to own config (QueryListenerConfig)
72+
public ApplicationQueryListener queryListener(MessageConverter converter, HandlerResolver resolver,
73+
ReactiveMessageSender sender, ReactiveMessageListener rlistener,
74+
DiscardNotifier discardNotifier) {
75+
final ApplicationQueryListener listener = new ApplicationQueryListener(rlistener,
76+
appName + ".query", resolver, sender, asyncProps.getDirect().getExchange(), converter,
77+
asyncProps.getGlobal().getExchange(), asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(),
78+
asyncProps.getRetryDelay(), asyncProps.getGlobal().getMaxLengthBytes(), discardNotifier);
79+
listener.startListener();
3680
return listener;
3781
}
82+
83+
@Bean
84+
public ApplicationCommandListener applicationCommandListener(ReactiveMessageListener listener,
85+
HandlerResolver resolver, MessageConverter converter,
86+
DiscardNotifier discardNotifier) {
87+
ApplicationCommandListener commandListener = new ApplicationCommandListener(listener, appName, resolver,
88+
asyncProps.getDirect().getExchange(), converter, asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(),
89+
asyncProps.getRetryDelay(), asyncProps.getDirect().getMaxLengthBytes(), discardNotifier);
90+
commandListener.startListener();
91+
return commandListener;
92+
}
93+
94+
@Bean
95+
public DynamicRegistry dynamicRegistry(HandlerResolver resolver, ReactiveMessageListener listener, IBrokerConfigProps props) {
96+
return new DynamicRegistryImp(resolver, listener.getTopologyCreator(), props);
97+
}
98+
99+
@Bean
100+
public HandlerResolver resolver(ApplicationContext context, DefaultCommandHandler defaultCommandHandler) {
101+
final Map<String, HandlerRegistry> registries = context.getBeansOfType(HandlerRegistry.class);
102+
103+
final ConcurrentMap<String, RegisteredQueryHandler> handlers = registries
104+
.values().stream()
105+
.flatMap(r -> r.getHandlers().stream())
106+
.collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
107+
ConcurrentHashMap::putAll);
108+
109+
final ConcurrentMap<String, RegisteredEventListener> eventListeners = registries
110+
.values().stream()
111+
.flatMap(r -> r.getEventListeners().stream())
112+
.collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
113+
ConcurrentHashMap::putAll);
114+
115+
final ConcurrentMap<String, RegisteredCommandHandler> commandHandlers = registries
116+
.values().stream()
117+
.flatMap(r -> r.getCommandHandlers().stream())
118+
.collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
119+
ConcurrentHashMap::putAll);
120+
121+
final ConcurrentMap<String, RegisteredEventListener> eventNotificationListener = registries
122+
.values()
123+
.stream()
124+
.flatMap(r -> r.getEventNotificationListener().stream())
125+
.collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
126+
ConcurrentHashMap::putAll);
127+
128+
return new HandlerResolver(handlers, eventListeners, commandHandlers, eventNotificationListener) {
129+
@Override
130+
@SuppressWarnings("unchecked")
131+
public <T> RegisteredCommandHandler<T> getCommandHandler(String path) {
132+
final RegisteredCommandHandler<T> handler = super.getCommandHandler(path);
133+
return handler != null ? handler : new RegisteredCommandHandler<>("", defaultCommandHandler, Object.class);
134+
}
135+
};
136+
}
137+
138+
@Bean
139+
@ConditionalOnMissingBean
140+
public DefaultQueryHandler defaultHandler() {
141+
return (DefaultQueryHandler<Object, Object>) command ->
142+
Mono.error(new RuntimeException("No Handler Registered"));
143+
}
144+
145+
@Bean
146+
@ConditionalOnMissingBean
147+
public DefaultCommandHandler defaultCommandHandler() {
148+
return message -> Mono.error(new RuntimeException("No Handler Registered"));
149+
}
38150
}

‎async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/NotificacionListenersConfig.java‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
package org.reactivecommons.async.impl.config;
2+
3+
import lombok.RequiredArgsConstructor;
4+
import org.reactivecommons.async.impl.DiscardNotifier;
5+
import org.reactivecommons.async.impl.HandlerResolver;
6+
import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
7+
import org.reactivecommons.async.impl.config.props.AsyncProps;
8+
import org.reactivecommons.async.impl.converters.MessageConverter;
9+
import org.reactivecommons.async.impl.listeners.ApplicationEventListener;
10+
import org.reactivecommons.async.impl.listeners.ApplicationNotificationListener;
11+
import org.springframework.beans.factory.annotation.Value;
12+
import org.springframework.context.annotation.Bean;
13+
import org.springframework.context.annotation.Configuration;
14+
import org.springframework.context.annotation.Import;
15+
16+
@Configuration
17+
@RequiredArgsConstructor
18+
@Import(RabbitMqConfig.class)
19+
public class NotificacionListenersConfig {
20+
21+
@Value("${spring.application.name}")
22+
private String appName;
23+
24+
private final AsyncProps asyncProps;
25+
26+
@Bean
27+
public ApplicationNotificationListener eventNotificationListener(HandlerResolver resolver, MessageConverter messageConverter,
28+
ReactiveMessageListener receiver, DiscardNotifier discardNotifier) {
29+
final ApplicationNotificationListener listener = new ApplicationNotificationListener(
30+
receiver,
31+
asyncProps.getDomain().getEvents().getExchange(),
32+
asyncProps.getNotificationProps().getQueueName(appName),
33+
resolver,
34+
messageConverter,
35+
discardNotifier);
36+
listener.startListener();
37+
return listener;
38+
}
39+
}

‎async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/RabbitMqConfig.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,7 @@ public ReactiveMessageListener messageListener(ConnectionFactoryProvider provide
100100

101101
@Bean
102102
@ConditionalOnMissingBean
103-
public ConnectionFactoryProvider connectionFactory(RabbitProperties properties) {
103+
public ConnectionFactoryProvider rabbitRConnectionFactory(RabbitProperties properties) {
104104
final ConnectionFactory factory = new ConnectionFactory();
105105
PropertyMapper map = PropertyMapper.get();
106106
map.from(properties::determineHost).whenNonNull().to(factory::setHost);
@@ -168,7 +168,14 @@ public HandlerResolver resolver(ApplicationContext context, DefaultCommandHandle
168168
.collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
169169
ConcurrentHashMap::putAll);
170170

171-
return new HandlerResolver(handlers, eventListeners, commandHandlers) {
171+
final ConcurrentMap<String, RegisteredEventListener> eventNotificationListener = registries
172+
.values()
173+
.stream()
174+
.flatMap(r -> r.getEventNotificationListener().stream())
175+
.collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
176+
ConcurrentHashMap::putAll);
177+
178+
return new HandlerResolver(handlers, eventListeners, commandHandlers, eventNotificationListener) {
172179
@Override
173180
@SuppressWarnings("unchecked")
174181
public <T> RegisteredCommandHandler<T> getCommandHandler(String path) {

‎async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/annotations/EnableMessageListeners.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import org.reactivecommons.async.impl.config.CommandListenersConfig;
44
import org.reactivecommons.async.impl.config.EventListenersConfig;
5+
import org.reactivecommons.async.impl.config.NotificacionListenersConfig;
56
import org.reactivecommons.async.impl.config.QueryListenerConfig;
67
import org.springframework.context.annotation.Configuration;
78
import org.springframework.context.annotation.Import;
@@ -18,7 +19,7 @@
1819
@Retention(RetentionPolicy.RUNTIME)
1920
@Target({ElementType.TYPE})
2021
@Documented
21-
@Import({CommandListenersConfig.class, QueryListenerConfig.class, EventListenersConfig.class})
22+
@Import({CommandListenersConfig.class, QueryListenerConfig.class, EventListenersConfig.class, NotificacionListenersConfig.class})
2223
@Configuration
2324
public @interface EnableMessageListeners {
2425
}

‎async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/annotations/EnableNotificationListener.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
package org.reactivecommons.async.impl.config.annotations;
2+
3+
import org.reactivecommons.async.impl.config.EventBusConfig;
4+
import org.reactivecommons.async.impl.config.NotificacionListenersConfig;
5+
import org.springframework.context.annotation.Configuration;
6+
import org.springframework.context.annotation.Import;
7+
8+
import java.lang.annotation.*;
9+
10+
11+
@Retention(RetentionPolicy.RUNTIME)
12+
@Target({ElementType.TYPE})
13+
@Documented
14+
@Import(NotificacionListenersConfig.class)
15+
@Configuration
16+
public @interface EnableNotificationListener {
17+
}
18+
19+
20+

‎async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/props/AsyncProps.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,9 @@ public class AsyncProps {
2323
@NestedConfigurationProperty
2424
private GlobalProps global = new GlobalProps();
2525

26+
@NestedConfigurationProperty
27+
private NotificationProps notificationProps = new NotificationProps();
28+
2629
private Integer maxRetries = 10;
2730

2831
private Integer prefetchCount = 250;

‎async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/props/BrokerConfigProps.java‎

Lines changed: 2 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import lombok.Getter;
44
import lombok.RequiredArgsConstructor;
55
import org.reactivecommons.async.impl.config.IBrokerConfigProps;
6+
import org.reactivecommons.async.impl.utils.NameGenerator;
67
import org.springframework.beans.factory.annotation.Value;
78
import org.springframework.context.annotation.Configuration;
89
import org.springframework.util.Base64Utils;
@@ -19,9 +20,7 @@ public class BrokerConfigProps implements IBrokerConfigProps {
1920

2021
@Value("${spring.application.name}")
2122
private String appName;
22-
2323
private final AsyncProps asyncProps;
24-
2524
private final AtomicReference<String> replyQueueName = new AtomicReference<>();
2625

2726
@Override
@@ -43,7 +42,7 @@ public String getCommandsQueue() {
4342
public String getReplyQueue() {
4443
final String name = replyQueueName.get();
4544
if (name == null) {
46-
final String replyName = newRandomQueueName();
45+
final String replyName = NameGenerator.generateNameFrom(appName);
4746
if (replyQueueName.compareAndSet(null, replyName)) {
4847
return replyName;
4948
} else {
@@ -62,14 +61,4 @@ public String getDomainEventsExchangeName() {
6261
public String getDirectMessagesExchangeName() {
6362
return asyncProps.getDirect().getExchange();
6463
}
65-
66-
private String newRandomQueueName() {
67-
UUID uuid = UUID.randomUUID();
68-
ByteBuffer bb = ByteBuffer.wrap(new byte[16]);
69-
bb.putLong(uuid.getMostSignificantBits())
70-
.putLong(uuid.getLeastSignificantBits());
71-
return appName + Base64Utils.encodeToUrlSafeString(bb.array())
72-
.replaceAll("=", "");
73-
}
74-
7564
}

‎async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/props/NotificationProps.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
package org.reactivecommons.async.impl.config.props;
2+
3+
import lombok.Getter;
4+
import lombok.RequiredArgsConstructor;
5+
import org.reactivecommons.async.impl.utils.NameGenerator;
6+
7+
import java.util.concurrent.atomic.AtomicReference;
8+
9+
@Getter
10+
@RequiredArgsConstructor
11+
public class NotificationProps {
12+
13+
private final AtomicReference<String> queueName = new AtomicReference<>();
14+
private final String queueSuffix = "notification";
15+
16+
public String getQueueName(String applicationName) {
17+
final String name = this.queueName.get();
18+
if(name == null) return getGeneratedName(applicationName);
19+
return name;
20+
}
21+
22+
private String getGeneratedName(String applicationName) {
23+
String generatedName = NameGenerator.generateNameFrom(applicationName, queueSuffix);
24+
return this.queueName
25+
.compareAndSet(null, generatedName) ?
26+
generatedName : this.queueName.get();
27+
}
28+
}

0 commit comments

Comments
 (0)