hantsy / spring-reactive-sample Goto Github PK

Add a sample for Rsocket reactive operations.

In SessionConfig.java MapReactive... does not exist - I'm assuming it's

import org.springframework.session.ReactiveMapSessionRepository;

Hello, I tried to use "boot-data-couchbase" sample but it fails to start, Do you know what is happening? I'm getting the following error:

2018-08-14 11:27:08.518  WARN 15304 --- [           main] onfigReactiveWebServerApplicationContext : Exception encountered during context initialization - cancelling refresh attempt: org.springframework.context.ApplicationContextException: Unable to start reactive web server; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'httpHandler' defined in class path resource [org/springframework/boot/autoconfigure/web/reactive/HttpHandlerAutoConfiguration$AnnotationConfig.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.http.server.reactive.HttpHandler]: Factory method 'httpHandler' threw exception; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'webHandler' defined in class path resource [org/springframework/boot/autoconfigure/web/reactive/WebFluxAutoConfiguration$EnableWebFluxConfiguration.class]: Initialization of bean failed; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'routerFunctionMapping' defined in class path resource [org/springframework/boot/autoconfigure/web/reactive/WebFluxAutoConfiguration$EnableWebFluxConfiguration.class]: Invocation of init method failed; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'org.springframework.web.reactive.function.server.support.RouterFunctionMapping$SortedRouterFunctionsContainer': Unsatisfied dependency expressed through method 'setRouterFunctions' parameter 0; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'routes' defined in com.example.demo.DemoApplication: Unsatisfied dependency expressed through method 'routes' parameter 0; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'postHandler' defined in file [/home/juan/IntelliJ/spring-reactive-sample-master/boot-data-couchbase/target/classes/com/example/demo/PostHandler.class]: Unsatisfied dependency expressed through constructor parameter 0; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type 'com.example.demo.PostRepository' available: expected at least 1 bean which qualifies as autowire candidate. Dependency annotations: {}
2018-08-14 11:27:08.523  INFO 15304 --- [           main] ConditionEvaluationReportLoggingListener : 

Error starting ApplicationContext. To display the conditions report re-run your application with 'debug' enabled.
2018-08-14 11:27:08.603 ERROR 15304 --- [           main] o.s.b.d.LoggingFailureAnalysisReporter   : 

***************************
APPLICATION FAILED TO START
***************************

Description:

Parameter 0 of constructor in com.example.demo.PostHandler required a bean of type 'com.example.demo.PostRepository' that could not be found.


Action:

Consider defining a bean of type 'com.example.demo.PostRepository' in your configuration.


Process finished with exit code 1

Actually, this is not an issue. But, do you have any example or sample which provides guidance for request interceptors or middlewares in spring reactive?
For example,

• https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/web/servlet/config/annotation/WebMvcConfigurerAdapter.html
• http://sparkjava.com/documentation#filters
• http://expressjs.com/en/guide/using-middleware.html

Thanks.

Hi dear Hantsy

Before all I want to thank you for this awesome collection of samples.

I'm working on the spring-reactive-sample-boot-data-mongo, my problem is with something I guess beyond the authentication, I know you have a portion of your documentation about it (https://github.com/hantsy/spring-reactive-sample#security-for-webflux) the thing is that no mater if I use your same curl with username and password of you sample, I always get this result in postman:

CSRF Token has been associated to this client

and in the backend I got this:

2018-11-21 22:07:28.968 DEBUG 46190 --- [ctor-http-nio-1] o.s.w.s.adapter.HttpWebHandlerAdapter : [feb8e527] HTTP POST "/posts"
2018-11-21 22:07:28.974 DEBUG 46190 --- [ctor-http-nio-1] o.s.w.s.adapter.HttpWebHandlerAdapter : [feb8e527] Completed 403 FORBIDDEN
2018-11-21 22:07:39.423 DEBUG 46190 --- [ctor-http-nio-1] o.s.w.s.adapter.HttpWebHandlerAdapter : [feb8e527] HTTP DELETE "/posts/5bf616be20058db33b1939ad"
2018-11-21 22:07:39.425 DEBUG 46190 --- [ctor-http-nio-1] o.s.w.s.adapter.HttpWebHandlerAdapter : [feb8e527] Completed 403 FORBIDDEN

a bunch of Completed 403 FORBIDDEN messages, do you have an idea what is it ?

Thanks a lot for you time

Best Regards
Diego Vargas

Hi!

I was reading one of your articles and found a small mistake. I see you've merged package boot-data-mongo-auditing with boot-data-mongo, but now link in the article isn't correct. Here is an anchor where incorrect link is located:

https://hantsy.github.io/spring-reactive-sample/data/data-mongo.html#data-auditing-support

