Improve fix of duplicate upstream subscription during reactive cache put

This commit fixes an issue where a Cacheable method which returns a
Flux (or multi-value publisher) will be invoked once, but the returned
publisher is actually subscribed twice.

The previous fix 988f3630c would cause the cached elements to depend on
the first usage pattern / request pattern, which is likely to be too
confusing to users. This fix reintroduces the notion of exhausting the
original Flux by having a second subscriber dedicated to that, but uses
`refCount(2)` to ensure that the original `Flux` returned by the cached
method is still only subscribed once.

Closes gh-32370

spring-context/src/test/java/org/springframework/cache/annotation/ReactiveCachingTests.java

@@ -111,6 +111,30 @@ class ReactiveCachingTests {
 		ctx.close();
 	}
 
+	@ParameterizedTest
+	@ValueSource(classes = {EarlyCacheHitDeterminationConfig.class,
+			EarlyCacheHitDeterminationWithoutNullValuesConfig.class,
+			LateCacheHitDeterminationConfig.class,
+			LateCacheHitDeterminationWithValueWrapperConfig.class})
+	void fluxCacheDoesntDependOnFirstRequest(Class<?> configClass) {
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(configClass, ReactiveCacheableService.class);
+		ReactiveCacheableService service = ctx.getBean(ReactiveCacheableService.class);
+
+		Object key = new Object();
+
+		List<Long> l1 = service.cacheFlux(key).take(1L, true).collectList().block();
+		List<Long> l2 = service.cacheFlux(key).take(3L, true).collectList().block();
+		List<Long> l3 = service.cacheFlux(key).collectList().block();
+
+		Long first = l1.get(0);
+
+		assertThat(l1).as("l1").containsExactly(first);
+		assertThat(l2).as("l2").containsExactly(first, 0L, -1L);
+		assertThat(l3).as("l3").containsExactly(first, 0L, -1L, -2L, -3L);
+
+		ctx.close();
+	}
+
 	@CacheConfig(cacheNames = "first")
 	static class ReactiveCacheableService {
 
@@ -132,7 +156,7 @@ class ReactiveCachingTests {
 		Flux<Long> cacheFlux(Object arg) {
 			// here counter not only reflects invocations of cacheFlux but subscriptions to
 			// the returned Flux as well. See https://github.com/spring-projects/spring-framework/issues/32370
-			return Flux.defer(() -> Flux.just(this.counter.getAndIncrement(), 0L));
+			return Flux.defer(() -> Flux.just(this.counter.getAndIncrement(), 0L, -1L, -2L, -3L));
 		}
 	}