diff --git a/.github/dependabot.yml b/.github/dependabot.yml index b7d0000d8d..f5c62d742d 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -12,8 +12,8 @@ updates: directory: / schedule: interval: daily - - package-ecosystem: maven - directory: examples/coalescing-bulkloader + - package-ecosystem: gradle + directory: examples/coalescing-bulkloader-reactor schedule: interval: daily - package-ecosystem: gradle diff --git a/.github/workflows/examples.yml b/.github/workflows/examples.yml index 46cdd06af4..db7307f4fb 100644 --- a/.github/workflows/examples.yml +++ b/.github/workflows/examples.yml @@ -98,9 +98,9 @@ jobs: - name: Write-behind (rxjava) working-directory: examples/write-behind-rxjava run: ./gradlew build - - name: Coalescing Bulkloader - working-directory: examples/coalescing-bulkloader - run: ./mvnw test + - name: Coalescing Bulkloader (reactor) + working-directory: examples/coalescing-bulkloader-reactor + run: ./gradlew build - name: Hibernate (jcache) working-directory: examples/hibernate run: ./gradlew build diff --git a/examples/coalescing-bulkloader-reactor/README.md b/examples/coalescing-bulkloader-reactor/README.md new file mode 100644 index 0000000000..0c2e6c5e8d --- /dev/null +++ b/examples/coalescing-bulkloader-reactor/README.md @@ -0,0 +1,121 @@ +[Reactor][reactor] data streams facilitate the consolidation of independent asynchronous loads into +batches at the cost of a small buffering delay. The [bufferTimeout][] operator accumulates requests +until reaching a maximum size or time limit. Since each request consists of a key and its pending +result, when the subscriber is notified it performs the batch load and completes the key's future +with its corresponding value. + +It some scenarios it may be desirable to only aggregate cache refreshes rather than imposing delays +on callers awaiting explicit loads. An automated reload initiated by `refreshAfterWrite` will occur +on the first stale request for an entry. While the key is being refreshed the previous value +continues to be returned, in contrast to eviction which forces retrievals to wait until the value +is loaded anew. In such cases, batching these optimistic reloads can minimize the impact on the +source system without adversely affecting the responsiveness of the explicit requests. + +### Refresh coalescing +A [Sink][sink] collects requests, buffering them up to the configured threshold, and subsequently +delivers the batch to the subscriber. The `parallelism` setting determines the number of concurrent +bulk loads that can be executed if the size constraint results in multiple batches. + +```java +public final class CoalescingBulkLoader implements CacheLoader { + private final Function, Map> mappingFunction; + private final Sinks.Many> sink; + + /** + * @param maxSize the maximum entries to collect before performing a bulk request + * @param maxTime the maximum duration to wait before performing a bulk request + * @param parallelism the number of parallel bulk loads that can be performed + * @param mappingFunction the function to compute the values + */ + public CoalescingBulkLoader(int maxSize, Duration maxTime, int parallelism, + Function, Map> mappingFunction) { + this.sink = Sinks.many().unicast().onBackpressureBuffer(); + this.mappingFunction = requireNonNull(mappingFunction); + sink.asFlux() + .bufferTimeout(maxSize, maxTime) + .map(requests -> requests.stream().collect( + toMap(Entry::getKey, Entry::getValue))) + .parallel(parallelism) + .runOn(Schedulers.boundedElastic()) + .subscribe(this::handle); + } +``` + +To ensure immediate responses for explicit loads these calls directly invoke the mapping function, +while the optimistic reloads are instead submitted to the sink. It's worth noting that this call is +`synchronized`, as a sink does not support concurrent submissions. + +```java + @Override public V load(K key) { + return loadAll(Set.of(key)).get(key); + } + + @Override public abstract Map loadAll(Set key) { + return mappingFunction.apply(keys); + } + + @Override public synchronized CompletableFuture asyncReload(K key, V oldValue, Executor e) { + var entry = Map.entry(key, new CompletableFuture()); + sink.tryEmitNext(entry).orThrow(); + return entry.getValue(); + } +``` + +The subscriber receives a batch of requests, each comprising of a key and a pending future result. +It performs the synchronous load and then either completes the key's future with the corresponding +value or an exception if a failure occurs. + +```java + private void handle(Map> requests) { + try { + var results = mappingFunction.apply(requests.keySet()); + requests.forEach((key, result) -> result.complete(results.get(key))); + } catch (Throwable t) { + requests.forEach((key, result) -> result.completeExceptionally(t)); + } + } +``` + +### Async coalescing +The previous logic can be streamlined if all loads should be collected into batches. This approach +is most suitable for an `AsyncLoadingCache` since it does not block any other map operations while +an entry is being loaded. + +```java +public final class CoalescingBulkLoader implements AsyncCacheLoader { + private final Function, Map> mappingFunction; + private final Sinks.Many> sink; + + public CoalescingBulkLoader(int maxSize, Duration maxTime, int parallelism, + Function, Map> mappingFunction) { + this.sink = Sinks.many().unicast().onBackpressureBuffer(); + this.mappingFunction = requireNonNull(mappingFunction); + sink.asFlux() + .bufferTimeout(maxSize, maxTime) + .map(requests -> requests.stream().collect( + toMap(Entry::getKey, Entry::getValue))) + .parallel(parallelism) + .runOn(Schedulers.boundedElastic()) + .subscribe(this::handle); + } + + @Override public synchronized CompletableFuture asyncLoad(K key, Executor e) { + var entry = Map.entry(key, new CompletableFuture()); + sink.tryEmitNext(entry).orThrow(); + return entry.getValue(); + } + + private void handle(Map> requests) { + try { + var results = mappingFunction.apply(requests.keySet()); + requests.forEach((key, result) -> result.complete(results.get(key))); + } catch (Throwable t) { + requests.forEach((key, result) -> result.completeExceptionally(t)); + } + } +} +``` + +[reactor]: https://projectreactor.io +[bufferTimeout]: https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html#bufferTimeout-int-java.time.Duration- +[sink]: https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Sinks.html diff --git a/examples/coalescing-bulkloader-reactor/build.gradle.kts b/examples/coalescing-bulkloader-reactor/build.gradle.kts new file mode 100644 index 0000000000..8f7f075b6a --- /dev/null +++ b/examples/coalescing-bulkloader-reactor/build.gradle.kts @@ -0,0 +1,27 @@ +plugins { + `java-library` + alias(libs.plugins.versions) +} + +dependencies { + implementation(libs.caffeine) + implementation(libs.reactor) + + testImplementation(libs.junit) + testImplementation(libs.truth) +} + +testing.suites { + val test by getting(JvmTestSuite::class) { + useJUnitJupiter() + } +} + +java.toolchain.languageVersion = JavaLanguageVersion.of( + System.getenv("JAVA_VERSION")?.toIntOrNull() ?: 11) + +tasks.withType().configureEach { + javaCompiler = javaToolchains.compilerFor { + languageVersion = java.toolchain.languageVersion + } +} diff --git a/examples/coalescing-bulkloader-reactor/gradle/libs.versions.toml b/examples/coalescing-bulkloader-reactor/gradle/libs.versions.toml new file mode 100644 index 0000000000..0d57cb7c6c --- /dev/null +++ b/examples/coalescing-bulkloader-reactor/gradle/libs.versions.toml @@ -0,0 +1,15 @@ +[versions] +caffeine = "3.1.7" +junit = "5.10.0" +reactor = "3.5.8" +truth = "1.1.5" +versions = "0.47.0" + +[libraries] +caffeine = { module = "com.github.ben-manes.caffeine:caffeine", version.ref = "caffeine" } +junit = { module = "org.junit.jupiter:junit-jupiter", version.ref = "junit" } +reactor = { module = "io.projectreactor:reactor-core", version.ref = "reactor" } +truth = { module = "com.google.truth:truth", version.ref = "truth" } + +[plugins] +versions = { id = "com.github.ben-manes.versions", version.ref = "versions" } diff --git a/examples/coalescing-bulkloader-reactor/gradle/wrapper/gradle-wrapper.jar b/examples/coalescing-bulkloader-reactor/gradle/wrapper/gradle-wrapper.jar new file mode 100644 index 0000000000..7f93135c49 Binary files /dev/null and b/examples/coalescing-bulkloader-reactor/gradle/wrapper/gradle-wrapper.jar differ diff --git a/examples/coalescing-bulkloader-reactor/gradle/wrapper/gradle-wrapper.properties b/examples/coalescing-bulkloader-reactor/gradle/wrapper/gradle-wrapper.properties new file mode 100644 index 0000000000..0fb2837297 --- /dev/null +++ b/examples/coalescing-bulkloader-reactor/gradle/wrapper/gradle-wrapper.properties @@ -0,0 +1,7 @@ +distributionBase=GRADLE_USER_HOME +distributionPath=wrapper/dists +distributionUrl=https\://services.gradle.org/distributions/gradle-8.3-rc-3-bin.zip +networkTimeout=10000 +validateDistributionUrl=true +zipStoreBase=GRADLE_USER_HOME +zipStorePath=wrapper/dists diff --git a/examples/coalescing-bulkloader-reactor/gradlew b/examples/coalescing-bulkloader-reactor/gradlew new file mode 100755 index 0000000000..0adc8e1a53 --- /dev/null +++ b/examples/coalescing-bulkloader-reactor/gradlew @@ -0,0 +1,249 @@ +#!/bin/sh + +# +# Copyright © 2015-2021 the original authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +############################################################################## +# +# Gradle start up script for POSIX generated by Gradle. +# +# Important for running: +# +# (1) You need a POSIX-compliant shell to run this script. If your /bin/sh is +# noncompliant, but you have some other compliant shell such as ksh or +# bash, then to run this script, type that shell name before the whole +# command line, like: +# +# ksh Gradle +# +# Busybox and similar reduced shells will NOT work, because this script +# requires all of these POSIX shell features: +# * functions; +# * expansions «$var», «${var}», «${var:-default}», «${var+SET}», +# «${var#prefix}», «${var%suffix}», and «$( cmd )»; +# * compound commands having a testable exit status, especially «case»; +# * various built-in commands including «command», «set», and «ulimit». +# +# Important for patching: +# +# (2) This script targets any POSIX shell, so it avoids extensions provided +# by Bash, Ksh, etc; in particular arrays are avoided. +# +# The "traditional" practice of packing multiple parameters into a +# space-separated string is a well documented source of bugs and security +# problems, so this is (mostly) avoided, by progressively accumulating +# options in "$@", and eventually passing that to Java. +# +# Where the inherited environment variables (DEFAULT_JVM_OPTS, JAVA_OPTS, +# and GRADLE_OPTS) rely on word-splitting, this is performed explicitly; +# see the in-line comments for details. +# +# There are tweaks for specific operating systems such as AIX, CygWin, +# Darwin, MinGW, and NonStop. +# +# (3) This script is generated from the Groovy template +# https://github.com/gradle/gradle/blob/HEAD/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt +# within the Gradle project. +# +# You can find Gradle at https://github.com/gradle/gradle/. +# +############################################################################## + +# Attempt to set APP_HOME + +# Resolve links: $0 may be a link +app_path=$0 + +# Need this for daisy-chained symlinks. +while + APP_HOME=${app_path%"${app_path##*/}"} # leaves a trailing /; empty if no leading path + [ -h "$app_path" ] +do + ls=$( ls -ld "$app_path" ) + link=${ls#*' -> '} + case $link in #( + /*) app_path=$link ;; #( + *) app_path=$APP_HOME$link ;; + esac +done + +# This is normally unused +# shellcheck disable=SC2034 +APP_BASE_NAME=${0##*/} +# Discard cd standard output in case $CDPATH is set (https://github.com/gradle/gradle/issues/25036) +APP_HOME=$( cd "${APP_HOME:-./}" > /dev/null && pwd -P ) || exit + +# Use the maximum available, or set MAX_FD != -1 to use that value. +MAX_FD=maximum + +warn () { + echo "$*" +} >&2 + +die () { + echo + echo "$*" + echo + exit 1 +} >&2 + +# OS specific support (must be 'true' or 'false'). +cygwin=false +msys=false +darwin=false +nonstop=false +case "$( uname )" in #( + CYGWIN* ) cygwin=true ;; #( + Darwin* ) darwin=true ;; #( + MSYS* | MINGW* ) msys=true ;; #( + NONSTOP* ) nonstop=true ;; +esac + +CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar + + +# Determine the Java command to use to start the JVM. +if [ -n "$JAVA_HOME" ] ; then + if [ -x "$JAVA_HOME/jre/sh/java" ] ; then + # IBM's JDK on AIX uses strange locations for the executables + JAVACMD=$JAVA_HOME/jre/sh/java + else + JAVACMD=$JAVA_HOME/bin/java + fi + if [ ! -x "$JAVACMD" ] ; then + die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME + +Please set the JAVA_HOME variable in your environment to match the +location of your Java installation." + fi +else + JAVACMD=java + if ! command -v java >/dev/null 2>&1 + then + die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. + +Please set the JAVA_HOME variable in your environment to match the +location of your Java installation." + fi +fi + +# Increase the maximum file descriptors if we can. +if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then + case $MAX_FD in #( + max*) + # In POSIX sh, ulimit -H is undefined. That's why the result is checked to see if it worked. + # shellcheck disable=SC3045 + MAX_FD=$( ulimit -H -n ) || + warn "Could not query maximum file descriptor limit" + esac + case $MAX_FD in #( + '' | soft) :;; #( + *) + # In POSIX sh, ulimit -n is undefined. That's why the result is checked to see if it worked. + # shellcheck disable=SC3045 + ulimit -n "$MAX_FD" || + warn "Could not set maximum file descriptor limit to $MAX_FD" + esac +fi + +# Collect all arguments for the java command, stacking in reverse order: +# * args from the command line +# * the main class name +# * -classpath +# * -D...appname settings +# * --module-path (only if needed) +# * DEFAULT_JVM_OPTS, JAVA_OPTS, and GRADLE_OPTS environment variables. + +# For Cygwin or MSYS, switch paths to Windows format before running java +if "$cygwin" || "$msys" ; then + APP_HOME=$( cygpath --path --mixed "$APP_HOME" ) + CLASSPATH=$( cygpath --path --mixed "$CLASSPATH" ) + + JAVACMD=$( cygpath --unix "$JAVACMD" ) + + # Now convert the arguments - kludge to limit ourselves to /bin/sh + for arg do + if + case $arg in #( + -*) false ;; # don't mess with options #( + /?*) t=${arg#/} t=/${t%%/*} # looks like a POSIX filepath + [ -e "$t" ] ;; #( + *) false ;; + esac + then + arg=$( cygpath --path --ignore --mixed "$arg" ) + fi + # Roll the args list around exactly as many times as the number of + # args, so each arg winds up back in the position where it started, but + # possibly modified. + # + # NB: a `for` loop captures its iteration list before it begins, so + # changing the positional parameters here affects neither the number of + # iterations, nor the values presented in `arg`. + shift # remove old arg + set -- "$@" "$arg" # push replacement arg + done +fi + + +# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. +DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"' + +# Collect all arguments for the java command; +# * $DEFAULT_JVM_OPTS, $JAVA_OPTS, and $GRADLE_OPTS can contain fragments of +# shell script including quotes and variable substitutions, so put them in +# double quotes to make sure that they get re-expanded; and +# * put everything else in single quotes, so that it's not re-expanded. + +set -- \ + "-Dorg.gradle.appname=$APP_BASE_NAME" \ + -classpath "$CLASSPATH" \ + org.gradle.wrapper.GradleWrapperMain \ + "$@" + +# Stop when "xargs" is not available. +if ! command -v xargs >/dev/null 2>&1 +then + die "xargs is not available" +fi + +# Use "xargs" to parse quoted args. +# +# With -n1 it outputs one arg per line, with the quotes and backslashes removed. +# +# In Bash we could simply go: +# +# readarray ARGS < <( xargs -n1 <<<"$var" ) && +# set -- "${ARGS[@]}" "$@" +# +# but POSIX shell has neither arrays nor command substitution, so instead we +# post-process each arg (as a line of input to sed) to backslash-escape any +# character that might be a shell metacharacter, then use eval to reverse +# that process (while maintaining the separation between arguments), and wrap +# the whole thing up as a single "set" statement. +# +# This will of course break if any of these variables contains a newline or +# an unmatched quote. +# + +eval "set -- $( + printf '%s\n' "$DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS" | + xargs -n1 | + sed ' s~[^-[:alnum:]+,./:=@_]~\\&~g; ' | + tr '\n' ' ' + )" '"$@"' + +exec "$JAVACMD" "$@" diff --git a/examples/coalescing-bulkloader-reactor/gradlew.bat b/examples/coalescing-bulkloader-reactor/gradlew.bat new file mode 100644 index 0000000000..93e3f59f13 --- /dev/null +++ b/examples/coalescing-bulkloader-reactor/gradlew.bat @@ -0,0 +1,92 @@ +@rem +@rem Copyright 2015 the original author or authors. +@rem +@rem Licensed under the Apache License, Version 2.0 (the "License"); +@rem you may not use this file except in compliance with the License. +@rem You may obtain a copy of the License at +@rem +@rem https://www.apache.org/licenses/LICENSE-2.0 +@rem +@rem Unless required by applicable law or agreed to in writing, software +@rem distributed under the License is distributed on an "AS IS" BASIS, +@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +@rem See the License for the specific language governing permissions and +@rem limitations under the License. +@rem + +@if "%DEBUG%"=="" @echo off +@rem ########################################################################## +@rem +@rem Gradle startup script for Windows +@rem +@rem ########################################################################## + +@rem Set local scope for the variables with windows NT shell +if "%OS%"=="Windows_NT" setlocal + +set DIRNAME=%~dp0 +if "%DIRNAME%"=="" set DIRNAME=. +@rem This is normally unused +set APP_BASE_NAME=%~n0 +set APP_HOME=%DIRNAME% + +@rem Resolve any "." and ".." in APP_HOME to make it shorter. +for %%i in ("%APP_HOME%") do set APP_HOME=%%~fi + +@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. +set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m" + +@rem Find java.exe +if defined JAVA_HOME goto findJavaFromJavaHome + +set JAVA_EXE=java.exe +%JAVA_EXE% -version >NUL 2>&1 +if %ERRORLEVEL% equ 0 goto execute + +echo. +echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. +echo. +echo Please set the JAVA_HOME variable in your environment to match the +echo location of your Java installation. + +goto fail + +:findJavaFromJavaHome +set JAVA_HOME=%JAVA_HOME:"=% +set JAVA_EXE=%JAVA_HOME%/bin/java.exe + +if exist "%JAVA_EXE%" goto execute + +echo. +echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME% +echo. +echo Please set the JAVA_HOME variable in your environment to match the +echo location of your Java installation. + +goto fail + +:execute +@rem Setup the command line + +set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar + + +@rem Execute Gradle +"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %* + +:end +@rem End local scope for the variables with windows NT shell +if %ERRORLEVEL% equ 0 goto mainEnd + +:fail +rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of +rem the _cmd.exe /c_ return code! +set EXIT_CODE=%ERRORLEVEL% +if %EXIT_CODE% equ 0 set EXIT_CODE=1 +if not ""=="%GRADLE_EXIT_CONSOLE%" exit %EXIT_CODE% +exit /b %EXIT_CODE% + +:mainEnd +if "%OS%"=="Windows_NT" endlocal + +:omega diff --git a/examples/coalescing-bulkloader-reactor/settings.gradle.kts b/examples/coalescing-bulkloader-reactor/settings.gradle.kts new file mode 100644 index 0000000000..f0fb67436a --- /dev/null +++ b/examples/coalescing-bulkloader-reactor/settings.gradle.kts @@ -0,0 +1,11 @@ +plugins { + id("org.gradle.toolchains.foojay-resolver-convention") version "0.6.0" +} + +dependencyResolutionManagement { + repositories { + mavenCentral() + } +} + +rootProject.name = "coalescing-bulkloader-reactor" diff --git a/examples/coalescing-bulkloader-reactor/src/main/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkLoader.java b/examples/coalescing-bulkloader-reactor/src/main/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkLoader.java new file mode 100644 index 0000000000..da3704fe53 --- /dev/null +++ b/examples/coalescing-bulkloader-reactor/src/main/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkLoader.java @@ -0,0 +1,115 @@ +/* + * Copyright 2019 Guus C. Bloemsma. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.github.benmanes.caffeine.examples.coalescing.bulkloader; + +import static java.util.Objects.requireNonNull; +import static java.util.stream.Collectors.toMap; + +import java.time.Duration; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; +import java.util.function.Function; + +import com.github.benmanes.caffeine.cache.AsyncCacheLoader; +import com.google.errorprone.annotations.CanIgnoreReturnValue; + +import reactor.core.publisher.Sinks; +import reactor.core.scheduler.Schedulers; + +/** + * An {@link AsyncCacheLoader} that accumulates keys until a specified maximum size or time limit is + * reached, at which point it performs a bulk load. This strategy assumes that the efficiency gained + * through batching justifies the wait. + * + * @author guus@bloemsma.net (Guus C. Bloemsma) + * @author ben.manes@gmail.com (Ben Manes) + */ +public final class CoalescingBulkLoader implements AsyncCacheLoader { + private final Sinks.Many>> sink; + private final Function, Map> mappingFunction; + + private CoalescingBulkLoader(Builder builder) { + this.mappingFunction = builder.mappingFunction; + sink = Sinks.many().unicast().onBackpressureBuffer(); + sink.asFlux() + .bufferTimeout(builder.maxSize, builder.maxTime) + .map(requests -> requests.stream().collect( + toMap(Entry::getKey, Entry::getValue))) + .parallel(builder.parallelism) + .runOn(Schedulers.boundedElastic()) + .subscribe(this::handle); + } + + @Override + public synchronized CompletableFuture asyncLoad(K key, Executor executor) { + var entry = Map.entry(key, new CompletableFuture()); + sink.tryEmitNext(entry).orThrow(); + return entry.getValue(); + } + + private void handle(Map> requests) { + try { + var results = mappingFunction.apply(requests.keySet()); + requests.forEach((key, result) -> result.complete(results.get(key))); + } catch (Throwable t) { + requests.forEach((key, result) -> result.completeExceptionally(t)); + } + } + + public static final class Builder { + private Function, Map> mappingFunction; + private Duration maxTime; + private int parallelism; + private int maxSize; + + /** The maximum collected size before performing a bulk request. */ + @CanIgnoreReturnValue + public Builder maxSize(int maxSize) { + this.maxSize = requireNonNull(maxSize); + return this; + } + + /** The maximum duration to wait before performing a bulk request. */ + @CanIgnoreReturnValue + public Builder maxTime(Duration maxTime) { + this.maxTime = requireNonNull(maxTime); + return this; + } + + /** The number of parallel bulk loads that can be performed. */ + @CanIgnoreReturnValue + public Builder parallelism(int parallelism) { + this.parallelism = requireNonNull(parallelism); + return this; + } + + /** The function to compute the values. */ + @CanIgnoreReturnValue + public Builder mappingFunction(Function, Map> mappingFunction) { + this.mappingFunction = requireNonNull(mappingFunction); + return this; + } + + /** Returns a loader that batches the individual lookups to the system of record. */ + public CoalescingBulkLoader build() { + requireNonNull(mappingFunction); + return new CoalescingBulkLoader<>(this); + } + } +} diff --git a/examples/coalescing-bulkloader-reactor/src/test/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkLoaderTest.java b/examples/coalescing-bulkloader-reactor/src/test/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkLoaderTest.java new file mode 100644 index 0000000000..622377403b --- /dev/null +++ b/examples/coalescing-bulkloader-reactor/src/test/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkLoaderTest.java @@ -0,0 +1,93 @@ +/* + * Copyright 2019 Guus C. Bloemsma. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.github.benmanes.caffeine.examples.coalescing.bulkloader; + +import static com.google.common.truth.Truth.assertThat; +import static java.util.stream.Collectors.toMap; + +import java.time.Duration; +import java.util.HashMap; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.Test; + +import com.github.benmanes.caffeine.cache.Caffeine; +import com.google.common.collect.Range; + +/** + * @author guus@bloemsma.net (Guus C. Bloemsma) + * @author ben.manes@gmail.com (Ben Manes) + */ +public final class CoalescingBulkLoaderTest { + + @Test + public void maxTime() { + var maxTime = Duration.ofMillis(50); + var end = new AtomicReference(); + var loader = new CoalescingBulkLoader.Builder() + .mappingFunction(keys -> { + end.set(System.nanoTime()); + return keys.stream().collect(toMap(key -> key, key -> -key)); + }) + .maxTime(maxTime) + .parallelism(3) + .maxSize(5) + .build(); + var cache = Caffeine.newBuilder().buildAsync(loader); + + var start = System.nanoTime(); + assertThat(cache.get(1).join()).isEqualTo(-1); + + long delay = TimeUnit.NANOSECONDS.toMillis(end.get() - start); + assertThat(delay).isIn(Range.closed(maxTime.toMillis(), (long) (1.5 * maxTime.toMillis()))); + } + + @Test + public void maxSize() { + int maxSize = 5; + int requests = 5 * maxSize; + var maxTime = Duration.ofSeconds(1); + var end = new AtomicReference(); + var batchSizes = new ConcurrentLinkedQueue(); + var loader = new CoalescingBulkLoader.Builder() + .mappingFunction(keys -> { + end.set(System.nanoTime()); + batchSizes.add(keys.size()); + return keys.stream().collect(toMap(key -> key, key -> -key)); + }) + .maxTime(maxTime) + .maxSize(maxSize) + .parallelism(3) + .build(); + var cache = Caffeine.newBuilder().buildAsync(loader); + + var start = System.nanoTime(); + var results = new HashMap>(); + for (int i = 0; i < requests; i++) { + results.put(i, cache.get(i)); + } + + results.forEach((key, future) -> assertThat(future.join()).isEqualTo(-key)); + batchSizes.forEach(batchSize-> assertThat(batchSize).isAtMost(maxSize)); + assertThat(batchSizes).hasSize(requests / maxSize); + + long delay = TimeUnit.NANOSECONDS.toMillis(end.get() - start); + assertThat(delay).isLessThan(maxTime.toMillis()); + } +} diff --git a/examples/coalescing-bulkloader/.mvn/wrapper/maven-wrapper.jar b/examples/coalescing-bulkloader/.mvn/wrapper/maven-wrapper.jar deleted file mode 100644 index bf82ff01c6..0000000000 Binary files a/examples/coalescing-bulkloader/.mvn/wrapper/maven-wrapper.jar and /dev/null differ diff --git a/examples/coalescing-bulkloader/.mvn/wrapper/maven-wrapper.properties b/examples/coalescing-bulkloader/.mvn/wrapper/maven-wrapper.properties deleted file mode 100755 index 5366408e61..0000000000 --- a/examples/coalescing-bulkloader/.mvn/wrapper/maven-wrapper.properties +++ /dev/null @@ -1,18 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# https://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. -distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.3/apache-maven-3.9.3-bin.zip -wrapperUrl=https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.2.0/maven-wrapper-3.2.0.jar diff --git a/examples/coalescing-bulkloader/mvnw b/examples/coalescing-bulkloader/mvnw deleted file mode 100755 index b7f064624f..0000000000 --- a/examples/coalescing-bulkloader/mvnw +++ /dev/null @@ -1,287 +0,0 @@ -#!/bin/sh -# ---------------------------------------------------------------------------- -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. -# ---------------------------------------------------------------------------- - -# ---------------------------------------------------------------------------- -# Apache Maven Wrapper startup batch script, version 3.1.1 -# -# Required ENV vars: -# ------------------ -# JAVA_HOME - location of a JDK home dir -# -# Optional ENV vars -# ----------------- -# MAVEN_OPTS - parameters passed to the Java VM when running Maven -# e.g. to debug Maven itself, use -# set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000 -# MAVEN_SKIP_RC - flag to disable loading of mavenrc files -# ---------------------------------------------------------------------------- - -if [ -z "$MAVEN_SKIP_RC" ] ; then - - if [ -f /usr/local/etc/mavenrc ] ; then - . /usr/local/etc/mavenrc - fi - - if [ -f /etc/mavenrc ] ; then - . /etc/mavenrc - fi - - if [ -f "$HOME/.mavenrc" ] ; then - . "$HOME/.mavenrc" - fi - -fi - -# OS specific support. $var _must_ be set to either true or false. -cygwin=false; -darwin=false; -mingw=false -case "`uname`" in - CYGWIN*) cygwin=true ;; - MINGW*) mingw=true;; - Darwin*) darwin=true - # Use /usr/libexec/java_home if available, otherwise fall back to /Library/Java/Home - # See https://developer.apple.com/library/mac/qa/qa1170/_index.html - if [ -z "$JAVA_HOME" ]; then - if [ -x "/usr/libexec/java_home" ]; then - JAVA_HOME="`/usr/libexec/java_home`"; export JAVA_HOME - else - JAVA_HOME="/Library/Java/Home"; export JAVA_HOME - fi - fi - ;; -esac - -if [ -z "$JAVA_HOME" ] ; then - if [ -r /etc/gentoo-release ] ; then - JAVA_HOME=`java-config --jre-home` - fi -fi - -# For Cygwin, ensure paths are in UNIX format before anything is touched -if $cygwin ; then - [ -n "$JAVA_HOME" ] && - JAVA_HOME=`cygpath --unix "$JAVA_HOME"` - [ -n "$CLASSPATH" ] && - CLASSPATH=`cygpath --path --unix "$CLASSPATH"` -fi - -# For Mingw, ensure paths are in UNIX format before anything is touched -if $mingw ; then - [ -n "$JAVA_HOME" ] && - JAVA_HOME="`(cd "$JAVA_HOME"; pwd)`" -fi - -if [ -z "$JAVA_HOME" ]; then - javaExecutable="`which javac`" - if [ -n "$javaExecutable" ] && ! [ "`expr \"$javaExecutable\" : '\([^ ]*\)'`" = "no" ]; then - # readlink(1) is not available as standard on Solaris 10. - readLink=`which readlink` - if [ ! `expr "$readLink" : '\([^ ]*\)'` = "no" ]; then - if $darwin ; then - javaHome="`dirname \"$javaExecutable\"`" - javaExecutable="`cd \"$javaHome\" && pwd -P`/javac" - else - javaExecutable="`readlink -f \"$javaExecutable\"`" - fi - javaHome="`dirname \"$javaExecutable\"`" - javaHome=`expr "$javaHome" : '\(.*\)/bin'` - JAVA_HOME="$javaHome" - export JAVA_HOME - fi - fi -fi - -if [ -z "$JAVACMD" ] ; then - if [ -n "$JAVA_HOME" ] ; then - if [ -x "$JAVA_HOME/jre/sh/java" ] ; then - # IBM's JDK on AIX uses strange locations for the executables - JAVACMD="$JAVA_HOME/jre/sh/java" - else - JAVACMD="$JAVA_HOME/bin/java" - fi - else - JAVACMD="`\\unset -f command; \\command -v java`" - fi -fi - -if [ ! -x "$JAVACMD" ] ; then - echo "Error: JAVA_HOME is not defined correctly." >&2 - echo " We cannot execute $JAVACMD" >&2 - exit 1 -fi - -if [ -z "$JAVA_HOME" ] ; then - echo "Warning: JAVA_HOME environment variable is not set." -fi - -# traverses directory structure from process work directory to filesystem root -# first directory with .mvn subdirectory is considered project base directory -find_maven_basedir() { - if [ -z "$1" ] - then - echo "Path not specified to find_maven_basedir" - return 1 - fi - - basedir="$1" - wdir="$1" - while [ "$wdir" != '/' ] ; do - if [ -d "$wdir"/.mvn ] ; then - basedir=$wdir - break - fi - # workaround for JBEAP-8937 (on Solaris 10/Sparc) - if [ -d "${wdir}" ]; then - wdir=`cd "$wdir/.."; pwd` - fi - # end of workaround - done - printf '%s' "$(cd "$basedir"; pwd)" -} - -# concatenates all lines of a file -concat_lines() { - if [ -f "$1" ]; then - echo "$(tr -s '\n' ' ' < "$1")" - fi -} - -BASE_DIR=$(find_maven_basedir "$(dirname $0)") -if [ -z "$BASE_DIR" ]; then - exit 1; -fi - -MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-"$BASE_DIR"}; export MAVEN_PROJECTBASEDIR -if [ "$MVNW_VERBOSE" = true ]; then - echo $MAVEN_PROJECTBASEDIR -fi - -########################################################################################## -# Extension to allow automatically downloading the maven-wrapper.jar from Maven-central -# This allows using the maven wrapper in projects that prohibit checking in binary data. -########################################################################################## -if [ -r "$BASE_DIR/.mvn/wrapper/maven-wrapper.jar" ]; then - if [ "$MVNW_VERBOSE" = true ]; then - echo "Found .mvn/wrapper/maven-wrapper.jar" - fi -else - if [ "$MVNW_VERBOSE" = true ]; then - echo "Couldn't find .mvn/wrapper/maven-wrapper.jar, downloading it ..." - fi - if [ -n "$MVNW_REPOURL" ]; then - wrapperUrl="$MVNW_REPOURL/org/apache/maven/wrapper/maven-wrapper/3.1.1/maven-wrapper-3.1.1.jar" - else - wrapperUrl="https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.1.1/maven-wrapper-3.1.1.jar" - fi - while IFS="=" read key value; do - case "$key" in (wrapperUrl) wrapperUrl="$value"; break ;; - esac - done < "$BASE_DIR/.mvn/wrapper/maven-wrapper.properties" - if [ "$MVNW_VERBOSE" = true ]; then - echo "Downloading from: $wrapperUrl" - fi - wrapperJarPath="$BASE_DIR/.mvn/wrapper/maven-wrapper.jar" - if $cygwin; then - wrapperJarPath=`cygpath --path --windows "$wrapperJarPath"` - fi - - if command -v wget > /dev/null; then - QUIET="--quiet" - if [ "$MVNW_VERBOSE" = true ]; then - echo "Found wget ... using wget" - QUIET="" - fi - if [ -z "$MVNW_USERNAME" ] || [ -z "$MVNW_PASSWORD" ]; then - wget $QUIET "$wrapperUrl" -O "$wrapperJarPath" - else - wget $QUIET --http-user="$MVNW_USERNAME" --http-password="$MVNW_PASSWORD" "$wrapperUrl" -O "$wrapperJarPath" - fi - [ $? -eq 0 ] || rm -f "$wrapperJarPath" - elif command -v curl > /dev/null; then - QUIET="--silent" - if [ "$MVNW_VERBOSE" = true ]; then - echo "Found curl ... using curl" - QUIET="" - fi - if [ -z "$MVNW_USERNAME" ] || [ -z "$MVNW_PASSWORD" ]; then - curl $QUIET -o "$wrapperJarPath" "$wrapperUrl" -f -L - else - curl $QUIET --user "$MVNW_USERNAME:$MVNW_PASSWORD" -o "$wrapperJarPath" "$wrapperUrl" -f -L - fi - [ $? -eq 0 ] || rm -f "$wrapperJarPath" - else - if [ "$MVNW_VERBOSE" = true ]; then - echo "Falling back to using Java to download" - fi - javaSource="$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.java" - javaClass="$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.class" - # For Cygwin, switch paths to Windows format before running javac - if $cygwin; then - javaSource=`cygpath --path --windows "$javaSource"` - javaClass=`cygpath --path --windows "$javaClass"` - fi - if [ -e "$javaSource" ]; then - if [ ! -e "$javaClass" ]; then - if [ "$MVNW_VERBOSE" = true ]; then - echo " - Compiling MavenWrapperDownloader.java ..." - fi - # Compiling the Java class - ("$JAVA_HOME/bin/javac" "$javaSource") - fi - if [ -e "$javaClass" ]; then - # Running the downloader - if [ "$MVNW_VERBOSE" = true ]; then - echo " - Running MavenWrapperDownloader.java ..." - fi - ("$JAVA_HOME/bin/java" -cp .mvn/wrapper MavenWrapperDownloader "$MAVEN_PROJECTBASEDIR") - fi - fi - fi -fi -########################################################################################## -# End of extension -########################################################################################## - -MAVEN_OPTS="$(concat_lines "$MAVEN_PROJECTBASEDIR/.mvn/jvm.config") $MAVEN_OPTS" - -# For Cygwin, switch paths to Windows format before running java -if $cygwin; then - [ -n "$JAVA_HOME" ] && - JAVA_HOME=`cygpath --path --windows "$JAVA_HOME"` - [ -n "$CLASSPATH" ] && - CLASSPATH=`cygpath --path --windows "$CLASSPATH"` - [ -n "$MAVEN_PROJECTBASEDIR" ] && - MAVEN_PROJECTBASEDIR=`cygpath --path --windows "$MAVEN_PROJECTBASEDIR"` -fi - -# Provide a "standardized" way to retrieve the CLI args that will -# work with both Windows and non-Windows executions. -MAVEN_CMD_LINE_ARGS="$MAVEN_CONFIG $@" -export MAVEN_CMD_LINE_ARGS - -WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain - -exec "$JAVACMD" \ - $MAVEN_OPTS \ - $MAVEN_DEBUG_OPTS \ - -classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \ - "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \ - ${WRAPPER_LAUNCHER} $MAVEN_CONFIG "$@" diff --git a/examples/coalescing-bulkloader/mvnw.cmd b/examples/coalescing-bulkloader/mvnw.cmd deleted file mode 100644 index 474c9d6b74..0000000000 --- a/examples/coalescing-bulkloader/mvnw.cmd +++ /dev/null @@ -1,187 +0,0 @@ -@REM ---------------------------------------------------------------------------- -@REM Licensed to the Apache Software Foundation (ASF) under one -@REM or more contributor license agreements. See the NOTICE file -@REM distributed with this work for additional information -@REM regarding copyright ownership. The ASF licenses this file -@REM to you under the Apache License, Version 2.0 (the -@REM "License"); you may not use this file except in compliance -@REM with the License. You may obtain a copy of the License at -@REM -@REM http://www.apache.org/licenses/LICENSE-2.0 -@REM -@REM Unless required by applicable law or agreed to in writing, -@REM software distributed under the License is distributed on an -@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -@REM KIND, either express or implied. See the License for the -@REM specific language governing permissions and limitations -@REM under the License. -@REM ---------------------------------------------------------------------------- - -@REM ---------------------------------------------------------------------------- -@REM Apache Maven Wrapper startup batch script, version 3.1.1 -@REM -@REM Required ENV vars: -@REM JAVA_HOME - location of a JDK home dir -@REM -@REM Optional ENV vars -@REM MAVEN_BATCH_ECHO - set to 'on' to enable the echoing of the batch commands -@REM MAVEN_BATCH_PAUSE - set to 'on' to wait for a keystroke before ending -@REM MAVEN_OPTS - parameters passed to the Java VM when running Maven -@REM e.g. to debug Maven itself, use -@REM set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000 -@REM MAVEN_SKIP_RC - flag to disable loading of mavenrc files -@REM ---------------------------------------------------------------------------- - -@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on' -@echo off -@REM set title of command window -title %0 -@REM enable echoing by setting MAVEN_BATCH_ECHO to 'on' -@if "%MAVEN_BATCH_ECHO%" == "on" echo %MAVEN_BATCH_ECHO% - -@REM set %HOME% to equivalent of $HOME -if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%") - -@REM Execute a user defined script before this one -if not "%MAVEN_SKIP_RC%" == "" goto skipRcPre -@REM check for pre script, once with legacy .bat ending and once with .cmd ending -if exist "%USERPROFILE%\mavenrc_pre.bat" call "%USERPROFILE%\mavenrc_pre.bat" %* -if exist "%USERPROFILE%\mavenrc_pre.cmd" call "%USERPROFILE%\mavenrc_pre.cmd" %* -:skipRcPre - -@setlocal - -set ERROR_CODE=0 - -@REM To isolate internal variables from possible post scripts, we use another setlocal -@setlocal - -@REM ==== START VALIDATION ==== -if not "%JAVA_HOME%" == "" goto OkJHome - -echo. -echo Error: JAVA_HOME not found in your environment. >&2 -echo Please set the JAVA_HOME variable in your environment to match the >&2 -echo location of your Java installation. >&2 -echo. -goto error - -:OkJHome -if exist "%JAVA_HOME%\bin\java.exe" goto init - -echo. -echo Error: JAVA_HOME is set to an invalid directory. >&2 -echo JAVA_HOME = "%JAVA_HOME%" >&2 -echo Please set the JAVA_HOME variable in your environment to match the >&2 -echo location of your Java installation. >&2 -echo. -goto error - -@REM ==== END VALIDATION ==== - -:init - -@REM Find the project base dir, i.e. the directory that contains the folder ".mvn". -@REM Fallback to current working directory if not found. - -set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR% -IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir - -set EXEC_DIR=%CD% -set WDIR=%EXEC_DIR% -:findBaseDir -IF EXIST "%WDIR%"\.mvn goto baseDirFound -cd .. -IF "%WDIR%"=="%CD%" goto baseDirNotFound -set WDIR=%CD% -goto findBaseDir - -:baseDirFound -set MAVEN_PROJECTBASEDIR=%WDIR% -cd "%EXEC_DIR%" -goto endDetectBaseDir - -:baseDirNotFound -set MAVEN_PROJECTBASEDIR=%EXEC_DIR% -cd "%EXEC_DIR%" - -:endDetectBaseDir - -IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig - -@setlocal EnableExtensions EnableDelayedExpansion -for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do set JVM_CONFIG_MAVEN_PROPS=!JVM_CONFIG_MAVEN_PROPS! %%a -@endlocal & set JVM_CONFIG_MAVEN_PROPS=%JVM_CONFIG_MAVEN_PROPS% - -:endReadAdditionalConfig - -SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe" -set WRAPPER_JAR="%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.jar" -set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain - -set WRAPPER_URL="https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.1.1/maven-wrapper-3.1.1.jar" - -FOR /F "usebackq tokens=1,2 delims==" %%A IN ("%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.properties") DO ( - IF "%%A"=="wrapperUrl" SET WRAPPER_URL=%%B -) - -@REM Extension to allow automatically downloading the maven-wrapper.jar from Maven-central -@REM This allows using the maven wrapper in projects that prohibit checking in binary data. -if exist %WRAPPER_JAR% ( - if "%MVNW_VERBOSE%" == "true" ( - echo Found %WRAPPER_JAR% - ) -) else ( - if not "%MVNW_REPOURL%" == "" ( - SET WRAPPER_URL="%MVNW_REPOURL%/org/apache/maven/wrapper/maven-wrapper/3.1.1/maven-wrapper-3.1.1.jar" - ) - if "%MVNW_VERBOSE%" == "true" ( - echo Couldn't find %WRAPPER_JAR%, downloading it ... - echo Downloading from: %WRAPPER_URL% - ) - - powershell -Command "&{"^ - "$webclient = new-object System.Net.WebClient;"^ - "if (-not ([string]::IsNullOrEmpty('%MVNW_USERNAME%') -and [string]::IsNullOrEmpty('%MVNW_PASSWORD%'))) {"^ - "$webclient.Credentials = new-object System.Net.NetworkCredential('%MVNW_USERNAME%', '%MVNW_PASSWORD%');"^ - "}"^ - "[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12; $webclient.DownloadFile('%WRAPPER_URL%', '%WRAPPER_JAR%')"^ - "}" - if "%MVNW_VERBOSE%" == "true" ( - echo Finished downloading %WRAPPER_JAR% - ) -) -@REM End of extension - -@REM Provide a "standardized" way to retrieve the CLI args that will -@REM work with both Windows and non-Windows executions. -set MAVEN_CMD_LINE_ARGS=%* - -%MAVEN_JAVA_EXE% ^ - %JVM_CONFIG_MAVEN_PROPS% ^ - %MAVEN_OPTS% ^ - %MAVEN_DEBUG_OPTS% ^ - -classpath %WRAPPER_JAR% ^ - "-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" ^ - %WRAPPER_LAUNCHER% %MAVEN_CONFIG% %* -if ERRORLEVEL 1 goto error -goto end - -:error -set ERROR_CODE=1 - -:end -@endlocal & set ERROR_CODE=%ERROR_CODE% - -if not "%MAVEN_SKIP_RC%"=="" goto skipRcPost -@REM check for post script, once with legacy .bat ending and once with .cmd ending -if exist "%USERPROFILE%\mavenrc_post.bat" call "%USERPROFILE%\mavenrc_post.bat" -if exist "%USERPROFILE%\mavenrc_post.cmd" call "%USERPROFILE%\mavenrc_post.cmd" -:skipRcPost - -@REM pause the script if MAVEN_BATCH_PAUSE is set to 'on' -if "%MAVEN_BATCH_PAUSE%"=="on" pause - -if "%MAVEN_TERMINATE_CMD%"=="on" exit %ERROR_CODE% - -cmd /C exit /B %ERROR_CODE% diff --git a/examples/coalescing-bulkloader/pom.xml b/examples/coalescing-bulkloader/pom.xml deleted file mode 100644 index 00b5db0167..0000000000 --- a/examples/coalescing-bulkloader/pom.xml +++ /dev/null @@ -1,107 +0,0 @@ - - - 4.0.0 - - org.github.benmanes.caffeine.examples - coalescing-bulkloader - 1.0-SNAPSHOT - - - 11 - 4.13.2 - 3.1.7 - 4.2.0 - 3.3.0 - 3.12.1 - 3.3.1 - 3.1.1 - 3.1.1 - 3.3.0 - 3.1.2 - 3.11.0 - 3.3.1 - UTF-8 - UTF-8 - - - - - com.github.ben-manes.caffeine - caffeine - ${caffeine.version} - - - junit - junit - ${junit.version} - - - org.awaitility - awaitility - ${awaitility.version} - - - - - - - maven-compiler-plugin - ${maven-compiler-plugin.version} - - ${java.version} - ${java.version} - - - - org.apache.maven.plugins - maven-enforcer-plugin - ${maven-enforcer-plugin.version} - - - enforce-maven - - enforce - - - - - 3.8.7 - - - - - - - - maven-clean-plugin - ${maven-clean-plugin.version} - - - maven-deploy-plugin - ${maven-deploy-plugin.version} - - - maven-install-plugin - ${maven-install-plugin.version} - - - maven-jar-plugin - ${maven-jar-plugin.version} - - - maven-site-plugin - ${maven-site-plugin.version} - - - maven-surefire-plugin - ${maven-surefire-plugin.version} - - - maven-resources-plugin - ${maven-resources-plugin.version} - - - - diff --git a/examples/coalescing-bulkloader/src/main/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkloader.java b/examples/coalescing-bulkloader/src/main/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkloader.java deleted file mode 100644 index 3c287519bc..0000000000 --- a/examples/coalescing-bulkloader/src/main/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkloader.java +++ /dev/null @@ -1,225 +0,0 @@ -/* - * Copyright 2019 Guus C. Bloemsma. All Rights Reserved. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package com.github.benmanes.caffeine.examples.coalescing.bulkloader; - -import com.github.benmanes.caffeine.cache.AsyncCacheLoader; - -import java.util.ArrayList; -import java.util.Collection; -import java.util.Iterator; -import java.util.List; -import java.util.Map; -import java.util.NoSuchElementException; -import java.util.Queue; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.Executor; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Consumer; -import java.util.function.Function; -import java.util.stream.Stream; - -import static java.util.concurrent.TimeUnit.MILLISECONDS; -import static java.util.stream.Collectors.toMap; - -/** - * An implementation of {@link AsyncCacheLoader} that delays fetching a bit until "enough" keys are - * collected to do a bulk call. The assumption is that doing a bulk call is so much more efficient - * that it is worth the wait. - * - * @param the type of the key in the cache - * @param the type of the value in the cache - * @author complain to: guus@bloemsma.net - */ -public class CoalescingBulkloader implements AsyncCacheLoader { - private final Consumer>> bulkLoader; - private final int maxLoadSize; // maximum number of keys to load in one call - private final long maxDelay; // maximum time between request of a value and loading it - private final Queue> waitingKeys = new ConcurrentLinkedQueue<>(); - private final ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(); - private final AtomicReference> schedule = new AtomicReference<>(); - // Queue.size() is expensive, so here we keep track of the queue size separately - private final AtomicInteger size = new AtomicInteger(0); - - private static class WaitingKey { - private final Key key; - private final CompletableFuture future = new CompletableFuture<>(); - private final long waitingSince = System.currentTimeMillis(); - - private WaitingKey(Key key) { - this.key = key; - } - } - - /** - * Wraps a bulk loader that returns values in the same order as the keys. - * - * @param maxLoadSize Maximum number of keys per bulk load - * @param maxDelay Maximum time to wait before bulk load is executed - * @param load Loader that takes keys and returns a future with the values in the same order as - * the keys. Extra values are ignored. Missing values lead to a {@link - * java.util.NoSuchElementException} for the corresponding future. - */ - public static CoalescingBulkloader byOrder( - int maxLoadSize, - long maxDelay, - final Function, CompletableFuture>> load) { - return new CoalescingBulkloader<>( - maxLoadSize, - maxDelay, - toLoad -> { - final Stream keys = toLoad.stream().map(wk -> wk.key); - load.apply(keys) - .thenAccept( - values -> { - final Iterator iv = values.iterator(); - for (CoalescingBulkloader.WaitingKey waitingKey : toLoad) { - if (iv.hasNext()) waitingKey.future.complete(iv.next()); - else - waitingKey.future.completeExceptionally( - new NoSuchElementException("No value for key " + waitingKey.key)); - } - }); - }); - } - - /** - * Wraps a bulk loader that returns values in a map accessed by key. - * - * @param maxLoadSize Maximum number of keys per bulk load - * @param maxDelay Maximum time to wait before bulk load is executed - * @param load Loader that takes keys and returns a future with a map with keys and values. Extra - * values are ignored. Missing values lead to a {@link java.util.NoSuchElementException} for - * the corresponding future. - */ - public static CoalescingBulkloader byMap( - int maxLoadSize, - long maxDelay, - final Function, CompletableFuture>> load) { - return new CoalescingBulkloader<>( - maxLoadSize, - maxDelay, - toLoad -> { - final Stream keys = toLoad.stream().map(wk -> wk.key); - load.apply(keys) - .thenAccept( - values -> { - for (CoalescingBulkloader.WaitingKey waitingKey : toLoad) { - if (values.containsKey(waitingKey.key)) - waitingKey.future.complete(values.get(waitingKey.key)); - else - waitingKey.future.completeExceptionally( - new NoSuchElementException("No value for key " + waitingKey.key)); - } - }); - }); - } - - /** - * Wraps a bulk loader that returns intermediate values from which keys and values can be - * extracted. - * - * @param Some internal type from which keys and values can be extracted. - * @param maxLoadSize Maximum number of keys per bulk load - * @param maxDelay Maximum time to wait before bulk load is executed - * @param keyExtractor How to extract key from intermediate value - * @param valueExtractor How to extract value from intermediate value - * @param load Loader that takes keys and returns a future with a map with keys and values. Extra - * values are ignored. Missing values lead to a {@link java.util.NoSuchElementException} for - * the corresponding future. - */ - public static CoalescingBulkloader byExtraction( - int maxLoadSize, - long maxDelay, - final Function keyExtractor, - final Function valueExtractor, - final Function, CompletableFuture>> load) { - return byMap( - maxLoadSize, - maxDelay, - keys -> - load.apply(keys) - .thenApply( - intermediates -> intermediates.collect(toMap(keyExtractor, valueExtractor)))); - } - - private CoalescingBulkloader( - int maxLoadSize, long maxDelay, Consumer>> bulkLoader) { - this.bulkLoader = bulkLoader; - assert maxLoadSize > 0; - assert maxDelay > 0; - this.maxLoadSize = maxLoadSize; - this.maxDelay = maxDelay; - } - - @Override - public CompletableFuture asyncLoad(Key key, Executor executor) { - final WaitingKey waitingKey = new WaitingKey<>(key); - waitingKeys.add(waitingKey); - - if (size.incrementAndGet() >= maxLoadSize) { - doLoad(); - } else { - ScheduledFuture existingSchedule = schedule.get(); - if (existingSchedule == null || existingSchedule.isDone()) { - startWaiting(); - } - } - - return waitingKey.future; - } - - private void startWaiting() { - ScheduledFuture oldSchedule = schedule.getAndSet( - timer.schedule(this::doLoad, maxDelay, MILLISECONDS)); - if (oldSchedule != null) { - oldSchedule.cancel(false); - } - } - - private synchronized void doLoad() { - do { - List> toLoad = new ArrayList<>(Math.min(size.get(), maxLoadSize)); - int counter = maxLoadSize; - while (counter > 0) { - final WaitingKey waitingKey = waitingKeys.poll(); - if (waitingKey == null) break; - else { - toLoad.add(waitingKey); - counter--; - } - } - - final int taken = maxLoadSize - counter; - if (taken > 0) { - size.updateAndGet(oldSize -> oldSize - taken); - bulkLoader.accept(toLoad); - } - - } while (size.get() >= maxLoadSize); - final WaitingKey nextWaitingKey = waitingKeys.peek(); - if (nextWaitingKey != null) { - schedule.set(timer.schedule( - this::doLoad, - nextWaitingKey.waitingSince + maxDelay - System.currentTimeMillis(), - MILLISECONDS)); - } - } -} diff --git a/examples/coalescing-bulkloader/src/main/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/package-info.java b/examples/coalescing-bulkloader/src/main/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/package-info.java deleted file mode 100644 index 756be6d1e2..0000000000 --- a/examples/coalescing-bulkloader/src/main/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/package-info.java +++ /dev/null @@ -1,4 +0,0 @@ -@CheckReturnValue -package com.github.benmanes.caffeine.examples.coalescing.bulkloader; - -import com.google.errorprone.annotations.CheckReturnValue; diff --git a/examples/coalescing-bulkloader/src/test/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkloaderTest.java b/examples/coalescing-bulkloader/src/test/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkloaderTest.java deleted file mode 100644 index a7dcf2e8dd..0000000000 --- a/examples/coalescing-bulkloader/src/test/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/CoalescingBulkloaderTest.java +++ /dev/null @@ -1,189 +0,0 @@ -/* - * Copyright 2019 Guus C. Bloemsma. All Rights Reserved. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package com.github.benmanes.caffeine.examples.coalescing.bulkloader; - -import com.github.benmanes.caffeine.cache.AsyncLoadingCache; -import com.github.benmanes.caffeine.cache.Caffeine; -import org.awaitility.Awaitility; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TestRule; -import org.junit.runner.Description; -import org.junit.runner.RunWith; -import org.junit.runners.Parameterized; -import org.junit.runners.Parameterized.Parameters; -import org.junit.runners.model.Statement; - -import java.util.Arrays; -import java.util.List; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Function; -import java.util.stream.Stream; - -import static com.github.benmanes.caffeine.examples.coalescing.bulkloader.CoalescingBulkloader.byExtraction; -import static com.github.benmanes.caffeine.examples.coalescing.bulkloader.CoalescingBulkloader.byMap; -import static com.github.benmanes.caffeine.examples.coalescing.bulkloader.CoalescingBulkloader.byOrder; -import static java.util.concurrent.TimeUnit.MILLISECONDS; -import static java.util.function.Function.identity; -import static java.util.stream.Collectors.toMap; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.sameInstance; -import static org.hamcrest.core.Is.is; -import static org.junit.Assert.assertFalse; - -@RunWith(Parameterized.class) -public final class CoalescingBulkloaderTest { - - private final Function< - Function, Stream>, CoalescingBulkloader> - cbl; - - public CoalescingBulkloaderTest( - Function, Stream>, CoalescingBulkloader> - cbl) { - this.cbl = cbl; - } - - private static final int maxLoadSize = 10; - private static final int maxDelay = 100; - private static final int delta = 20; - private static final int actualLoadTime = 50; - - @Parameters - public static List< - Function< - Function, Stream>, CoalescingBulkloader>> - loaderTypes() { - return Arrays.asList( - fun -> - byOrder( - maxLoadSize, - maxDelay, - ints -> CompletableFuture.supplyAsync(() -> fun.apply(ints))), - fun -> - byMap( - maxLoadSize, - maxDelay, - ints -> - CompletableFuture.supplyAsync( - () -> fun.apply(ints).collect(toMap(identity(), identity())))), - fun -> - byExtraction( - maxLoadSize, - maxDelay, - identity(), - identity(), - ints -> CompletableFuture.supplyAsync(() -> fun.apply(ints)))); - } - - private AsyncLoadingCache createCache(AtomicInteger loaderCalled) { - return Caffeine.newBuilder() - .buildAsync( - cbl.apply( - ints -> { - loaderCalled.incrementAndGet(); - try { - Thread.sleep(actualLoadTime); - } catch (InterruptedException e) { - e.printStackTrace(); - } - return ints; - })); - } - - @Test - public void maxDelayIsNotMissedTooMuch() throws InterruptedException { - AtomicInteger loaderCalled = new AtomicInteger(0); - final AsyncLoadingCache cache = createCache(loaderCalled); - - // a cache get won't take too long - final CompletableFuture result = cache.get(1); - Awaitility.await() - .pollThread(Thread::new) - .pollInterval(1, MILLISECONDS) - .between(maxDelay - delta, MILLISECONDS, maxDelay + delta, MILLISECONDS) - .untilAtomic(loaderCalled, is(1)); - assertFalse("delay in load", result.isDone()); - Thread.sleep(actualLoadTime); - assertThat(result.getNow(0), is(1)); - } - - @Test - public void whenEnoughKeysAreRequestedTheLoadWillHappenImmediately() throws InterruptedException { - AtomicInteger loaderCalled = new AtomicInteger(0); - final AsyncLoadingCache cache = createCache(loaderCalled); - - CompletableFuture[] results = new CompletableFuture[maxLoadSize]; - for (int i = 0; i < maxLoadSize - 1; i++) results[i] = cache.get(i); - Thread.sleep(delta); - // requesting 9 keys does not trigger a load - assertThat(loaderCalled.get(), is(0)); - - for (int i = 0; i < maxLoadSize - 1; i++) { - final CompletableFuture result = cache.get(i); - assertThat(result, sameInstance(results[i])); - assertFalse("no load therefore unknown result", result.isDone()); - } - Thread.sleep(delta); - // requesting the same 9 keys still doesn't trigger a load - assertThat(loaderCalled.get(), is(0)); - - // requesting one more key will trigger immediately - results[maxLoadSize - 1] = cache.get(maxLoadSize - 1); - Awaitility.await() - .pollInterval(1, MILLISECONDS) - .atMost(delta, MILLISECONDS) - .untilAtomic(loaderCalled, is(1)); - - // values are not immediately available because of the sleep in the loader - for (int i = 0; i < maxLoadSize; i++) { - assertThat(results[i].getNow(-1), is(-1)); - } - Thread.sleep(actualLoadTime + delta); - // slept enough - for (int i = 0; i < maxLoadSize; i++) { - assertThat(results[i].getNow(-1), is(i)); - } - } - - @Rule - // Because the jvm may have to warm up or whatever other influences, timing may be off, causing - // these tests to fail. - // So retry a couple of times. - public TestRule retry = - (final Statement base, final Description description) -> - new Statement() { - - @Override - public void evaluate() throws Throwable { - try_(3); - } - - void try_(int tries) throws Throwable { - try { - base.evaluate(); - } catch (Throwable throwable) { - System.err.println( - description.getDisplayName() + " failed, " + (tries - 1) + " attempts left."); - if (tries > 1) try_(tries - 1); - else { - throw throwable; - } - } - } - }; -} diff --git a/examples/coalescing-bulkloader/src/test/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/package-info.java b/examples/coalescing-bulkloader/src/test/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/package-info.java deleted file mode 100644 index 756be6d1e2..0000000000 --- a/examples/coalescing-bulkloader/src/test/java/com/github/benmanes/caffeine/examples/coalescing/bulkloader/package-info.java +++ /dev/null @@ -1,4 +0,0 @@ -@CheckReturnValue -package com.github.benmanes.caffeine.examples.coalescing.bulkloader; - -import com.google.errorprone.annotations.CheckReturnValue; diff --git a/examples/resilience-failsafe/src/test/java/com/github/benmanes/caffeine/examples/resilience/package-info.java b/examples/resilience-failsafe/src/test/java/com/github/benmanes/caffeine/examples/resilience/package-info.java deleted file mode 100644 index 4ab0a9c804..0000000000 --- a/examples/resilience-failsafe/src/test/java/com/github/benmanes/caffeine/examples/resilience/package-info.java +++ /dev/null @@ -1,4 +0,0 @@ -@CheckReturnValue -package com.github.benmanes.caffeine.examples.resilience; - -import com.google.errorprone.annotations.CheckReturnValue; diff --git a/examples/write-behind-rxjava/README.md b/examples/write-behind-rxjava/README.md index a6003c7b02..5eae3faf56 100644 --- a/examples/write-behind-rxjava/README.md +++ b/examples/write-behind-rxjava/README.md @@ -9,11 +9,11 @@ operation. ```java var subject = PublishSubject.>create().toSerialized(); -subject.buffer(10, TimeUnit.SECONDS) +subject.buffer(1, TimeUnit.SECONDS) .map(entries -> entries.stream().collect( - toMap(Entry::getKey, Entry::getValue, (v1, v2) -> /* latest */ v2))) + toMap(Entry::getKey, Entry::getValue, (v1, v2) -> v2))) .subscribeOn(Schedulers.io()) - .subscribe(entries -> System.out.println(entries)); + .subscribe(System.out::println); Cache cache = Caffeine.newBuilder().build(); cache.asMap().compute(key, (k, v) -> { diff --git a/examples/write-behind-rxjava/src/main/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/WriteBehindCacheWriter.java b/examples/write-behind-rxjava/src/main/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/WriteBehindCacheWriter.java index 49e8b07eef..f0a3e8201e 100644 --- a/examples/write-behind-rxjava/src/main/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/WriteBehindCacheWriter.java +++ b/examples/write-behind-rxjava/src/main/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/WriteBehindCacheWriter.java @@ -18,7 +18,7 @@ import static java.util.Objects.requireNonNull; import static java.util.stream.Collectors.toMap; -import java.io.Closeable; +import java.time.Duration; import java.util.Map; import java.util.Map.Entry; import java.util.concurrent.TimeUnit; @@ -28,33 +28,32 @@ import com.google.errorprone.annotations.CanIgnoreReturnValue; -import io.reactivex.rxjava3.disposables.Disposable; import io.reactivex.rxjava3.subjects.PublishSubject; import io.reactivex.rxjava3.subjects.Subject; /** - * This class allows a cache to have write-behind semantics. The passed in {@code writeAction} - * will only be called every {@code bufferTime} time with a {@linkplain} Map} containing the keys - * and the values that have been updated in the cache each time. + * This class implements write-behind semantics for caching. The provided {@code writeAction} is + * invoked periodically every {@code bufferTime} interval, receiving a {@linkplain Map} containing + * the updated keys and corresponding values in the cache since the last invocation. *

