SynchronousEventEmitter.kt

import java.util.concurrent.*
import kotlin.reflect.*
import kotlinx.coroutines.*


class SynchronousEventEmitter(
	private val onError: (error: Throwable, event: Event) -> Unit,
) : EventEmitter, EventSource {

	private val subscriptions = CopyOnWriteArraySet<Subscription<*>>()


	override suspend fun emit(event: Event) {
		subscriptions
			.mapNotNull { it.dispatchAsync(event) }
			.forEach { it.await() }
	}


	override fun <TEvent : Event> subscribeIn(scope: CoroutineScope, event: KClass<out TEvent>, action: suspend (TEvent) -> Unit): EventSubscription {
		val subscription = Subscription(action = action, eventClass = event, scope = scope)
		subscriptions += subscription

		scope.coroutineContext.job.invokeOnCompletion {
			subscription.unsubscribe()
		}

		return subscription
	}


	private inner class Subscription<in TEvent : Event>(
		private val action: suspend (TEvent) -> Unit,
		private val eventClass: KClass<out TEvent>,
		private val scope: CoroutineScope,
	) : EventSubscription {

		@Suppress("NAME_SHADOWING")
		fun dispatchAsync(event: Event): Deferred<Unit>? {
			val event = eventClass.safeCast(event) ?: return null

			return scope.async {
				try {
					action(event)
				}
				catch (error: Throwable) {
					onError(error, event)
				}
			}
		}


		override fun unsubscribe() {
			subscriptions -= this
		}
	}
}