As is in the title. I believe you could use the StepVerifier here instead of block(). Correct me if I'm wrong, please :)

@Value("#{@nettyContext.address().getPort()}")

private int port

when test, the console output :

18:14:09.131 [main] DEBUG org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate - Storing ApplicationContext in cache under key [[MergedContextConfiguration@3fd7a715 testClass = ApplicationTests, locations = '{}', classes = '{class com.example.demo.Application}', contextInitializerClasses = '[]', activeProfiles = '{}', propertySourceLocations = '{}', propertySourceProperties = '{server.session.timeout:1}', contextCustomizers = set[[empty]], contextLoader = 'org.springframework.test.context.support.DelegatingSmartContextLoader', parent = [null]]]
18:14:09.132 [main] DEBUG org.springframework.test.context.cache - Spring test ApplicationContext cache statistics: [DefaultContextCache@37ebc9d8 size = 1, maxSize = 32, parentContextCount = 0, hitCount = 0, missCount = 1]
18:14:09.173 [main] ERROR org.springframework.test.context.TestContextManager - Caught exception while allowing TestExecutionListener [org.springframework.test.context.support.DependencyInjectionTestExecutionListener@2b552920] to prepare test instance [com.example.demo.ApplicationTests@64729b1e]
org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'com.example.demo.ApplicationTests': Unsatisfied dependency expressed through field 'port'; nested exception is org.springframework.beans.factory.BeanExpressionException: Expression parsing failed; nested exception is org.springframework.expression.spel.SpelEvaluationException: EL1058E: A problem occurred when trying to resolve bean 'nettyContext':'Could not resolve bean reference against BeanFactory'
at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor$AutowiredFieldElement.inject(AutowiredAnnotationBeanPostProcessor.java:596)
at org.springframework.beans.factory.annotation.InjectionMetadata.inject(InjectionMetadata.java:90)
at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor.postProcessProperties(AutowiredAnnotationBeanPostProcessor.java:374)
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.populateBean(AbstractAutowireCapableBeanFactory.java:1395)
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.autowireBeanProperties(AbstractAutowireCapableBeanFactory.java:391)
at org.springframework.test.context.support.DependencyInjectionTestExecutionListener.injectDependencies(DependencyInjectionTestExecutionListener.java:119)
at org.springframework.test.context.support.DependencyInjectionTestExecutionListener.prepareTestInstance(DependencyInjectionTestExecutionListener.java:83)
at org.springframework.test.context.TestContextManager.prepareTestInstance(TestContextManager.java:246)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.createTest(SpringJUnit4ClassRunner.java:227)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner$1.runReflectiveCall(SpringJUnit4ClassRunner.java:289)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.methodBlock(SpringJUnit4ClassRunner.java:291)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:246)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190)
at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java:89)
at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java:41)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:541)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:763)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java:463)
at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java:209)
Caused by: org.springframework.beans.factory.BeanExpressionException: Expression parsing failed; nested exception is org.springframework.expression.spel.SpelEvaluationException: EL1058E: A problem occurred when trying to resolve bean 'nettyContext':'Could not resolve bean reference against BeanFactory'
at org.springframework.context.expression.StandardBeanExpressionResolver.evaluate(StandardBeanExpressionResolver.java:164)
at org.springframework.beans.factory.support.AbstractBeanFactory.evaluateBeanDefinitionString(AbstractBeanFactory.java:1481)
at org.springframework.beans.factory.support.DefaultListableBeanFactory.doResolveDependency(DefaultListableBeanFactory.java:1191)
at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveDependency(DefaultListableBeanFactory.java:1167)
at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor$AutowiredFieldElement.inject(AutowiredAnnotationBeanPostProcessor.java:593)
... 28 common frames omitted
Caused by: org.springframework.expression.spel.SpelEvaluationException: EL1058E: A problem occurred when trying to resolve bean 'nettyContext':'Could not resolve bean reference against BeanFactory'
at org.springframework.expression.spel.ast.BeanReference.getValueInternal(BeanReference.java:59)
at org.springframework.expression.spel.ast.CompoundExpression.getValueRef(CompoundExpression.java:53)
at org.springframework.expression.spel.ast.CompoundExpression.getValueInternal(CompoundExpression.java:89)
at org.springframework.expression.spel.ast.SpelNodeImpl.getValue(SpelNodeImpl.java:109)
at org.springframework.expression.spel.standard.SpelExpression.getValue(SpelExpression.java:265)
at org.springframework.context.expression.StandardBeanExpressionResolver.evaluate(StandardBeanExpressionResolver.java:161)
... 32 common frames omitted
Caused by: org.springframework.expression.AccessException: Could not resolve bean reference against BeanFactory
at org.springframework.context.expression.BeanFactoryResolver.resolve(BeanFactoryResolver.java:54)
at org.springframework.expression.spel.ast.BeanReference.getValueInternal(BeanReference.java:55)
... 37 common frames omitted
Caused by: org.springframework.beans.factory.NoSuchBeanDefinitionException: No bean named 'nettyContext' available
at org.springframework.beans.factory.support.DefaultListableBeanFactory.getBeanDefinition(DefaultListableBeanFactory.java:775)
at org.springframework.beans.factory.support.AbstractBeanFactory.getMergedLocalBeanDefinition(AbstractBeanFactory.java:1221)
at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:294)
at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:199)
at org.springframework.context.expression.BeanFactoryResolver.resolve(BeanFactoryResolver.java:51)
... 38 common frames omitted

