Kotlin Coroutines

The optional cocache-coroutines module (6.4+) adds a suspending view over the async API. It is the same cache: the same L2, the same invalidation fences, the same in-flight loads and the same Redis leases. Methods have the same names as the synchronous API, they suspend instead of blocking, and they throw the original exception.

Adding the module

implementation("me.ahoo.cocache:cocache-coroutines")

The version comes from cocache-bom. The module brings kotlinx-coroutines-core with it. No other CoCache module depends on it.

Getting the view

val users = userCache.suspending()            // SuspendingCache<String, User>

suspend fun profile(id: String): User? = users.get(id)

suspend fun rename(id: String, user: User) {
    users.put(id, user)
}

• Cache.suspending() returns a SuspendingCache, and SetCache.suspending() returns a SuspendingSetCache with add, remove and clear; its get returns an empty set, never null, like SetCache.get. AsyncCache.suspending() works too.
• Each method calls the async view and awaits the result. An L2 hit returns at once without suspending, because the async view returns an already-completed future.
• No thread is blocked while CoCache waits for Redis, for another instance's lease, or for your data source.

Data sources

A suspending data source is a SuspendingCacheLoader. Set it in the cache's spec:

cacheSpec<String, User>("user") {
    codec = jacksonCodec()
    suspendingLoader { id -> userRepository.findById(id) }   // a suspend function
}

• suspendingLoader(loader) / suspendingLoader(context, loader) set the spec's asyncLoader. Each load starts a coroutine in context, which defaults to Dispatchers.Default. Pass Dispatchers.IO if the loader calls blocking code.
• The loads run under one SupervisorJob, so one failed load doesn't affect the others. If context has a Job, it is the parent: cancelling it cancels running loads and every later one, and they fail with SourceLoadCancelledException for every caller, synchronous, async or suspending. This holds even when the Job was already cancelled before the load started.
• That SupervisorJob never completes. Give context a long-lived Job, such as an application-wide one, not the Job of a structured scope: that scope would never finish.
• Create the loader once and reuse it. Each suspendingLoader or asAsyncCacheLoader call creates a new SupervisorJob under the Job, so creating one per load piles up child jobs.
• loadAll defaults to calling load concurrently. Override it with one batch query when the source has one.
• suspendingVersionedLoader(loader) / suspendingVersionedLoader(context, loader) (since 6.8) take a SuspendingCacheLoader<K, Versioned<V?>> (return Versioned(null, n) for a key deleted at version n) and set the spec's versionedAsyncLoader, which makes the cache versioned. They behave like suspendingLoader in every other way. The suspending view has putIfNewer, invalidateIfNewer and getVersioned.
• asAsyncCacheLoader() / asAsyncCacheLoader(context) turn a SuspendingCacheLoader into an AsyncCacheLoader if you want to set it yourself. Used that way, it has no re-entry check, because it doesn't know which cache it belongs to.
• Don't call the synchronous API from a coroutine running on a single-thread dispatcher, if the cache's loader runs on that same dispatcher. The thread blocks waiting for a load that needs that thread, so it deadlocks. Use the suspending view there.

A per-call loader works like get(key, loader) on the synchronous API, and it gets the same protection:

val user = users.get(id) { key -> userClient.fetch(key) }

The per-call loader runs in the caller's coroutine context without its Job:

• It runs on the caller's dispatcher. It runs on Dispatchers.Default instead if there is no dispatcher, or if the caller runs on a runBlocking event loop.
• ThreadContextElements go with it, such as a tracing context, or MDC through MDCContext from kotlinx-coroutines-slf4j (plain MDC is a thread-local and doesn't follow a coroutine to another thread).
• It is outside structured concurrency. Other callers may share the load, so cancelling the caller, ending the caller's scope, or even closing the runtime doesn't cancel it. The load's coroutine can outlive them all, until the data source returns.

If a loader itself throws CancellationException while its coroutine is still active (for example from a withTimeout inside it), CoCache rethrows it as SourceLoadCancelledException, whose cause is the original. A load may be shared, and every caller would otherwise take the exception as its own cancellation (launch swallows it silently).

The engine's load-async-timeout applies to both kinds of loader. When it fires, the caller fails with SourceLoadTimeoutException and the lease is released. The loader's coroutine is not cancelled; its late result is discarded. This works the same as for an AsyncCacheLoader.

Cancellation

Cancelling a coroutine that is waiting on the cache ends only that wait. A load that other callers share keeps running and fills the cache, as with the async view. The next read hits.

Exceptions

The suspending methods throw the exception the async view completed with: your loader's exception (or SourceLoadCancelledException, see Data sources), CacheWriteException, SourceLoadRejectedException, SourceLoadTimeoutException and so on. They are never wrapped in CompletionException.

In coroutine debug mode, which is on whenever the JVM runs with -ea, kotlinx-coroutines does two things:

• it may rethrow a copy of the exception with a recovered stack trace (the original is its cause);
• it adds @coroutine#N to thread names.

Production JVMs normally don't run with -ea. If your tests compare exception instances or thread names, set kotlinx.coroutines.debug=off.

Re-entry

A loader that reads the key it is loading, from the same cache, would wait for itself until load-async-timeout. Instead, it fails at once with IllegalStateException, as it does on the synchronous API. The check goes with the loader's coroutine context, so it holds after the loader suspends and resumes on another thread. It covers reads through the suspending view (get, getAll, lookup). Other keys, and other caches, are not affected: for per-call loaders that means caches with other names or other instances, for spec-level loaders caches with other names (see below). Keys are compared as K values, not as encoded storage keys: two unequal keys that encode to the same storage key aren't caught, and fail only at load-async-timeout.

• Per-call loaders are matched by cache instance: a same-named cache on another runtime, as in a multi-node test, is a different cache.
• Spec-level loaders (suspendingLoader) are matched by cache name, because the spec doesn't know its runtime yet.
• A spec-level load has no caller, so it doesn't inherit the caller's check. A cycle across caches, such as A's loader reading B and B's loader reading the same key of A, fails only at load-async-timeout.
• A loader set with asAsyncCacheLoader on its own has no check.

Spring

The starter doesn't look for SuspendingCacheLoader beans. Declare the cache as a CacheSpec bean and set suspendingLoader { … } in it, or adapt the loader yourself with asAsyncCacheLoader() and expose it as an AsyncCacheLoader bean. When the module is on the classpath, the starter registers native-image hints for the suspending views.

WebFlux

In a WebFlux application, controller methods can be suspend funs that call the suspending view. Spring needs kotlinx-coroutines-reactor on the classpath to run them; spring-boot-starter-webflux doesn't bring it, and its version comes from Spring Boot.

@RestController
class ProfileController(
    @Qualifier("profile") profileCache: Cache<String, String>,
    private val profileRepository: ProfileRepository,
) {
    private val profiles = profileCache.suspending()   // create once, keep in a field

    @GetMapping("profile/{id}")
    suspend fun profile(@PathVariable id: String): String? = profiles.get(id)

    @PutMapping("profile/{id}")
    suspend fun update(@PathVariable id: String, @RequestParam nickname: String) {
        profileRepository.save(id, nickname)   // update the source first
        profiles.invalidate(id)                // then invalidate (inside a transaction: after commit)
    }
}

The cocache-example app has the complete version: CoroutineController reads with cache.suspending(), uses a per-call loader, and invalidates after a write; the profile cache in UserCacheConfiguration declares its source with suspendingLoader { … }.