- * If a key is updated multiple times during that period then the {@code binaryOperator} has to - * decide which value should be taken. + * In scenarios where a key is updated multiple times within the same period, a coalescing function + * is responsible for determining the value to retain. *

- * An example usage of this class is keeping track of users and their activity on a web site. You - * don't want to update the database each time any of your users does something. It is better to - * batch the updates every x seconds and just write the most recent time in the database. + * A practical use case for this class is user activity tracking on a website. Instead of + * immediately updating the database for every user action, you can aggregate updates and write them + * to the database in batches at regular intervals (e.g., every x seconds), recording only the most + * recent timestamp. * * @param the type of the key in the cache * @param the type of the value in the cache * @author wim.deblauwe@gmail.com (Wim Deblauwe) */ -public final class WriteBehindCacheWriter implements BiConsumer, Closeable { - final Subject> subject; - final Disposable subscription; +public final class WriteBehindCacheWriter implements BiConsumer { + private final Subject> subject; private WriteBehindCacheWriter(Builder builder) { subject = PublishSubject.>create().toSerialized(); - subscription = subject.buffer(builder.bufferTimeNanos, TimeUnit.NANOSECONDS) + subject.buffer(builder.bufferTime.toNanos(), TimeUnit.NANOSECONDS) .map(entries -> entries.stream().collect( toMap(Entry::getKey, Entry::getValue, builder.coalescer))) .subscribe(builder.writeAction::accept); @@ -64,44 +63,40 @@ private WriteBehindCacheWriter(Builder builder) { subject.onNext(Map.entry(key, value)); } - @Override public void close() { - subscription.dispose(); - } - public static final class Builder { private Consumer> writeAction; private BinaryOperator coalescer; - private long bufferTimeNanos; + private Duration bufferTime; /** * The duration that the calls to the cache should be buffered before calling the * {@code writeAction}. */ @CanIgnoreReturnValue - public Builder bufferTime(long duration, TimeUnit unit) { - this.bufferTimeNanos = TimeUnit.NANOSECONDS.convert(duration, unit); + public Builder bufferTime(Duration duration) { + this.bufferTime = requireNonNull(duration); return this; } - /** The callback to perform the writing to the database or repository. */ + /** The callback to perform the batch write. */ @CanIgnoreReturnValue public Builder writeAction(Consumer> writeAction) { this.writeAction = requireNonNull(writeAction); return this; } - /** The action that decides which value to take in case a key was updated multiple times. */ + /** The strategy that decides which value to take in case a key was updated multiple times. */ @CanIgnoreReturnValue public Builder coalesce(BinaryOperator coalescer) { this.coalescer = requireNonNull(coalescer); return this; } - /** Returns a writer that batches changes to the system of record. */ + /** Returns a writer that batches changes to the data store. */ public WriteBehindCacheWriter build() { requireNonNull(coalescer); + requireNonNull(bufferTime); requireNonNull(writeAction); - requireNonNull(bufferTimeNanos); return new WriteBehindCacheWriter<>(this); } } diff --git a/examples/write-behind-rxjava/src/main/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/package-info.java b/examples/write-behind-rxjava/src/main/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/package-info.java deleted file mode 100644 index ed1e9c2aaf..0000000000 --- a/examples/write-behind-rxjava/src/main/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/package-info.java +++ /dev/null @@ -1,4 +0,0 @@ -@CheckReturnValue -package com.github.benmanes.caffeine.examples.writebehind.rxjava; - -import com.google.errorprone.annotations.CheckReturnValue; diff --git a/examples/write-behind-rxjava/src/test/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/WriteBehindCacheWriterTest.java b/examples/write-behind-rxjava/src/test/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/WriteBehindCacheWriterTest.java index 16cff98324..07f82aa857 100644 --- a/examples/write-behind-rxjava/src/test/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/WriteBehindCacheWriterTest.java +++ b/examples/write-behind-rxjava/src/test/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/WriteBehindCacheWriterTest.java @@ -19,8 +19,8 @@ import static org.awaitility.Awaitility.await; import static org.hamcrest.Matchers.is; +import java.time.Duration; import java.time.ZonedDateTime; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -43,23 +43,22 @@ public void singleKey() { var writerCalled = new AtomicBoolean(); // Given this cache... - try (var writer = new WriteBehindCacheWriter.Builder() - .bufferTime(1, TimeUnit.SECONDS) - .writeAction(entries -> writerCalled.set(true)) + var writer = new WriteBehindCacheWriter.Builder() .coalesce(BinaryOperator.maxBy(ZonedDateTime::compareTo)) - .build()) { - Cache cache = Caffeine.newBuilder().build(); - - // When this cache update happens... - cache.asMap().computeIfAbsent(1, key -> { - var value = ZonedDateTime.now(); - writer.accept(key, value); - return value; - }); - - // Then the write behind action is called - await().untilTrue(writerCalled); - } + .writeAction(entries -> writerCalled.set(true)) + .bufferTime(Duration.ofSeconds(1)) + .build(); + Cache cache = Caffeine.newBuilder().build(); + + // When this cache update happens... + cache.asMap().computeIfAbsent(1, key -> { + var value = ZonedDateTime.now(); + writer.accept(key, value); + return value; + }); + + // Then the write behind action is called + await().untilTrue(writerCalled); } @Test @@ -67,25 +66,24 @@ public void multipleKeys() { var numberOfEntries = new AtomicInteger(); // Given this cache... - try (var writer = new WriteBehindCacheWriter.Builder() - .bufferTime(1, TimeUnit.SECONDS) - .coalesce(BinaryOperator.maxBy(ZonedDateTime::compareTo)) + var writer = new WriteBehindCacheWriter.Builder() .writeAction(entries -> numberOfEntries.addAndGet(entries.size())) - .build()) { - Cache cache = Caffeine.newBuilder().build(); - - // When these cache updates happen... - for (int i = 1; i <= 3; i++) { - cache.asMap().computeIfAbsent(i, key -> { - var value = ZonedDateTime.now(); - writer.accept(key, value); - return value; - }); - } - - // Then the write behind action gets 3 entries to write - await().untilAtomic(numberOfEntries, is(3)); + .coalesce(BinaryOperator.maxBy(ZonedDateTime::compareTo)) + .bufferTime(Duration.ofSeconds(1)) + .build(); + Cache cache = Caffeine.newBuilder().build(); + + // When these cache updates happen... + for (int i = 1; i <= 3; i++) { + cache.asMap().computeIfAbsent(i, key -> { + var value = ZonedDateTime.now(); + writer.accept(key, value); + return value; + }); } + + // Then the write behind action gets 3 entries to write + await().untilAtomic(numberOfEntries, is(3)); } @Test @@ -94,9 +92,9 @@ public void singleKey_mostRecent() { var numberOfEntries = new AtomicInteger(); // Given this cache... - try (var writer = new WriteBehindCacheWriter.Builder() - .bufferTime(1, TimeUnit.SECONDS) + var writer = new WriteBehindCacheWriter.Builder() .coalesce(BinaryOperator.maxBy(ZonedDateTime::compareTo)) + .bufferTime(Duration.ofSeconds(1)) .writeAction(entries -> { // We might get here before the cache has been written to, // so just wait for the next time we are called @@ -107,24 +105,23 @@ public void singleKey_mostRecent() { var zonedDateTime = entries.values().iterator().next(); timeInWriteBehind.set(zonedDateTime); numberOfEntries.set(entries.size()); - }).build()) { - Cache cache = Caffeine.newBuilder().build(); - - // When these cache updates happen... - var latest = ZonedDateTime.now().truncatedTo(DAYS); - for (int i = 0; i < 4; i++) { - latest = latest.plusNanos(200); - - var value = latest; - cache.asMap().compute(1L, (key, oldValue) -> { - writer.accept(key, value); - return value; - }); - } - - // Then the write behind action gets 1 entry to write with the most recent time - await().untilAtomic(numberOfEntries, is(1)); - await().untilAtomic(timeInWriteBehind, is(latest)); + }).build(); + Cache cache = Caffeine.newBuilder().build(); + + // When these cache updates happen... + var latest = ZonedDateTime.now().truncatedTo(DAYS); + for (int i = 0; i < 4; i++) { + latest = latest.plusNanos(200); + + var value = latest; + cache.asMap().compute(1L, (key, oldValue) -> { + writer.accept(key, value); + return value; + }); } + + // Then the write behind action gets 1 entry to write with the most recent time + await().untilAtomic(numberOfEntries, is(1)); + await().untilAtomic(timeInWriteBehind, is(latest)); } } diff --git a/examples/write-behind-rxjava/src/test/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/package-info.java b/examples/write-behind-rxjava/src/test/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/package-info.java deleted file mode 100644 index ed1e9c2aaf..0000000000 --- a/examples/write-behind-rxjava/src/test/java/com/github/benmanes/caffeine/examples/writebehind/rxjava/package-info.java +++ /dev/null @@ -1,4 +0,0 @@ -@CheckReturnValue -package com.github.benmanes.caffeine.examples.writebehind.rxjava; - -import com.google.errorprone.annotations.CheckReturnValue;