diff --git a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport.kt b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport.kt index 8df42865..fdb0e83a 100644 --- a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport.kt +++ b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport.kt @@ -127,7 +127,8 @@ public class JdkHttpTransport private constructor( /** * Asynchronously executes [request]. Returns a [CompletableFuture] that completes with * the [Response] on success or completes exceptionally with the transport failure on - * error. + * error — including a request-adaptation failure, which runs on the calling thread but is + * delivered through the future rather than thrown synchronously. * * Cancelling the returned future cancels the underlying JDK exchange, and a response that * arrives in the cancellation race window is closed so its connection is not leaked (see @@ -137,7 +138,19 @@ public class JdkHttpTransport private constructor( * completions to release the body's connection back to the pool. */ override fun executeAsync(request: Request): CompletableFuture { - val jdkRequest = requestAdapter.adapt(request, responseTimeout) + val jdkRequest = + try { + requestAdapter.adapt(request, responseTimeout) + } catch (e: Exception) { + // The async contract is that errors arrive through the returned future. Request + // adaptation runs on the caller's thread and can throw (e.g. a CONNECT request the + // JDK client rejects), so route the failure into a failed future instead of + // throwing synchronously where a future-composing caller's .exceptionally/.handle + // would never observe it. Errors (OOM and other JVM-fatal conditions) are left to + // propagate up the caller's stack rather than be packaged into a future that may + // never be awaited. + return CompletableFuture.failedFuture(e) + } val inFlight = client.sendAsync(jdkRequest, HttpResponse.BodyHandlers.ofInputStream()) return bridgeAsyncResponse(inFlight) { jdkResponse -> responseAdapter.adapt(request, jdkResponse) } } diff --git a/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt b/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt index 42e2e4c6..0553c0b5 100644 --- a/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt +++ b/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt @@ -35,6 +35,7 @@ import java.time.Duration import java.util.concurrent.CancellationException import java.util.concurrent.CompletionException import java.util.concurrent.CountDownLatch +import java.util.concurrent.ExecutionException import java.util.concurrent.Flow import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean @@ -44,6 +45,7 @@ import kotlin.test.Test import kotlin.test.assertContentEquals import kotlin.test.assertEquals import kotlin.test.assertFails +import kotlin.test.assertFailsWith import kotlin.test.assertNotNull import kotlin.test.assertTrue @@ -144,6 +146,29 @@ class JdkHttpTransportTest { assertEquals(payload, recorded.body?.utf8()) } + // -------- async adaptation failures -------- + + @Test + fun `executeAsyncDeliversAdaptationFailureThroughFuture`() { + // A CONNECT request makes request adaptation throw synchronously inside executeAsync + // (the JDK client reserves CONNECT for internal tunnelling). The contract is that + // executeAsync completes exceptionally on error, so the failure must arrive through the + // returned future — a future-composing caller's .exceptionally/.handle would never + // observe a synchronous throw. + val request = + Request.builder() + .method(Method.CONNECT) + .url(server.url("/async-adapt-fail").toUrl()) + .build() + // Must return a future rather than throwing on the caller's thread. + val future = transport.executeAsync(request) + val ex = assertFailsWith { future.get(5, TimeUnit.SECONDS) } + assertTrue( + ex.cause is IllegalArgumentException, + "adaptation failure must surface as the future's cause, was: ${ex.cause?.let { it::class }}", + ) + } + // -------- headers round-trip -------- @Test diff --git a/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransport.kt b/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransport.kt index 9f0950ad..c2a30777 100644 --- a/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransport.kt +++ b/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransport.kt @@ -103,10 +103,24 @@ public class OkHttpTransport private constructor( /** * Asynchronously executes [request]. Returns a [CompletableFuture] that completes with * the [Response] on success or completes exceptionally with the transport failure on - * error. Cancelling the future cancels the underlying OkHttp [Call]. + * error — including a request-adaptation failure, which runs on the calling thread but is + * delivered through the future rather than thrown synchronously. Cancelling the future + * cancels the underlying OkHttp [Call]. */ override fun executeAsync(request: Request): CompletableFuture { - val okRequest = requestAdapter.adapt(request) + val okRequest = + try { + requestAdapter.adapt(request) + } catch (e: Exception) { + // The async contract is that errors arrive through the returned future. Request + // adaptation runs on the caller's thread and can throw (e.g. a method/body + // mismatch OkHttp rejects), so route the failure into a completed-exceptionally + // future instead of throwing synchronously where a future-composing caller's + // .exceptionally/.handle would never observe it. Errors (OOM and other JVM-fatal + // conditions) are left to propagate up the caller's stack rather than be packaged + // into a future that may never be awaited. + return failedFuture(e) + } val call = client.newCall(okRequest) val future = CompletableFuture() call.enqueue( @@ -148,6 +162,14 @@ public class OkHttpTransport private constructor( return future } + /** + * A future already completed exceptionally with [t]. Routes a synchronous request-adaptation + * failure through the async contract's single error channel rather than throwing on the + * caller's thread. ([CompletableFuture.failedFuture] is JDK 9+; this module targets Java 8.) + */ + private fun failedFuture(t: Throwable): CompletableFuture = + CompletableFuture().apply { completeExceptionally(t) } + /** * Best-effort close on a discard path. Used when an adapted [Response] loses the race to a * cancelled future: the response is already being discarded, so a failure to close it has diff --git a/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt b/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt index c466801f..a908db47 100644 --- a/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt +++ b/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt @@ -37,6 +37,7 @@ import java.time.Duration import java.util.concurrent.CancellationException import java.util.concurrent.CompletionException import java.util.concurrent.CountDownLatch +import java.util.concurrent.ExecutionException import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger @@ -46,6 +47,7 @@ import kotlin.test.Test import kotlin.test.assertContentEquals import kotlin.test.assertEquals import kotlin.test.assertFails +import kotlin.test.assertFailsWith import kotlin.test.assertFalse import kotlin.test.assertNotNull import kotlin.test.assertTrue @@ -134,6 +136,29 @@ class OkHttpTransportTest { assertEquals(payload, recorded.body?.utf8()) } + // -------- async adaptation failures -------- + + @Test + fun executeAsyncDeliversAdaptationFailureThroughFuture() { + // A body on a method OkHttp forbids one for (GET) makes request adaptation throw + // synchronously inside executeAsync. The contract is that executeAsync completes + // exceptionally on error, so the failure must arrive through the returned future — a + // future-composing caller's .exceptionally/.handle would never observe a synchronous throw. + val request = + Request.builder() + .method(Method.GET) + .url(server.url("/async-adapt-fail").toUrl()) + .body(RequestBody.create("x".toByteArray())) + .build() + // Must return a future rather than throwing on the caller's thread. + val future = transport.executeAsync(request) + val ex = assertFailsWith { future.get(5, TimeUnit.SECONDS) } + assertTrue( + ex.cause is IllegalArgumentException, + "adaptation failure must surface as the future's cause, was: ${ex.cause?.let { it::class }}", + ) + } + // -------- headers round-trip -------- @Test