Support for infinite streams with spring-cloud-function-web

[fouadh]

Hi,

I am wondering if it is possible to expose an infinite reactive stream with spring-cloud-function-web.

When I try to do it, I get a timeout exception. If I try to expose a finite stream, it works.

Here is an example of a reactive stream I tried:

@Bean
public Supplier<Flux<String>> infinite() {
    return () -> Flux
            .interval(Duration.ofSeconds(1))
            .log()
            .map(counter -> String.format("Counter: %s", counter));

}

And here is how I am calling it from a terminal:

curl -vvv -H "Accept:text/event-stream" localhost:8080/infinite

*   Trying ::1...
* TCP_NODELAY set
* Connected to localhost (::1) port 8080 (#0)
> GET /infinite HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.64.1
> Accept:text/event-stream
> 
< HTTP/1.1 503 
< Content-Length: 0
< Date: Wed, 19 Jan 2022 08:14:36 GMT
< Connection: close
< 
* Closing connection 0

The logs show that the stream is triggered and, after 30 seconds (http timeout I suppose), it is canceled and the error happens:

2022-01-19 09:05:45.049  INFO 2855 --- [     parallel-1] reactor.Flux.Interval.1                  : onNext(27)
2022-01-19 09:05:46.049  INFO 2855 --- [     parallel-1] reactor.Flux.Interval.1                  : onNext(28)
2022-01-19 09:05:47.049  INFO 2855 --- [     parallel-1] reactor.Flux.Interval.1                  : onNext(29)
2022-01-19 09:05:47.426  INFO 2855 --- [nio-8080-exec-5] reactor.Flux.Interval.1                  : cancel()
2022-01-19 09:05:47.429  WARN 2855 --- [nio-8080-exec-5] .w.s.m.s.DefaultHandlerExceptionResolver : Resolved [org.springframework.web.context.request.async.AsyncRequestTimeoutException]

I am using:

• Java 11
• Spring Boot 2.6.2
• Spring Cloud 2021.0.0

I attach a full sample project to this ticket.

20220119-spring-cloud-function-web.zip

Thanks in advance for your help !

Regards,
Fouad

[olegz]

@fouadh Is there anything else you see in the logs as it seems to be working for me

2022-01-24 10:25:35.885  INFO 11225 --- [     parallel-1] reactor.Flux.Interval.1                  : onNext(100)
2022-01-24 10:25:36.885  INFO 11225 --- [     parallel-1] reactor.Flux.Interval.1                  : onNext(101)
2022-01-24 10:25:37.884  INFO 11225 --- [     parallel-1] reactor.Flux.Interval.1                  : onNext(102)
2022-01-24 10:25:38.883  INFO 11225 --- [     parallel-1] reactor.Flux.Interval.1                  : onNext(103)
. . . keeps going

[olegz]

That said, i did discover few other issues with it, so still need fixing, but i believe it is unrelated to your issue, so still wondering

[olegz]

So, I am back-porting it, but the fix is in. I used your code for additional sample and also pushed a client to validate the responses - https://github.com/spring-cloud/spring-cloud-function/blob/main/spring-cloud-function-samples/function-sample/src/main/java/com/example/Client.java

[fouadh]

I've tested it with the 3.2.x version and it works perfectly fine, thank you very much for the fix !