reactive-commons/reactive-commons-java

Common abstractions for reactive microservices

reactive-commons-java

The purpose of reactive-commons is to provide a set of abstractions and implementations over different patterns and practices that make the foundation of a reactive microservices architecture.

Docs: https://reactivecommons.org/reactive-commons-java/

Other projects: https://github.com/bancolombia

Sponsor by: https://medium.com/bancolombia-tech

Even though the main purpose is to provide such abstractions in a mostly generic way such abstractions would be of little use without a concrete implementation so we provide some implementations in a best effors maner that aim to be easy to change, personalize and extend.

The first approach to this work was to release a very simple abstractions and a corresponding implementation over asyncronous message driven communication between microservices build on top of project-reactor and spring boot.

reactive-commons-java's Issues

Tolerant reader in all message receivers

Listeners should implement tolerant reader pattern.

Listen event on all instances

Currently when an Event is received it is balanced between all the instances of an app, It would be useful on some cases that all the instances get notified

Integrate acceptance test with travis.

We must integrate some acceptance test with travis ci pipeline.

Add default Message processing timeout and make it configurable

Add some unit tests to NameGenerator class

Please add some unit tests to NameGenerator class of module async-commons

Actuator shows UP when rabbitmq is down

When rabbitmq is DOWN and the app is trying to reconnect the actuator endpoint shows that the app is healthy, the expected behavior would be that it returns DOWN

Error Using deprecated '-debug' fallback for parameter name resolution. Compile the affected code with '-parameters' instead or avoid its introspection: org.springframework.amqp.rabbit.core.RabbitAdmin

Hola buenas tardes, estoy implementando reactive-commons con RabbitMQ pero no he podido iniciar una api exponiendo un evento, relaciono a continuación:
por.xml :

4.0.0

org.springframework.boot
spring-boot-starter-parent
3.1.1


com.example
demo
0.0.1-SNAPSHOT
demo
Demo project for Spring Boot

<java.version>17</java.version>



org.springframework.boot
spring-boot-starter

	<dependency>
		<groupId>org.projectlombok</groupId>
		<artifactId>lombok</artifactId>
		<optional>true</optional>
	</dependency>
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-test</artifactId>
		<scope>test</scope>
	</dependency>
	<dependency>
		<groupId>org.reactivecommons</groupId>
		<artifactId>async-commons-rabbit-starter</artifactId>
		<version>2.0.0</version>
	</dependency>
	<dependency>
		<groupId>org.reactivecommons</groupId>
		<artifactId>domain-events-api</artifactId>
		<version>2.0.0</version>
	</dependency>
	<dependency>
		<groupId>org.reactivecommons</groupId>
		<artifactId>async-commons-api</artifactId>
		<version>2.0.0</version>
	</dependency>
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-web</artifactId>
	</dependency>
	<dependency>
		<groupId>io.projectreactor</groupId>
		<artifactId>reactor-test</artifactId>
	</dependency>
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-amqp</artifactId>
		<version>RELEASE</version>
		<scope>compile</scope>
	</dependency>

</dependencies>

<build>
	<plugins>
		<plugin>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-maven-plugin</artifactId>
			<configuration>
				<excludes>
					<exclude>
						<groupId>org.projectlombok</groupId>
						<artifactId>lombok</artifactId>
					</exclude>
				</excludes>
			</configuration>
		</plugin>
	</plugins>
</build>

implementación:
package com.example.demo.event_driven;

import org.reactivecommons.async.api.HandlerRegistry;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import reactor.core.publisher.Mono;

@configuration
@EnableMessageListeners
public class PersonListener {

public final String EVENT_PERSON_GET_INFO = "event.subscribe.person.get.info";

@Bean
public HandlerRegistry notificationEvents() {
    return HandlerRegistry.register()
            .serveQuery(EVENT_PERSON_GET_INFO, this::getInfoByPersonId, PersonQuery.class);
}

private Mono<PersonReply> getInfoByPersonId(PersonQuery query) {
    return Mono.just(PersonReply.builder().name("Diego").address("manzana ### casa ###").build());

    }
}

Error en log:

Using deprecated '-debug' fallback for parameter name resolution. Compile the affected code with '-parameters' instead or avoid its introspection: org.springframework.amqp.rabbit.core.RabbitAdmin

@dericop ya este tipo de implementación esta deprecada? como podría implementarlo? Mil gracias!

Async commons starter is not using the ObjectMapper instance configured by Spring

As the title says, the developer is not able to customize the behavior of the ObjectMapper because it is instantiated and isolated by the JacksonMessageConverter, the expected behavior it's that the Converter should receive the instance with dependency injection

Broken behavior: empty responses are not propagated anymore

In previous versions of reactive commons, when a query handler responded with an empty mono, the empty signal was propagated to the requester, but, since this commit the behavior was broken.

@juancgalvis is this change intentional?

Increase coverage by testing classes (QueryExecutor, EventExecutor and CommandExecutor)

Please add required unit tests to cover the mentioned classes defined in the async-commons module

Internal representation for Handlers

Inside Listeners, the types: RegisteredQueryHandler, RegisteredEventListener and RegisteredCommandHandler must have a single internal representation in order to centralize common behaviour.

Type inference not working for queries

Fix type inference in query handler

Support backoff strategies for message retries

Assign retry delay to a message following some criteria, like exponential backoff

Invoke all registered handlers for a event

A handler with x.y.event should be invoked the same as x.y.# handler in the same application.

Unwrap String in request-reply when reply is a String

Unwrap string in request-reply when reply is a plain string as: "test1" (Modify serializer)

Graceful shutdown

The connection with the broker and the message handling should be gracefully stopped. Today, When a spring application is shutdown, reactive commons keeps receiving messages and failing because other beans were destroyed.

Update DynamicRegistryImp reactive commons SQS to new DynamicRegistry especification

On the branch feature/sqs there are the modules that implements reactive commons for sqs, in the module async-sqs the class DynamicRegistryImp is implementing a deprecated method of DynamicRegistry, update this class to the new specification.

Dropped messages listener

Notify a parameterized listener when a message is dropped for any reason, that for logging purposes

Wildcard '#' support

Hello there,

I've had an issue when I tried to use the hash character '#' when I subscribed to an event with listenEvent method with this wildcard supported by Rabbit.
Currently, I'm using version 2.0.3 and I'm getting a NullPointerException error:

java.lang.NullPointerException: Cannot invoke "org.reactivecommons.async.api.handlers.registered.RegisteredEventListener.getInputClass()" because "handler" is null
	at org.reactivecommons.async.rabbit.listeners.ApplicationEventListener.rawMessageHandler(ApplicationEventListener.java:84) ~[async-rabbit-2.0.3.jar:na]
	at org.reactivecommons.async.rabbit.listeners.GenericMessageListener.lambda$computeRawMessageHandler$4(GenericMessageListener.java:167) ~[async-rabbit-2.0.3.jar:na]
	at java.base/java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1708) ~[na:na]
	at org.reactivecommons.async.rabbit.listeners.GenericMessageListener.computeRawMessageHandler(GenericMessageListener.java:166) ~[async-rabbit-2.0.3.jar:na]
	at org.reactivecommons.async.rabbit.listeners.GenericMessageListener.getExecutor(GenericMessageListener.java:162) ~[async-rabbit-2.0.3.jar:na]
	at org.reactivecommons.async.rabbit.listeners.GenericMessageListener.handle(GenericMessageListener.java:97) ~[async-rabbit-2.0.3.jar:na]
	at org.reactivecommons.async.rabbit.listeners.GenericMessageListener.lambda$consumeFaultTolerant$3(GenericMessageListener.java:140) ~[async-rabbit-2.0.3.jar:na]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:386) ~[reactor-core-3.5.7.jar:3.5.7]
	at reactor.core.publisher.FluxCreate$BufferAsyncSink.drain(FluxCreate.java:814) ~[reactor-core-3.5.7.jar:3.5.7]
	at reactor.core.publisher.FluxCreate$BufferAsyncSink.next(FluxCreate.java:739) ~[reactor-core-3.5.7.jar:3.5.7]
	at reactor.core.publisher.FluxCreate$SerializedFluxSink.next(FluxCreate.java:161) ~[reactor-core-3.5.7.jar:3.5.7]
	at reactor.rabbitmq.Receiver.lambda$null$9(Receiver.java:216) ~[reactor-rabbitmq-1.5.5.jar:1.5.5]
	at com.rabbitmq.client.impl.recovery.AutorecoveringChannel$2.handleDelivery(AutorecoveringChannel.java:588) ~[amqp-client-5.17.0.jar:5.17.0]
	at com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:149) ~[amqp-client-5.17.0.jar:5.17.0]
	at com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:111) ~[amqp-client-5.17.0.jar:5.17.0]
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[na:na]
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[na:na]
	at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]

