marcialrosales/reactor-rabbitmq-streams-getting-started

Getting started tutorials for Reactor RabbitMQ Streams

Getting started tutorials for Reactor RabbitMQ Streams

Getting started tutorials to learn how to use Reactor RabbitMQ Streams API. Tutorials are Spring Boot applications. They depend on reactor-rabbitmq-streams-autoconfigure library.

Reference documentation found at https://pivotal.github.io/reactor-rabbitmq-streams/docs/current/

Javadocs found at https://pivotal.github.io/reactor-rabbitmq-streams/api/reactor-rabbitmq-streams/current/

reactor-rabbitmq-streams-getting-started's Issues

Question: Does reactor-rabbitmq-streams have a future?

Hi!

Just curious because I wasn't able to find any public roadmaps, is reactor-rabbitmq-streams still something that pivotal will push forward with, and will it end up being opensource?

Autoconfigure rootSchemaFolder Bug (Windows)

I know this isn't the repo for the main project but since that is not available I thought I would let you know here that it is essentially not possible to use the rabbit.schema.rootSchemaFolder property without crashing the application.

I've included the stack trace below but basically from what I can see in the de-compiled sources the issue is that once ReactiveRabbitBuilder.loadSchemasFromFilesystem() populates the rootSchemaFolderURL URL from the config you end up with something that looks like file:/C:/Users/.../target/classes/avro (notice the leading slash before the /C:) and when this value is accessed via Paths.get(rootSchemaFolderURL.getPath()) you end up with the exception below:

java.nio.file.InvalidPathException: Illegal char <:> at index 2: /C:/Users/.../target/classes/avro

This can mitigated by changing the call to: Paths.get(rootSchemaFolderURL.toURI()) which should work cross-platform with no issue.

Thanks for all the great example and the project as a whole, overall has been a great experience!

org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'edgeConfiguration': Unsatisfied dependency expressed through field 'rabbit'; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'defaultRabbitEndpoint' defined in class path resource [com/pivotal/rabbitmq/ReactiveRabbitAutoConfiguration.class]: Unsatisfied dependency expressed through method 'defaultRabbitEndpoint' parameter 0; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'reactiveRabbit' defined in class path resource [com/pivotal/rabbitmq/ReactiveRabbitAutoConfiguration.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.pivotal.rabbitmq.ReactiveRabbit]: Factory method 'reactiveRabbit' threw exception; nested exception is java.nio.file.InvalidPathException: Illegal char <:> at index 2: /C:/Users/Alex/Git/icgc-argo/workflow-graph-edge/target/classes/avro
	at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor$AutowiredFieldElement.inject(AutowiredAnnotationBeanPostProcessor.java:643) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.annotation.InjectionMetadata.inject(InjectionMetadata.java:130) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor.postProcessProperties(AutowiredAnnotationBeanPostProcessor.java:399) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.populateBean(AbstractAutowireCapableBeanFactory.java:1420) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:593) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:516) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:324) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:226) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:322) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:202) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:897) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:879) ~[spring-context-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:551) ~[spring-context-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:758) ~[spring-boot-2.3.2.RELEASE.jar:2.3.2.RELEASE]
	at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:750) ~[spring-boot-2.3.2.RELEASE.jar:2.3.2.RELEASE]
	at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397) ~[spring-boot-2.3.2.RELEASE.jar:2.3.2.RELEASE]
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:315) ~[spring-boot-2.3.2.RELEASE.jar:2.3.2.RELEASE]
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:1237) ~[spring-boot-2.3.2.RELEASE.jar:2.3.2.RELEASE]
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226) ~[spring-boot-2.3.2.RELEASE.jar:2.3.2.RELEASE]
	at org.icgc_argo.workflow_graph_edge.WorkflowGraphEdgeApplication.main(WorkflowGraphEdgeApplication.java:10) ~[classes/:na]
