title: Thread safe reactive cache
body: <p>I always get stuck with the reactive way of thinking once I have to start doing multiple external calls. Right now im tasked with implementing the following scenario using Reactor and Webflux.</p>
  
  <ol>
  <li>Get token from cache
  
  <ol>
  <li>Token not found
  
  <ol>
  <li>Request from auth system new token</li>
  <li>Persist token in cache</li>
  </ol></li>
  <li>Token found
  
  <ol>
  <li>Use token for request to external system
  
  <ol>
  <li>External system responds 2xx
  
  <ol>
  <li>Return response</li>
  </ol></li>
  <li>External system responds 403
  
  <ol>
  <li>Invalidate token in cache</li>
  <li>Retry whole flow from the top</li>
  </ol></li>
  </ol></li>
  </ol></li>
  </ol></li>
  </ol>
  
  <p>What I have so far is just an example on how I would like to use the cache:</p>
  
  <pre><code>  public static Mono&lt;ServerResponse&gt; doRequest(
        Function&lt;Mono&lt;Token&gt;, Mono&lt;ClientResponse&gt;&gt; clientRequest,
        Function&lt;Mono&lt;ClientResponse&gt;, Mono&lt;ServerResponse&gt;&gt; errorHandler
    ) {
      Token token = null;
      return token.getToken()
          .transform(clientRequest)
          .flatMap(clientResponse -&gt; {
                if (clientResponse.statusCode().value() == 403) {
                  // HOW WOULD I INVALIDATE ANYTHING HERE? Or where should i do it?
                  return Mono.error(new TokenExpired("Token expired"));
                }
                return Mono.just(clientResponse);
          })
          .transform(errorHandler)
          .retryWhen(companion -&gt; {
        return companion.zipWith(Flux.range(1,4),
            (error, index) -&gt; {
              if (index &lt; 4 &amp;&amp; (error instanceof TokenExpired)) {
                return index;
              } else {
                throw Exceptions.propagate(error);
              }
            });
      });
    }
  </code></pre>
  
  <p>I have looked at <a href="https://github.com/reactor/reactor-addons" rel="nofollow noreferrer">addons</a> cache but failed at understanding how to manually invalidate the cache since its only time based? Also when using the Caffeine cache the behavior of invalidate is undefined in a multi threaded environment. I feel that my use case is standard but has been unable to find any patterns for how to do this.</p>
  
  <p>My two issues I get stuck with are:
  Making sure the updating of the cache only happens once and is non blocking.
  How to invalidate the entry in the cache in a non blocking way.</p>
  
  <p>Really im stuck with how to approach the problem in a reactive way, an accepted answer does not have to be Reactor it could be any reactive library showing the thinking needed to solve the problem. </p>