Always empty, not working as expected

Author: ZNevzz

src/main/java/spring/znevzz/reactive/controller/CacheController.java

@@ -10,6 +10,8 @@
 import spring.znevzz.reactive.bean.SimpleCacheRequest;
 import spring.znevzz.reactive.handler.IRequestHandler;
 
+import java.util.Optional;
+
 @AllArgsConstructor
 @RestController
 public class CacheController {
@@ -23,7 +25,17 @@ public String welcome() {
 
     @PostMapping("/cache")
     public Flux<ICacheResponse> fromCache(@RequestBody SimpleCacheRequest request) {
-        return handler.getFromCache(request);
+        Flux<ICacheResponse> firstFetch = handler.viewFromCache(request).log("viewFromCache");
+        Flux firstResult = firstFetch
+                .map(ICacheResponse::getPayload)
+                .filter(Optional::isPresent)
+                .log("result of viewFromCache");
+
+        firstFetch
+                .switchIfEmpty(handler.getFromCache(request))
+                .log("secondFetch");
+
+        return firstResult;
     }
 
 }

src/main/java/spring/znevzz/reactive/handler/CacheRequestHandler.java

@@ -29,9 +29,16 @@ public Mono<ICacheResponse> get(ICacheRequest iCacheRequest) {
     }
 
     @Override
-    public Flux<ICacheResponse> getFromCache(ICacheRequest request) {
+    public Flux<ICacheResponse> viewFromCache(ICacheRequest request) {
         return manager.handleCacheRequest(request, Request.VIEW)
                 .timeout(Duration.ofSeconds(5))
                 .retry(2);
     }
+
+    @Override
+    public Flux<ICacheResponse> getFromCache(ICacheRequest request) {
+        return manager.handleCacheRequest(request, Request.GET)
+                .timeout(Duration.ofSeconds(5))
+                .retry(2);
+    }
 }

src/main/java/spring/znevzz/reactive/handler/IRequestHandler.java

@@ -12,6 +12,8 @@ public interface  IRequestHandler<T,R> extends HandlerFunction {
 
     Mono<R> get(T t);
 
+    Flux<R> viewFromCache(T t);
+
     Flux<R> getFromCache(T t);
 
     default <T extends ServerResponse> Mono<T> welcome(ServerRequest serverRequest) {

src/main/java/spring/znevzz/reactive/service/CacheManager.java

@@ -31,7 +31,7 @@ public Flux<ICacheResponse> handleCacheRequest(ICacheRequest request, Request vi
 
         delegateAction(request, action);
 
-        return responses.get(request);
+        return responses.get(request).log();
     }
 
     private void delegateAction(ICacheRequest request, Action action) {

src/main/java/spring/znevzz/reactive/service/DataService.java

@@ -1,15 +1,15 @@
 package spring.znevzz.reactive.service;
 
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import spring.znevzz.reactive.bean.ICacheRequest;
 
 import java.time.Duration;
 
 public class DataService {
 
-    public Flux<String> someAPI(ICacheRequest request){
-        return Flux.just("{}")
-                .delayElements(Duration.ofSeconds(10))
+    public Mono<String> someAPI(ICacheRequest request){
+        return Mono.just("{}")
                 .log();
     }
 }

src/main/java/spring/znevzz/reactive/service/SimpleCacheService.java

@@ -2,6 +2,7 @@
 
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import spring.znevzz.reactive.bean.ICacheRequest;
 import spring.znevzz.reactive.bean.ICacheResponse;
 import spring.znevzz.reactive.bean.SimpleCacheResponse;
@@ -24,11 +25,13 @@ public Flux<ICacheResponse> get(ICacheRequest request) {
 
     @Override
     public Flux<ICacheResponse> put(ICacheRequest request) {
-        Flux<String> stringFlux = service.someAPI(request);
-        cache.put(request.getPayload(), stringFlux.blockFirst());
+
+        service.someAPI(request)
+                .subscribe(response -> cache.put(request.getPayload(), response));
+
         return Flux.just(
                 new SimpleCacheResponse("Fetching"),
-                new SimpleCacheResponse(stringFlux.blockFirst())
+                new SimpleCacheResponse(cache.get(request.getPayload()).toString())
         );
     }
 

src/test/java/spring/znevzz/reactive/controller/CacheControllerTest.java

@@ -56,6 +56,23 @@ void getFromCache() {
                 .expectStatus()
                 .is2xxSuccessful();
     }
+
+
+    @Test
+    void getFromCacheCheckResponse() {
+        SimpleCacheRequest request = new SimpleCacheRequest();
+        request.setUser("test");
+        request.setPayload("{}");
+        webTestClient.post()
+                .uri("/cache")
+                .accept(MediaType.APPLICATION_JSON)
+                .bodyValue(request)
+                .exchange()
+                .expectStatus()
+                .isOk()
+                .expectBody()
+                .json("[{}]");
+    }
     @TestConfiguration
     static class TestConfig {
         @Bean