Caused by: org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'defaultRabbitEndpoint' defined in class path resource [com/pivotal/rabbitmq/ReactiveRabbitAutoConfiguration.class]: Unsatisfied dependency expressed through method 'defaultRabbitEndpoint' parameter 0; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'reactiveRabbit' defined in class path resource [com/pivotal/rabbitmq/ReactiveRabbitAutoConfiguration.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.pivotal.rabbitmq.ReactiveRabbit]: Factory method 'reactiveRabbit' threw exception; nested exception is java.nio.file.InvalidPathException: Illegal char <:> at index 2: /C:/Users/Alex/Git/icgc-argo/workflow-graph-edge/target/classes/avro
	at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:797) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:538) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1336) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1176) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:556) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:516) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:324) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:226) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:322) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:202) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.config.DependencyDescriptor.resolveCandidate(DependencyDescriptor.java:276) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.doResolveDependency(DefaultListableBeanFactory.java:1307) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveDependency(DefaultListableBeanFactory.java:1227) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor$AutowiredFieldElement.inject(AutowiredAnnotationBeanPostProcessor.java:640) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	... 19 common frames omitted
Caused by: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'reactiveRabbit' defined in class path resource [com/pivotal/rabbitmq/ReactiveRabbitAutoConfiguration.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.pivotal.rabbitmq.ReactiveRabbit]: Factory method 'reactiveRabbit' threw exception; nested exception is java.nio.file.InvalidPathException: Illegal char <:> at index 2: /C:/Users/Alex/Git/icgc-argo/workflow-graph-edge/target/classes/avro
	at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:655) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:635) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1336) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1176) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:556) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:516) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:324) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:226) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:322) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:202) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.config.DependencyDescriptor.resolveCandidate(DependencyDescriptor.java:276) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.doResolveDependency(DefaultListableBeanFactory.java:1307) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveDependency(DefaultListableBeanFactory.java:1227) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.ConstructorResolver.resolveAutowiredArgument(ConstructorResolver.java:884) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:788) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	... 32 common frames omitted
Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.pivotal.rabbitmq.ReactiveRabbit]: Factory method 'reactiveRabbit' threw exception; nested exception is java.nio.file.InvalidPathException: Illegal char <:> at index 2: /C:/Users/Alex/Git/icgc-argo/workflow-graph-edge/target/classes/avro
	at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:185) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:650) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	... 46 common frames omitted
