Skip to content

Commit 427df35

Browse files
stainless-app[bot]stainless-app[bot]
committed
feat(client): expose sleeper option
fix(client): ensure single timer is created per client
1 parent 9c8977c commit 427df35

File tree

8 files changed

+158
-48
lines changed

8 files changed

+158
-48
lines changed

‎orb-java-client-okhttp/src/main/kotlin/com/withorb/api/client/okhttp/OrbOkHttpClient.kt‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import com.fasterxml.jackson.databind.json.JsonMapper
66
import com.withorb.api.client.OrbClient
77
import com.withorb.api.client.OrbClientImpl
88
import com.withorb.api.core.ClientOptions
9+
import com.withorb.api.core.Sleeper
910
import com.withorb.api.core.Timeout
1011
import com.withorb.api.core.http.AsyncStreamResponse
1112
import com.withorb.api.core.http.Headers
@@ -133,6 +134,17 @@ class OrbOkHttpClient private constructor() {
133134
clientOptions.streamHandlerExecutor(streamHandlerExecutor)
134135
}
135136

137+
/**
138+
* The interface to use for delaying execution, like during retries.
139+
*
140+
* This is primarily useful for using fake delays in tests.
141+
*
142+
* Defaults to real execution delays.
143+
*
144+
* This class takes ownership of the sleeper and closes it when closed.
145+
*/
146+
fun sleeper(sleeper: Sleeper) = apply { clientOptions.sleeper(sleeper) }
147+
136148
/**
137149
* The clock to use for operations that require timing, like retries.
138150
*

‎orb-java-client-okhttp/src/main/kotlin/com/withorb/api/client/okhttp/OrbOkHttpClientAsync.kt‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import com.fasterxml.jackson.databind.json.JsonMapper
66
import com.withorb.api.client.OrbClientAsync
77
import com.withorb.api.client.OrbClientAsyncImpl
88
import com.withorb.api.core.ClientOptions
9+
import com.withorb.api.core.Sleeper
910
import com.withorb.api.core.Timeout
1011
import com.withorb.api.core.http.AsyncStreamResponse
1112
import com.withorb.api.core.http.Headers
@@ -133,6 +134,17 @@ class OrbOkHttpClientAsync private constructor() {
133134
clientOptions.streamHandlerExecutor(streamHandlerExecutor)
134135
}
135136

137+
/**
138+
* The interface to use for delaying execution, like during retries.
139+
*
140+
* This is primarily useful for using fake delays in tests.
141+
*
142+
* Defaults to real execution delays.
143+
*
144+
* This class takes ownership of the sleeper and closes it when closed.
145+
*/
146+
fun sleeper(sleeper: Sleeper) = apply { clientOptions.sleeper(sleeper) }
147+
136148
/**
137149
* The clock to use for operations that require timing, like retries.
138150
*

‎orb-java-core/src/main/kotlin/com/withorb/api/core/ClientOptions.kt‎

Lines changed: 46 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,16 @@ private constructor(
5454
* This class takes ownership of the executor and shuts it down, if possible, when closed.
5555
*/
5656
@get:JvmName("streamHandlerExecutor") val streamHandlerExecutor: Executor,
57+
/**
58+
* The interface to use for delaying execution, like during retries.
59+
*
60+
* This is primarily useful for using fake delays in tests.
61+
*
62+
* Defaults to real execution delays.
63+
*
64+
* This class takes ownership of the sleeper and closes it when closed.
65+
*/
66+
@get:JvmName("sleeper") val sleeper: Sleeper,
5767
/**
5868
* The clock to use for operations that require timing, like retries.
5969
*
@@ -148,6 +158,7 @@ private constructor(
148158
private var checkJacksonVersionCompatibility: Boolean = true
149159
private var jsonMapper: JsonMapper = jsonMapper()
150160
private var streamHandlerExecutor: Executor? = null
161+
private var sleeper: Sleeper? = null
151162
private var clock: Clock = Clock.systemUTC()
152163
private var baseUrl: String? = null
153164
private var headers: Headers.Builder = Headers.builder()
@@ -164,6 +175,7 @@ private constructor(
164175
checkJacksonVersionCompatibility = clientOptions.checkJacksonVersionCompatibility
165176
jsonMapper = clientOptions.jsonMapper
166177
streamHandlerExecutor = clientOptions.streamHandlerExecutor
178+
sleeper = clientOptions.sleeper
167179
clock = clientOptions.clock
168180
baseUrl = clientOptions.baseUrl
169181
headers = clientOptions.headers.toBuilder()
@@ -219,6 +231,17 @@ private constructor(
219231
else streamHandlerExecutor
220232
}
221233

234+
/**
235+
* The interface to use for delaying execution, like during retries.
236+
*
237+
* This is primarily useful for using fake delays in tests.
238+
*
239+
* Defaults to real execution delays.
240+
*
241+
* This class takes ownership of the sleeper and closes it when closed.
242+
*/
243+
fun sleeper(sleeper: Sleeper) = apply { this.sleeper = PhantomReachableSleeper(sleeper) }
244+
222245
/**
223246
* The clock to use for operations that require timing, like retries.
224247
*
@@ -410,6 +433,25 @@ private constructor(
410433
*/
411434
fun build(): ClientOptions {
412435
val httpClient = checkRequired("httpClient", httpClient)
436+
val streamHandlerExecutor =
437+
streamHandlerExecutor
438+
?: PhantomReachableExecutorService(
439+
Executors.newCachedThreadPool(
440+
object : ThreadFactory {
441+
442+
private val threadFactory: ThreadFactory =
443+
Executors.defaultThreadFactory()
444+
private val count = AtomicLong(0)
445+
446+
override fun newThread(runnable: Runnable): Thread =
447+
threadFactory.newThread(runnable).also {
448+
it.name =
449+
"orb-stream-handler-thread-${count.getAndIncrement()}"
450+
}
451+
}
452+
)
453+
)
454+
val sleeper = sleeper ?: PhantomReachableSleeper(DefaultSleeper())
413455
val apiKey = checkRequired("apiKey", apiKey)
414456

415457
val headers = Headers.builder()
@@ -433,26 +475,15 @@ private constructor(
433475
httpClient,
434476
RetryingHttpClient.builder()
435477
.httpClient(httpClient)
478+
.sleeper(sleeper)
436479
.clock(clock)
437480
.maxRetries(maxRetries)
438481
.idempotencyHeader("Idempotency-Key")
439482
.build(),
440483
checkJacksonVersionCompatibility,
441484
jsonMapper,
442-
streamHandlerExecutor
443-
?: Executors.newCachedThreadPool(
444-
object : ThreadFactory {
445-
446-
private val threadFactory: ThreadFactory =
447-
Executors.defaultThreadFactory()
448-
private val count = AtomicLong(0)
449-
450-
override fun newThread(runnable: Runnable): Thread =
451-
threadFactory.newThread(runnable).also {
452-
it.name = "orb-stream-handler-thread-${count.getAndIncrement()}"
453-
}
454-
}
455-
),
485+
streamHandlerExecutor,
486+
sleeper,
456487
clock,
457488
baseUrl,
458489
headers.build(),
@@ -479,5 +510,6 @@ private constructor(
479510
fun close() {
480511
httpClient.close()
481512
(streamHandlerExecutor as? ExecutorService)?.shutdown()
513+
sleeper.close()
482514
}
483515
}

‎orb-java-core/src/main/kotlin/com/withorb/api/core/DefaultSleeper.kt‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
package com.withorb.api.core
2+
3+
import java.time.Duration
4+
import java.util.Timer
5+
import java.util.TimerTask
6+
import java.util.concurrent.CompletableFuture
7+
8+
class DefaultSleeper : Sleeper {
9+
10+
private val timer = Timer("DefaultSleeper", true)
11+
12+
override fun sleep(duration: Duration) = Thread.sleep(duration.toMillis())
13+
14+
override fun sleepAsync(duration: Duration): CompletableFuture<Void> {
15+
val future = CompletableFuture<Void>()
16+
timer.schedule(
17+
object : TimerTask() {
18+
override fun run() {
19+
future.complete(null)
20+
}
21+
},
22+
duration.toMillis(),
23+
)
24+
return future
25+
}
26+
27+
override fun close() = timer.cancel()
28+
}

‎orb-java-core/src/main/kotlin/com/withorb/api/core/PhantomReachableSleeper.kt‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
package com.withorb.api.core
2+
3+
import java.time.Duration
4+
import java.util.concurrent.CompletableFuture
5+
6+
/**
7+
* A delegating wrapper around a [Sleeper] that closes it once it's only phantom reachable.
8+
*
9+
* This class ensures the [Sleeper] is closed even if the user forgets to do it.
10+
*/
11+
internal class PhantomReachableSleeper(private val sleeper: Sleeper) : Sleeper {
12+
13+
init {
14+
closeWhenPhantomReachable(this, sleeper)
15+
}
16+
17+
override fun sleep(duration: Duration) = sleeper.sleep(duration)
18+
19+
override fun sleepAsync(duration: Duration): CompletableFuture<Void> =
20+
sleeper.sleepAsync(duration)
21+
22+
override fun close() = sleeper.close()
23+
}

‎orb-java-core/src/main/kotlin/com/withorb/api/core/Sleeper.kt‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
package com.withorb.api.core
2+
3+
import java.time.Duration
4+
import java.util.concurrent.CompletableFuture
5+
6+
/**
7+
* An interface for delaying execution for a specified amount of time.
8+
*
9+
* Useful for testing and cleaning up resources.
10+
*/
11+
interface Sleeper : AutoCloseable {
12+
13+
/** Synchronously pauses execution for the given [duration]. */
14+
fun sleep(duration: Duration)
15+
16+
/** Asynchronously pauses execution for the given [duration]. */
17+
fun sleepAsync(duration: Duration): CompletableFuture<Void>
18+
19+
/** Overridden from [AutoCloseable] to not have a checked exception in its signature. */
20+
override fun close()
21+
}

‎orb-java-core/src/main/kotlin/com/withorb/api/core/http/RetryingHttpClient.kt‎

Lines changed: 9 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
package com.withorb.api.core.http
22

3+
import com.withorb.api.core.DefaultSleeper
34
import com.withorb.api.core.RequestOptions
5+
import com.withorb.api.core.Sleeper
46
import com.withorb.api.core.checkRequired
57
import com.withorb.api.errors.OrbIoException
68
import com.withorb.api.errors.OrbRetryableException
@@ -11,8 +13,6 @@ import java.time.OffsetDateTime
1113
import java.time.format.DateTimeFormatter
1214
import java.time.format.DateTimeParseException
1315
import java.time.temporal.ChronoUnit
14-
import java.util.Timer
15-
import java.util.TimerTask
1616
import java.util.UUID
1717
import java.util.concurrent.CompletableFuture
1818
import java.util.concurrent.ThreadLocalRandom
@@ -130,7 +130,10 @@ private constructor(
130130
return executeWithRetries(modifiedRequest, requestOptions)
131131
}
132132

133-
override fun close() = httpClient.close()
133+
override fun close() {
134+
httpClient.close()
135+
sleeper.close()
136+
}
134137

135138
private fun isRetryable(request: HttpRequest): Boolean =
136139
// Some requests, such as when a request body is being streamed, cannot be retried because
@@ -235,33 +238,14 @@ private constructor(
235238
class Builder internal constructor() {
236239

237240
private var httpClient: HttpClient? = null
238-
private var sleeper: Sleeper =
239-
object : Sleeper {
240-
241-
private val timer = Timer("RetryingHttpClient", true)
242-
243-
override fun sleep(duration: Duration) = Thread.sleep(duration.toMillis())
244-
245-
override fun sleepAsync(duration: Duration): CompletableFuture<Void> {
246-
val future = CompletableFuture<Void>()
247-
timer.schedule(
248-
object : TimerTask() {
249-
override fun run() {
250-
future.complete(null)
251-
}
252-
},
253-
duration.toMillis(),
254-
)
255-
return future
256-
}
257-
}
241+
private var sleeper: Sleeper? = null
258242
private var clock: Clock = Clock.systemUTC()
259243
private var maxRetries: Int = 2
260244
private var idempotencyHeader: String? = null
261245

262246
fun httpClient(httpClient: HttpClient) = apply { this.httpClient = httpClient }
263247

264-
@JvmSynthetic internal fun sleeper(sleeper: Sleeper) = apply { this.sleeper = sleeper }
248+
fun sleeper(sleeper: Sleeper) = apply { this.sleeper = sleeper }
265249

266250
fun clock(clock: Clock) = apply { this.clock = clock }
267251

@@ -272,17 +256,10 @@ private constructor(
272256
fun build(): HttpClient =
273257
RetryingHttpClient(
274258
checkRequired("httpClient", httpClient),
275-
sleeper,
259+
sleeper ?: DefaultSleeper(),
276260
clock,
277261
maxRetries,
278262
idempotencyHeader,
279263
)
280264
}
281-
282-
internal interface Sleeper {
283-
284-
fun sleep(duration: Duration)
285-
286-
fun sleepAsync(duration: Duration): CompletableFuture<Void>
287-
}
288265
}

‎orb-java-core/src/test/kotlin/com/withorb/api/core/http/RetryingHttpClientTest.kt‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import com.github.tomakehurst.wiremock.junit5.WireMockTest
66
import com.github.tomakehurst.wiremock.stubbing.Scenario
77
import com.withorb.api.client.okhttp.OkHttpClient
88
import com.withorb.api.core.RequestOptions
9+
import com.withorb.api.core.Sleeper
910
import com.withorb.api.errors.OrbRetryableException
1011
import java.io.InputStream
1112
import java.time.Duration
@@ -294,12 +295,14 @@ internal class RetryingHttpClientTest {
294295
.httpClient(failingHttpClient)
295296
.maxRetries(2)
296297
.sleeper(
297-
object : RetryingHttpClient.Sleeper {
298+
object : Sleeper {
298299

299300
override fun sleep(duration: Duration) {}
300301

301302
override fun sleepAsync(duration: Duration): CompletableFuture<Void> =
302303
CompletableFuture.completedFuture(null)
304+
305+
override fun close() {}
303306
}
304307
)
305308
.build()
@@ -333,12 +336,14 @@ internal class RetryingHttpClientTest {
333336
.httpClient(httpClient)
334337
// Use a no-op `Sleeper` to make the test fast.
335338
.sleeper(
336-
object : RetryingHttpClient.Sleeper {
339+
object : Sleeper {
337340

338341
override fun sleep(duration: Duration) {}
339342

340343
override fun sleepAsync(duration: Duration): CompletableFuture<Void> =
341344
CompletableFuture.completedFuture(null)
345+
346+
override fun close() {}
342347
}
343348
)
344349

0 commit comments

Comments
 (0)