Firstly,This is not an issue ,is a question about reactor method onBackpressureBuffer 。
The code is:

 //发布速度远大于消费速度，理应报错，但是并没有。
        Flux.interval(Duration.of(1, ChronoUnit.MILLIS))
                .onBackpressureBuffer(10, x -> System.out.println("over value：" + x))
                .subscribe(
                        i -> {
                            System.out.println(i);
                            try {
                                Thread.sleep(100);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        });
        TimeUnit.MINUTES.sleep(10);

The code run well, but it seems should get Exception...
And while use rxjava2 ,the code is:

        Observable.interval(1, TimeUnit.MILLISECONDS)
                .toFlowable(BackpressureStrategy.BUFFER)
                .onBackpressureBuffer(256)
                .observeOn(Schedulers.newThread())
                .subscribe(
                        i -> {
                            System.out.println(i);
                            try {
                                Thread.sleep(100);
                            } catch (Exception e) {
                            }
                        });
        TimeUnit.SECONDS.sleep(1);

And then I get exception

0
1
2
io.reactivex.exceptions.OnErrorNotImplementedException: Buffer is full
	at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
	at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
	at io.reactivex.internal.subscribers.LambdaSubscriber.onError(LambdaSubscriber.java:79)
	at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.checkTerminated(FlowableObserveOn.java:207)
	at io.reactivex.internal.operators.flowable.FlowableObserveOn$ObserveOnSubscriber.runAsync(FlowableObserveOn.java:392)
	at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.run(FlowableObserveOn.java:176)
	at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:66)
	at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:57)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

Why the result is different ?

It would be great if Kafka samples would be added. This is a pretty common use case that doesn't have lots of examples. Especially when combined with libraries like Spring Data.

Originally posted by @almogtavor in #260

I have a problem that is when i using spring-boot-starter-data-mongodb-reactive and it is a lot of data,
but i can't return all data beacuse of too much.At this moment,i must be using limit,but i don't know how to limit,can you tell me or show me demo ? thanks!

大哥，我在stackoverflow看到你的提问，现在我想知道你解决了吗？就是webflux的websocket怎么在不接收消息的情况，主动服务端推送消息啊

In the Java9 section , when i run the ApplicationTests ,the test result is error , so with the picture

And the code follows:
`@RestController
@RequestMapping(value = "/posts")
public class PostController {

@GetMapping
public Flow.Publisher<Post> all() {
    SubmissionPublisher publisher = new SubmissionPublisher();
    publisher.submit(new Post(1L, "post one", "content of post one"));
    publisher.submit(new Post(2L, "post two", "content of post two"));

    return publisher;
}

}`

But when i modify the the code as the below ,the response id correct:

`@RestController
@RequestMapping(value = "/posts")
public class PostController {

@GetMapping
public Flow.Publisher<Post> all() {
   return JdkFlowAdapter.publisherToFlowPublisher(Flux.just(new Post(2L, "post two", "content of post two")));
}

}`

Hi @hantsy : Thanks a lot for putting our this useful sample code/tutorial. A great show case of spring reactive.

It would be great if you can also put together a performance gain benefit comparison between using spring reactive/ and typical spring mvc, focusing Reactive key features like backpressure or async, non-blocking I/O. For example, let's say we flush the Mongo DB with hundreds of thousands of write "Post", can we show somewhere in the logs that the server is telling the client to slowdown etc.

Regards,
Tuan

Hi. Do you have guideline on how to use WebFilter to validate a custom request body??

For example,

The server is receiving the following body like

{
// Other key value pair
  "persons": [
    {
      "firstname": "",
      "lastname": "",
      "ID": ""
    }
  ],
// Other key value pair
}

The ID field has to send to another server for validation and if one of the person's IDs is invalid, we stop proceeding with the whole request. I want to use Webfilter to decouple this business logic from the main logic. How can I do it??

There is a Reactor RabbitMQ under Reactor, add a Reactor RabbitMQ example.