vertx 4: CircuitBreaker and coroutine
175 views
Skip to first unread message
Petros
Jan 26, 2021, 2:28:42 AM1/26/21
to vert.x
Hello, i am using `CircuitBreaker` for some retrying, but i need to start new coroutine in it so i can call suspend function etc. like this
```kotlin
suspend fun <T> retry(
context: Context,
customize: RetryCustomConf? = null,
block: Producer<T>
): T {
val future = circuitBreaker.execute<T> { promise ->
GlobalScope.launch(context.vertx.dispatcher()) {
try {
promise.complete(block()) // promise completion
} catch (exc: Throwable) {
customize?.filter?.let { list ->
list.forEach { filtered ->
if (exc::class.isSubclassOf(filtered).not()) {
unhandledException = exc // this exception should not been handled here
promise.complete() // quit retry loop
}
}
}
customize?.negative_filter?.let { list ->
list.forEach { filtered ->
if (exc::class.isSubclassOf(filtered)) {
unhandledException = exc // this exception should not been handled here
promise.complete() // quit retry loop
}
}
}
throw exc
}
return@launch
}
}
val result = future.await()
unhandledException?.let { throw it } // throw exception after retry loop
return result
}
```
is there some other solution so i dont need use `GlobalScope.launch(context.vertx.dispatcher())` in it?
Thomas SEGISMONT
Jan 26, 2021, 4:58:00 AM1/26/21
to vert.x
I don't believe it would be necessary if you do this inside a CoroutineVerticle class.
--
You received this message because you are subscribed to the Google Groups "vert.x" group.
To unsubscribe from this group and stop receiving emails from it, send an email to vertx+un...@googlegroups.com.
To view this discussion on the web, visit https://groups.google.com/d/msgid/vertx/eca39700-c243-45e6-ae98-981e5d148fd0n%40googlegroups.com.
Petros
Jan 26, 2021, 5:37:53 AM1/26/21
to vert.x
I dont have coroutine verticle as it cannot be combined with vertx grpc. i run coroutine in grpc method same as in circuit breaker (`
GlobalScope.launch(context.vertx.dispatcher()) `)
Dne úterý 26. ledna 2021 v 10:58:00 UTC+1 uživatel tsegi...@gmail.com napsal:
Mattias Fagerström
Jan 26, 2021, 9:09:23 AM1/26/21
to vert.x
I have two extension functions for CircuitBreaker that I use when wrapping a suspending function:
val circuitBreaker = this
return withContext(coroutineContext) {
circuitBreaker.execute { promise: Promise<T> -> executeCommand(command, promise) }.await()
}
}
suspend fun <T> CircuitBreaker.executeCoroutineWithFallbackAwait(command: suspend () -> T, fallback: (Throwable) -> T): T {
val circuitBreaker = this
return withContext(coroutineContext) {
circuitBreaker.executeWithFallback({ promise: Promise<T> -> executeCommand(command, promise) }, fallback).await()
}
}
private fun <T> CoroutineScope.executeCommand(command: suspend () -> T, promise: Promise<T>) {
this.launch {
runCatching {
command()
}.onFailure {
promise.fail(it)
}.onSuccess {
promise.complete(it)
}
}
}
```kotlin
suspend fun <T> CircuitBreaker.executeCoroutineAwait(command: suspend () -> T): T {val circuitBreaker = this
return withContext(coroutineContext) {
circuitBreaker.execute { promise: Promise<T> -> executeCommand(command, promise) }.await()
}
}
suspend fun <T> CircuitBreaker.executeCoroutineWithFallbackAwait(command: suspend () -> T, fallback: (Throwable) -> T): T {
val circuitBreaker = this
return withContext(coroutineContext) {
circuitBreaker.executeWithFallback({ promise: Promise<T> -> executeCommand(command, promise) }, fallback).await()
}
}
private fun <T> CoroutineScope.executeCommand(command: suspend () -> T, promise: Promise<T>) {
this.launch {
runCatching {
command()
}.onFailure {
promise.fail(it)
}.onSuccess {
promise.complete(it)
}
}
}
```
Reply all
Reply to author
Forward
0 new messages