Caused by: java.nio.file.InvalidPathException: Illegal char <:> at index 2: /C:/Users/Alex/Git/icgc-argo/workflow-graph-edge/target/classes/avro
	at java.base/sun.nio.fs.WindowsPathParser.normalize(WindowsPathParser.java:182) ~[na:na]
	at java.base/sun.nio.fs.WindowsPathParser.parse(WindowsPathParser.java:153) ~[na:na]
	at java.base/sun.nio.fs.WindowsPathParser.parse(WindowsPathParser.java:77) ~[na:na]
	at java.base/sun.nio.fs.WindowsPath.parse(WindowsPath.java:92) ~[na:na]
	at java.base/sun.nio.fs.WindowsFileSystem.getPath(WindowsFileSystem.java:229) ~[na:na]
	at java.base/java.nio.file.Path.of(Path.java:147) ~[na:na]
	at java.base/java.nio.file.Paths.get(Paths.java:69) ~[na:na]
	at com.pivotal.rabbitmq.ReactiveRabbitBuilder.loadSchemasFromFilesystem(ReactiveRabbitBuilder.java:201) ~[reactor-rabbitmq-streams-0.0.5.jar:na]
	at com.pivotal.rabbitmq.ReactiveRabbitBuilder.schemaManager(ReactiveRabbitBuilder.java:280) ~[reactor-rabbitmq-streams-0.0.5.jar:na]
	at com.pivotal.rabbitmq.ReactiveRabbitBuilder.build(ReactiveRabbitBuilder.java:101) ~[reactor-rabbitmq-streams-0.0.5.jar:na]
	at com.pivotal.rabbitmq.ReactiveRabbitAutoConfiguration.reactiveRabbit(ReactiveRabbitAutoConfiguration.java:63) ~[reactor-rabbitmq-streams-autoconfigure-0.0.5.jar:na]
	at com.pivotal.rabbitmq.ReactiveRabbitAutoConfiguration$$EnhancerBySpringCGLIB$$f090410d.CGLIB$reactiveRabbit$2(<generated>) ~[reactor-rabbitmq-streams-autoconfigure-0.0.5.jar:na]
	at com.pivotal.rabbitmq.ReactiveRabbitAutoConfiguration$$EnhancerBySpringCGLIB$$f090410d$$FastClassBySpringCGLIB$$15581659.invoke(<generated>) ~[reactor-rabbitmq-streams-autoconfigure-0.0.5.jar:na]
	at org.springframework.cglib.proxy.MethodProxy.invokeSuper(MethodProxy.java:244) ~[spring-core-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.intercept(ConfigurationClassEnhancer.java:331) ~[spring-context-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	at com.pivotal.rabbitmq.ReactiveRabbitAutoConfiguration$$EnhancerBySpringCGLIB$$f090410d.reactiveRabbit(<generated>) ~[reactor-rabbitmq-streams-autoconfigure-0.0.5.jar:na]
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:na]
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
	at java.base/java.lang.reflect.Method.invoke(Method.java:566) ~[na:na]
	at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:154) ~[spring-beans-5.2.8.RELEASE.jar:5.2.8.RELEASE]
	... 47 common frames omitted

Support for spring boot 2.4.x

Attempting to upgrade to spring boot 2.4.x is resulting in the following exception:

An attempt was made to call a method that does not exist. The attempt was made from the following location:

    com.pivotal.rabbitmq.RabbitEndpoint$HttpClientSupplier.get(RabbitEndpoint.java:163)

The following method did not exist:

    'reactor.netty.http.client.HttpClient reactor.netty.http.client.HttpClient.addressSupplier(java.util.function.Supplier)'

The method's class, reactor.netty.http.client.HttpClient, is available from the following locations:

    jar:file:/.m2/repository/io/projectreactor/netty/reactor-netty-http/1.0.1/reactor-netty-http-1.0.1.jar!/reactor/netty/http/client/HttpClient.class

The class hierarchy was loaded from the following locations:

    reactor.netty.http.client.HttpClient: file:/.m2/repository/io/projectreactor/netty/reactor-netty-http/1.0.1/reactor-netty-http-1.0.1.jar
    reactor.netty.transport.ClientTransport: file:/.m2/repository/io/projectreactor/netty/reactor-netty-core/1.0.1/reactor-netty-core-1.0.1.jar
    reactor.netty.transport.Transport: file:/.m2/repository/io/projectreactor/netty/reactor-netty-core/1.0.1/reactor-netty-core-1.0.1.jar


Action:

Correct the classpath of your application so that it contains a single, compatible version of reactor.netty.http.client.HttpClient

OnDemandSource does not continue sending messages after the RabbitMQ Connection is lost and recovered

OnDemandSource does not continue sending messages after the RabbitMQ Connection is lost and recovered

Reproducer environment: On-demand RabbitMQ for PCF version 1.21 (1.20 introduced hot-swap which created new VMs with new IPs rather than reusing existing IPs)

1. Launch a producer application which uses OnDemandSource
2. Perform a deployment recreate operation
3. Observe that the library is able to reconnect but the producer does not resume its operation

Support Quorum Queues

Up until v0.0.8, we could only declare classic queues via the TopologyBuilder.

Quorum queue features which are common with classic queues, such as deal-letter-exchanges, are declared via the same api calls. To declare quorum we need to use the method called quourum(). Similarly to what we do when we had to refer to features exclusive to classic queues, we had to call classic().