Thanks for sharing this project.

Test de class TopologyCreator of module async-rabbit

Test de class TopologyCreator of module async-rabbit

Remove subsEvents binding when listenEvent is removed

When a listenEvent is removed from the handler registry, the binding between the domainEvents and the subsEvents should be removed too

Reactor rabbit uses different channel for resource management

How to reproduce:
Execute the SampleReceiverApp in reactive commons with different credentials than the ones that rabbitmq uses by default, for example modifying application.properties with:

spring.rabbitmq.host=remotehost.com
spring.rabbitmq.virtual-host=myvhost
spring.rabbitmq.username=user
spring.rabbitmq.password=password134

When Reactive Commons tries to declare de domainEvents exchange using the reactor-rabbit Sender it's creates a new channel different that the ones provided by the channel pool, using the default connection factory declared here thats because the Sender uses a different channelMono for resource management as you can see here
So it tries to connect with the default rabbitmq credentials on the localhost throwing a connection error

Upgrade to ReactorRabbit 1.2.0

El metodo listenNotificationEvent en el entry-point no esta funcionando

Buenos dias, desde hace dias necesito escuchar un evento usando el metodo listenNotificationEvent pero cuando publico el evento este metodo no recibe ningun mensaje. Estoy usando la version

Intente probar con otras versiones y el error se sigue presentando de la version 2 en adelante. Ademas investigando un poco porque no estaba recibiendo el mensaje se vio que la siguiente clase no se esta levantando el bean

Se creo esa misma clase dentro de la carpeta del entry-point para revisar si el problema se soluciona y efectivamente la clase levanta el bean y el metodo funciona, sin embargo esta solucion se realizo usando la version

Al implementar esta solucion en la version que estoy usando sale un nuevo error al momento de levantar el bean de la clase NotificacionListenersConfig

Parameter 0 of method customEventNotificationListener in co.com.bancolombia.events.config.CustomNotificacionListenersConfig required a bean of type 'org.reactivecommons.async.rabbit.HandlerResolver' that could not be found.

Quiero recordar que la version que estoy usando es

Dynamic listeners registry

Would be useful to have the option to add and remove listeners at runtime. For example in the case where the listeners can be configured by the user at runtime.

The implementation must fulfill the following criteria:

• The configuration must survive after restarts
• If the configuration is changed in a specific replica, it must be updated in all the other replicas.

Add unit tests for classes (RabbitDomainEventBus, RabbitDiscardNotifier)

Add unit tests for classes (RabbitDomainEventBus, RabbitDiscardNotifier)

Add application name to amqp connection

Add the app Name to the AMQP connection name

Adjust to reactive-commons library according to the CloudEvents specification.

Hello everyone,

We found the need for the reactive-commons library to be adjusted so that there is the possibility of listening and emit events and commands with a structure aligned to the CloudEvent specification, as seen in the following structure:

{
    "specversion" : "1.0",
    "type" : "com.github.pull_request.opened",
    "source" : "https://github.com/cloudevents/spec/pull",
    "id" : "A234-1234-1234",

    "time" : "2018-04-05T17:31:00Z",
    "subject" : "123",
    "datacontenttype" : "application/json",
    "data" : { ... },

    "comexampleextension1" : "value",
    "comexampleothervalue" : 5
}

It should be noted that the key comexampleextension1 and comexampleothervalue can be any additional extension context attribute with any name that is needed for each use case, so as many extension context attributes can be included as required, according to the official specification of CloudEvents .

