Stop truncating Flux results to first element with @⁠CacheEvict

Prior to this commit, ReactiveCachingHandler.processCacheEvicts()
adapted every reactive return value via Mono.from(), which subscribes
for only the first element and cancels the upstream Publisher. For a
@⁠CacheEvict method that returns a Flux, this silently truncated the
returned sequence to its first element. In addition, the `#result`
variable in `condition` SpEL expressions was bound to only that first
emitted element.

To address that, this commit mirrors the existing multi-value handling
in processPutRequest(). When the adapter reports isMultiValue(), a side
Subscriber is subscribed via publish().refCount(2) that exhausts the
Flux and collects its values into a List for eviction, while the
original, unmodified Flux is returned to the caller. Consequently, the
`#result` variable in `condition` SpEL expressions for a Flux-returning
@⁠CacheEvict method is now the full List of emitted elements rather
than just the first element, making it consistent with @⁠Cacheable and
@⁠CachePut.

This commit also improves spr14235AdaptsToReactorFlux() in
CacheReproTests. Previously it exercised @⁠CacheEvict only with a
single-element Flux and never asserted on the returned sequence. Now it
uses a multi-element Flux and verifies that all elements are both
returned to the caller and visible to the `condition` expression.

Last but not least, this commit documents the aforementioned
`#result`/Flux semantics in @⁠CacheEvict's Javadoc and in the reference
manual, since both were previously invalid or incomplete for this
scenario.

Closes gh-37309

framework-docs/modules/ROOT/pages/integration/cache/annotations.adoc

@@ -381,10 +381,12 @@ available to the context so that you can use them for key and conditional comput
 
 | `result`
 | Evaluation context
-| The result of the method call (the value to be cached). Only available in `unless`
+| The result of the method call (the value to be cached, or evaluated for eviction). Only
-  expressions, `cache put` expressions (to compute the `key`), or `cache evict`
+  available in `unless` expressions, `cache put` expressions (`key`, `condition`, or
-  expressions (when `beforeInvocation` is `false`). For supported wrappers (such as
+  `unless`), or `cache evict` expressions (`key` or `condition`, when `beforeInvocation`
-  `Optional`), `#result` refers to the actual object, not the wrapper.
+  is `false`). For supported wrappers (such as `Optional`), `#result` refers to the
+  actual object, not the wrapper. For a method returning a `Flux`, `#result` refers to a
+  `List` containing all values collected from the `Flux`.
 | `#result`
 |===
 
@@ -464,7 +466,12 @@ not the case with `@Cacheable` which adds data to the cache or updates data in t
 and, thus, requires a result.
 
 As of 6.1, `@CacheEvict` takes `CompletableFuture` and reactive return types into account,
-performing an after-invocation evict operation whenever processing has completed.
+performing an after-invocation evict operation whenever processing has completed. As with
+`@Cacheable` and `@CachePut`, for a method returning a `Flux`, all emitted elements are
+collected into a `List` before the evict operation is performed. When `beforeInvocation`
+is `false`, that same `List` is what the `condition` SpEL expression sees as the
+`#result`. Either way, the `Flux` returned to the caller is unaffected by this collection
+process.
 
 TIP: When `@CacheEvict` is combined with `@Retryable`, the retry advice is applied
 outermost, so eviction runs again on every retry attempt -- and, with

spring-context/src/main/java/org/springframework/cache/annotation/CacheEvict.java

