Reactive Streams: Spring WebFlux - subscribe to existing publisher

4
votes

I am currently migrating our existing Spring asynchronous REST architecture to Spring's new WebFlux library and have a question around joining multiple requests so that they can listen for the same published response.

Use Case is as follows:

  1. Client A connects to our web server and requests data
  2. We hit our cache to check if we have the data there
  3. We don't, so we go and retrieve this data (Client A has subscribed and waits for a response)
  4. Client B connects to our web server and requests the same data (hits the same endpoint)
  5. We check the cache, data is still not there
  6. As we are already fetching this data for Client A we don't want to make another request, however, we also do not want to turn Client B away. Client B should be able to listen for the same information

How can Client B subscribe to the same response stream that Client A is waiting for?

1
javaspringspring-webfluxreactivereactive-streams
Could you give code snippets to show how the caching library API looks like? Also, what is the expected behavior here? I assume that the cache may not contain the raw response to be sent to the client so it's about sharing the cached data at the service layer, not necessarily at the web layer? – Brian Clozel
I'm actually just wondering if it's possible with reactive patterns. The cache is used in the example to show why a request might not get a response immediately. I'm wondering if two requests hit the same endpoint, is it possible with WebFlux to join them and allow them to both wait on the same response? Obviously this can't be done with classic REST and Spring, but I was hoping this sort of behaviour was possible with reactive streams. – wild_nothing

1 Answers

8
votes

"Client A has subscribed and waits for a response" I suppose the request is coded as a Mono and client A sibscribes to it literally:

Subscriber<Response> clientA = ... Mono<Response> request = makeRequest(...); request.subscribe(clientA);

then clientB should subscribe the same way:

Subscriber<Response> clientB = ... request.subscribe(clientB);

Moreover, the cache should contain not the previously saved response data, but the requests themselves, of type Mono<Response>. Then, if such a request is found in the cache, new clients simply subscribe to it, regardless of was that request already completed or not.