update to Gradle to 5.1.1

As the title says

ClassCastException when requested query has no handler

When a sender does a request to a unhandled query the receiver enters in an infinite bucle with the following exception java.lang.Class cannot be cast to java.lang.reflect.ParameterizedType

In-memory broker

Based on the in-memory message broker defined by Smallrye reactive messaging (https://smallrye.io/smallrye-reactive-messaging/smallrye-reactive-messaging/2/testing/testing.html) it would be nice to have a similar implementation as a utility for implementing integration tests without depending on RabbitMQ or another production broker

Add parameter for maximum message retries

When a message fails for unrecoverable errors it is retried infinite times, it should be dropped after an amount of retries

Query handler returning an empty publisher causes timeout on the receiver

I declared a query handler that sometimes does not emit values, however it does emit the completion signal, I think the completion signal should be propagated to the receiver on such cases

Serve resource handler dinamically

There may be cases where can be needed to, for example, create an event listener with a dynamic event name, that listener must able to register at runtime

Update to ReactorRabbit 1.1.0

Discard queries older than the parameterized timeout

Circular dependency with HandlerResolver and Dynamic registry

Allow multi broker configuration

It would be useful to be able to configure multiple brokers in the started component.

Centralize bintray task configuration in a single place

Bintray configuration is spread across many *.gradle files, we must centralize it in a single place ( bintray.gradle)

Compile error message when using reactive-commons-java and Spring Boot

I am using the reactive-commons-java version 2.0.3 dependency for Rabbit MQ and when running the microservice I get the following error message:

> Task :rabbit-configurator:compileJava
Errors occurred while building effective model from C:\Users\lugomez\.gradle\caches\modules-2\files-2.1\org.reactivecommons\async-rabbit\2.0.3\ec19910dfe70fbce006cec5f604b1d21271089e5\async-rabbit-2.0.3.pom:
	'dependencies.dependency.version' for io.projectreactor.rabbitmq:reactor-rabbitmq:jar is missing. in org.reactivecommons:async-rabbit:2.0.3

> Task :rabbit-configurator:classes

> Task :async-event-bus:compileJava
Errors occurred while building effective model from C:\Users\lugomez\.gradle\caches\modules-2\files-2.1\org.reactivecommons\async-rabbit\2.0.3\ec19910dfe70fbce006cec5f604b1d21271089e5\async-rabbit-2.0.3.pom:
	'dependencies.dependency.version' for io.projectreactor.rabbitmq:reactor-rabbitmq:jar is missing. in org.reactivecommons:async-rabbit:2.0.3

This is my dependency configuration:

dependencies {
    implementation project(':log')
    implementation 'org.springframework:spring-context'
    implementation "org.reactivecommons:async-commons-rabbit-starter:2.0.3"
    implementation "org.springframework.boot:spring-boot-starter-amqp"
}

This message is presented since version 2.0.0 and checking the commit history the version of the dependency io.projectreactor.rabbitmq:reactor-rabbitmq was explicitly removed, and I think that is why the error message.

Although the error message appears, the microservice does not fail and works correctly, I would like to know what configurations should I apply to remove it?

Acerca de Reactive Commons y Secretos AWS Secrets Manager

Al usar la librería y hacer la implementación con el Broker MQ, encuentro documentacion de como configurar el MQ a traves del application.yaml.

Pero si se necesita que ese usuario y contraseña vayan en un secreto en AWS Secrets Manager no veo documentación de como implementarlo con el Reactive Commons, existe alguna interfaz para configurar la conexión al MQ programaticamente? o es necesario evolucionar el Reactive Commons con esta funcionalidad?

separate ApplicationEventListener and ApplicationQueryListener

Actualmente cuando se configura @EnableMessageListeners, se crea 2 exchange por defecto, el relacionado al subsEvents y query, se necesita tener la forma de adicionar con que patron se desea trabajar con subsEvents, query o comandos

Add interaction diagrams to documentation

As you can know the project implements some communication patterns, it could be clearest if the readme includes some graphs for Domain Events, Commands, Async Queries and Notifications.