@@ -77,7 +77,8 @@ public @interface CacheEvict {
 	 * <li>{@code #result} for a reference to the result of the method invocation, which
 	 * can only be used if {@link #beforeInvocation()} is {@code false}. For supported
 	 * wrappers such as {@code Optional}, {@code #result} refers to the actual object,
-	 * not the wrapper</li>
+	 * not the wrapper. For a method that returns a {@code Flux}, {@code #result} refers
+	 * to a {@code List} containing all values collected from the {@code Flux}.</li>
 	 * <li>{@code #root.method}, {@code #root.target}, and {@code #root.caches} for
 	 * references to the {@link java.lang.reflect.Method method}, target object, and
 	 * affected cache(s) respectively.</li>
@@ -123,6 +124,11 @@ public @interface CacheEvict {
 	 * <p>The SpEL expression evaluates against a dedicated context that provides the
 	 * following meta-data:
 	 * <ul>
+	 * <li>{@code #result} for a reference to the result of the method invocation, which
+	 * can only be used if {@link #beforeInvocation()} is {@code false}. For supported
+	 * wrappers such as {@code Optional}, {@code #result} refers to the actual object,
+	 * not the wrapper. For a method that returns a {@code Flux}, {@code #result} refers
+	 * to a {@code List} containing all values collected from the {@code Flux}.</li>
 	 * <li>{@code #root.method}, {@code #root.target}, and {@code #root.caches} for
 	 * references to the {@link java.lang.reflect.Method method}, target object, and
 	 * affected cache(s) respectively.</li>

spring-context/src/main/java/org/springframework/cache/interceptor/CacheAspectSupport.java

@@ -1105,6 +1105,39 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
 	}
 
 
+	/**
+	 * Reactive Streams Subscriber for exhausting the Flux and collecting a List
+	 * to evaluate for eviction.
+	 */
+	private final class CacheEvictListSubscriber implements Subscriber<Object> {
+
+		private final List<CacheOperationContext> contexts;
+
+		private final List<Object> cacheValue = new ArrayList<>();
+
+		public CacheEvictListSubscriber(List<CacheOperationContext> contexts) {
+			this.contexts = contexts;
+		}
+
+		@Override
+		public void onSubscribe(Subscription s) {
+			s.request(Integer.MAX_VALUE);
+		}
+		@Override
+		public void onNext(Object o) {
+			this.cacheValue.add(o);
+		}
+		@Override
+		public void onError(Throwable t) {
+			this.cacheValue.clear();
+		}
+		@Override
+		public void onComplete() {
+			performCacheEvicts(this.contexts, this.cacheValue);
+		}
+	}
+
+
 	/**
 	 * Inner class to avoid a hard dependency on the Reactive Streams API at runtime.
 	 */
@@ -1186,8 +1219,16 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
 		public @Nullable Object processCacheEvicts(List<CacheOperationContext> contexts, @Nullable Object result) {
 			ReactiveAdapter adapter = (result != null ? this.registry.getAdapter(result.getClass()) : null);
 			if (adapter != null) {
-				return adapter.fromPublisher(Mono.from(adapter.toPublisher(result))
+				if (adapter.isMultiValue()) {
-						.doOnSuccess(value -> performCacheEvicts(contexts, value)));
+					Flux<?> source = Flux.from(adapter.toPublisher(result))
+							.publish().refCount(2);
+					source.subscribe(new CacheEvictListSubscriber(contexts));
+					return adapter.fromPublisher(source);
+				}
+				else {
+					return adapter.fromPublisher(Mono.from(adapter.toPublisher(result))
+							.doOnSuccess(value -> performCacheEvicts(contexts, value)));
+				}
 			}
 			return NOT_HANDLED;
 		}

spring-context/src/test/java/org/springframework/cache/CacheReproTests.java

@@ -56,6 +56,7 @@ import static org.mockito.Mockito.verify;
  * @author Phillip Webb
  * @author Juergen Hoeller
  * @author Stephane Nicoll
+ * @author Sam Brannen
  */
 class CacheReproTests {
 
@@ -312,7 +313,7 @@ class CacheReproTests {
 		assertThat(bean.findById("tb1").collectList().block()).isEqualTo(tb);
 		assertThat(cache.get("tb1").get()).isEqualTo(tb);
 
-		bean.clear().blockLast();
+		assertThat(bean.clear().collectList().block()).containsExactly(1, 2, 3);
 		List<TestBean> tb2 = bean.findById("tb1").collectList().block();
 		assertThat(tb2).isNotEmpty();
 		assertThat(tb2).isNotEqualTo(tb);
@@ -705,9 +706,9 @@ class CacheReproTests {
 			return Flux.fromIterable(item);
 		}
 
-		@CacheEvict(cacheNames = "itemCache", allEntries = true, condition = "#result > 0")
+		@CacheEvict(cacheNames = "itemCache", allEntries = true, condition = "#result == {1, 2, 3}")
 		public Flux<Integer> clear() {
-			return Flux.just(1);
+			return Flux.just(1, 2, 3);
 		}
 	}