Add a CoExchangeFilterFunction for WebClient

This commit introduces a Kotlin-specific implementation
of the ExchangeFilterFunction interface that allows for
using coroutines, similar to what CoWebFilter does on
server-side.

Closes gh-30650

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/CoExchangeFilterFunction.kt

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2002-2023 the original author or 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.
+ */
+
+package org.springframework.web.reactive.function.client
+
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.reactor.awaitSingle
+import kotlinx.coroutines.reactor.mono
+import reactor.core.publisher.Mono
+
+/**
+ * Kotlin-specific implementation of the [ExchangeFilterFunction] interface that allows for
+ * using coroutines.
+ *
+ * @author Sebastien Deleuze
+ * @since 6.1.0
+ */
+abstract class CoExchangeFilterFunction : ExchangeFilterFunction {
+
+	final override fun filter(request: ClientRequest, next: ExchangeFunction): Mono<ClientResponse> {
+		return mono(Dispatchers.Unconfined) {
+			filter(request, object : CoExchangeFunction {
+				override suspend fun exchange(request: ClientRequest): ClientResponse {
+					return next.exchange(request).awaitSingle()
+				}
+			})
+		}
+	}
+
+	/**
+	 * Apply this filter to the given request and exchange function.
+	 *
+	 * The given [CoExchangeFunction] represents the next entity in the
+	 * chain, to be invoked via [CoExchangeFunction.exchange] in order to
+	 * proceed with the exchange, or not invoked to short-circuit the chain.
+	 *
+	 * **Note:** When a filter handles the response after the
+	 * call to [CoExchangeFunction.exchange], extra care must be taken
+	 * to always consume its content or otherwise propagate it downstream for
+	 * further handling, for example by the [WebClient]. Please see the
+	 * reference documentation for more details on this.
+	 *
+	 * @param request the current request
+	 * @param next the next exchange function in the chain
+	 * @return the filtered response
+	 */
+	protected abstract suspend fun filter(request: ClientRequest, next: CoExchangeFunction): ClientResponse
+}
+
+/**
+ * Kotlin-specific adaption of [ExchangeFunction] that allows for coroutines.
+ *
+ * @author Sebastien Deleuze
+ * @since 6.1.0
+ */
+interface CoExchangeFunction {
+
+	/**
+	 * Exchange the given request for a [ClientResponse].
+	 *
+	 * **Note:** When calling this method from an
+	 * [CoExchangeFilterFunction] that handles the response in some way,
+	 * extra care must be taken to always consume its content or otherwise
+	 * propagate it downstream for further handling, for example by the
+	 * [WebClient]. Please, see the reference documentation for more
+	 * details on this.
+	 *
+	 * @param request the request to exchange
+	 * @return the delayed response
+	 */
+	suspend fun exchange(request: ClientRequest): ClientResponse
+}

spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/CoExchangeFilterFunctionTests.kt

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2002-2023 the original author or 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.
+ */
+
+package org.springframework.web.reactive.function.client
+
+import io.mockk.mockk
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+
+/**
+ * Test for [CoExchangeFilterFunction].
+ *
+ * @author Sebastien Deleuze
+ */
+class CoExchangeFilterFunctionTests {
+
+	@Test
+	fun exchange() {
+		val response = mockk<ClientResponse>()
+		val exchangeFunction = MyCoExchangeFunction(response)
+		runBlocking {
+			assertThat(exchangeFunction.exchange(mockk())).isEqualTo(response)
+		}
+	}
+}
+
+private class MyCoExchangeFunction(private val response: ClientResponse) : CoExchangeFunction {
+
+	override suspend fun exchange(request: ClientRequest): ClientResponse {
+		delay(1)
+		return response
+	}
+}