How to use project-reactor's CacheMono which backed by Caffeine cache.

CaffeineBackedCachingTests.java

package reactor.caching;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Expiry;
import lombok.Builder;
import lombok.Getter;
import org.junit.Test;
import org.mockito.Mockito;
import reactor.cache.CacheMono;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Signal;
import reactor.test.StepVerifier;

import java.io.Serializable;
import java.time.Duration;
import java.time.LocalDateTime;

/**
 * @author zakyalvan
 */
public class CaffeineBackedCachingTests {
    @Test
    public void testCaffeineBackedReactorCaching() {
        String sessionKey = "sample.session";

        Cache<String, Session> sessionCache = Mockito.spy(Caffeine.newBuilder()
                .maximumSize(1)
                .expireAfter(new Expiry<String, Session>() {
                    @Override
                    public long expireAfterCreate(String key, Session value, long currentTime) {
                        System.out.println("Create session cache with duration : " + value.getExpiryAfter().toNanos() + " ns");
                        return value.getExpiryAfter().toNanos();
                    }

                    @Override
                    public long expireAfterUpdate(String key, Session value, long currentTime, long currentDuration) {
                        System.out.println("Update session cache with duration : " + value.getExpiryAfter().toNanos() + " ns");
                        return value.getExpiryAfter().toNanos();
                    }

                    @Override
                    public long expireAfterRead(String key, Session value, long currentTime, long currentDuration) {
                        System.out.println("After read session, current duration : " + currentDuration);
                        return currentDuration;
                    }
                })
                .build());

        Mono<Session> cachedSession = CacheMono.lookup(key -> Mono.justOrEmpty(sessionCache.getIfPresent(key)).map(Signal::next), sessionKey)
                .onCacheMissResume(() -> Mono.just(Session.builder().createdTime(LocalDateTime.now()).expiryAfter(Duration.ofMillis(1000)).build()))
                .andWriteWith((key, signal) -> Mono.fromRunnable(() -> sessionCache.put(sessionKey, signal.get())));

        StepVerifier.create(cachedSession.delayElement(Duration.ofMillis(100)).repeat(12))
                .expectSubscription().thenAwait()
                .expectNextCount(13)
                .expectComplete()
                .verify();

        Mockito.verify(sessionCache, Mockito.times(13)).getIfPresent(Mockito.anyString());
        Mockito.verify(sessionCache, Mockito.times(2)).put(Mockito.anyString(), Mockito.any(Session.class));
    }

    @Getter
    @Builder
    @SuppressWarnings("serial")
    static class Session implements Serializable {
        private String id;
        private LocalDateTime createdTime;
        private Duration expiryAfter;
    }
}