MDS/openide
1
0
Fork 0
mirror of https://gitflic.ru/project/openide/openide.git synced 2026-09-27 10:03:11 +07:00
Code Issues Packages Projects Releases Wiki Activity

[starter] LocalEventsFlow: exit postAndWaitProcessing early if there are no subscribers

Browse Source 
GitOrigin-RevId: 97a16726c3c8dfa15aab73cd2a967cceb1d9230f
This commit is contained in:
Anastasia Katsman
2025-10-07 17:13:18 +00:00
committed by intellij-monorepo-bot
parent 009f9c8f38
commit 65d856f7bf
1 changed files with 32 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/performanceTesting/event-bus/src/com/intellij/tools/ide/starter/bus/local/LocalEventsFlow.kt
+32 -27
View File
@@ -32,6 +32,7 @@ class LocalEventsFlow : EventsFlow {
unsubscribeNoLock(eventClass, subscriber)
}
}
private fun <EventType : Event> unsubscribeNoLock(eventClass: Class<EventType>, subscriber: Any) {
val eventClassName = eventClass.simpleName
val subscriberName = getSubscriberObject(subscriber)
@@ -84,41 +85,45 @@ class LocalEventsFlow : EventsFlow {
}
}
}
if (subscribersForEvent.isNullOrEmpty()) {
return
}
val exceptions = CopyOnWriteArrayList<Throwable>()
(subscribersForEvent as? List<Subscriber<T>>)
?.map { subscriber ->
// In case the job is interrupted (e.g. due to timeout), the coroutine may enter Cancelling state
// and finish before the 'catch' block is executed. Using CompletableDeferred ensures we wait
// until either successful completion or proper exception handling has occurred.
val result = CompletableDeferred<Unit>()
LOG.debug("Post event $eventClassName for $subscriber.")
// Launching a new coroutine for each subscriber
scope.launch(Dispatchers.IO) {
LOG.debug("Start execution $eventClassName for $subscriber")
// Enforces a timeout for the entire subscriber execution
withTimeout(subscriber.timeout) {
// Ensures the operation inside is interruptible — if the thread is blocked,
// it will be interrupted when the coroutine is cancelled (e.g. by timeout)
runInterruptible {
try {
runBlocking {
subscriber.callback(event)
}
result.complete(Unit)
}
catch (e: Throwable) {
exceptions.add(e)
result.complete(Unit)
val tasks = (subscribersForEvent as List<Subscriber<T>>).map { subscriber ->
// In case the job is interrupted (e.g. due to timeout), the coroutine may enter Cancelling state
// and finish before the 'catch' block is executed. Using CompletableDeferred ensures we wait
// until either successful completion or proper exception handling has occurred.
val result = CompletableDeferred<Unit>()
LOG.debug("Post event $eventClassName for $subscriber.")
// Launching a new coroutine for each subscriber
scope.launch(Dispatchers.IO) {
LOG.debug("Start execution $eventClassName for $subscriber")
// Enforces a timeout for the entire subscriber execution
withTimeout(subscriber.timeout) {
// Ensures the operation inside is interruptible — if the thread is blocked,
// it will be interrupted when the coroutine is cancelled (e.g. by timeout)
runInterruptible {
try {
runBlocking {
subscriber.callback(event)
}
result.complete(Unit)
}
catch (e: Throwable) {
exceptions.add(e)
result.complete(Unit)
}
}
LOG.debug("Finished execution $eventClassName for $subscriber")
}
return@map result
LOG.debug("Finished execution $eventClassName for $subscriber")
}
return@map result
}
runBlocking {
// awaitAll shouldn't be used as it would stop after the first exception from any coroutine
tasks?.forEach {
tasks.forEach {
try {
it.await()
}