Skip to content

Listener semmantics and dynamic scheduling #89

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Avanatiker merged 2 commits into from
Dec 7, 2024
Merged

Listener semmantics and dynamic scheduling #89

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f7ea93b
Refactor listener registrations `listener` -> `listen`, new scheduling
Avanatiker Dec 6, 2024
9b7786f
Cleanup
Avanatiker Dec 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions common/src/main/kotlin/com/lambda/config/Configuration.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ import com.lambda.Lambda.LOG
import com.lambda.Lambda.gson
import com.lambda.config.configurations.ModuleConfig
import com.lambda.event.events.ClientEvent
import com.lambda.event.listener.UnsafeListener.Companion.unsafeListener
import com.lambda.event.listener.UnsafeListener.Companion.listenUnsafe
import com.lambda.threading.runIO
import com.lambda.util.Communication.info
import com.lambda.util.Communication.logError
Expand Down Expand Up @@ -57,9 +57,9 @@ abstract class Configuration : Jsonable {
get() = File("${primary.parent}/${primary.nameWithoutExtension}-backup.${primary.extension}")

init {
unsafeListener<ClientEvent.Startup> { tryLoad() }
listenUnsafe<ClientEvent.Startup> { tryLoad() }

unsafeListener<ClientEvent.Shutdown>(Int.MIN_VALUE) { trySave() }
listenUnsafe<ClientEvent.Shutdown>(Int.MIN_VALUE) { trySave() }

register()
}
Expand Down
8 changes: 4 additions & 4 deletions common/src/main/kotlin/com/lambda/core/PingManager.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ package com.lambda.core

import com.lambda.event.events.PacketEvent
import com.lambda.event.events.TickEvent
import com.lambda.event.listener.SafeListener.Companion.listener
import com.lambda.event.listener.SafeListener.Companion.listen
import com.lambda.util.collections.LimitedOrderedSet
import net.minecraft.network.packet.c2s.query.QueryPingC2SPacket
import net.minecraft.network.packet.s2c.query.PingResultS2CPacket
Expand All @@ -33,12 +33,12 @@ object PingManager : Loadable {
get() = pings.lastOrNull() ?: 0

init {
listener<TickEvent.Pre> {
listen<TickEvent.Pre> {
connection.sendPacket(QueryPingC2SPacket(Util.getMeasuringTimeMs()))
}

listener<PacketEvent.Receive.Pre> { event ->
if (event.packet !is PingResultS2CPacket) return@listener
listen<PacketEvent.Receive.Pre> { event ->
if (event.packet !is PingResultS2CPacket) return@listen

pings.add(Util.getMeasuringTimeMs() - event.packet.startTime)
}
Expand Down
2 changes: 1 addition & 1 deletion common/src/main/kotlin/com/lambda/event/EventFlow.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ object EventFlow {
* the oldest event will be dropped to accommodate a new event.
*/
val concurrentFlow = MutableSharedFlow<Event>(
extraBufferCapacity = 1000,
extraBufferCapacity = 10000,
onBufferOverflow = BufferOverflow.DROP_OLDEST
)

Expand Down
113 changes: 59 additions & 54 deletions common/src/main/kotlin/com/lambda/event/listener/SafeListener.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@ import com.lambda.threading.runConcurrent
import com.lambda.threading.runSafe
import com.lambda.util.Pointer
import com.lambda.util.selfReference
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.Dispatchers
import kotlin.properties.ReadOnlyProperty
import kotlin.properties.ReadWriteProperty
import kotlin.reflect.KProperty
Expand Down Expand Up @@ -99,11 +101,11 @@ class SafeListener<T : Event>(
*
* Usage:
* ```kotlin
* listener<MyEvent> { event ->
* listen<MyEvent> { event ->
* player.sendMessage("Event received: $event")
* }
*
* listener<MyEvent>(priority = 1) { event ->
* listen<MyEvent>(priority = 1) { event ->
* player.sendMessage("Event received before the previous listener: $event")
* }
* ```
Expand All @@ -114,18 +116,65 @@ class SafeListener<T : Event>(
* @param function The function to be executed when the event is posted. This function should take a SafeContext and an event of type T as parameters.
* @return The newly created and registered [SafeListener].
*/
inline fun <reified T : Event> Any.listener(
inline fun <reified T : Event> Any.listen(
priority: Int = 0,
alwaysListen: Boolean = false,
noinline function: SafeContext.(T) -> Unit = {},
): SafeListener<T> {
val listener = SafeListener<T>(priority, this, alwaysListen) { event -> function(event) }
val listener = SafeListener<T>(priority, this, alwaysListen) { event ->
function(event)
}

EventFlow.syncListeners.subscribe(listener)

return listener
}

/**
* Registers a new [SafeListener] for a generic [Event] type [T] within the context of a [Task].
* The [function] is executed on the same thread where the [Event] was dispatched.
* The [function] will only be executed when the context satisfies certain safety conditions.
* These conditions are met when none of the following [SafeContext] properties are null:
* - [SafeContext.world]
* - [SafeContext.player]
* - [SafeContext.interaction]
* - [SafeContext.connection]
*
* Usage:
* ```kotlin
* myTask.listen<MyEvent> { event ->
* player.sendMessage("Event received: $event")
* }
*
* myTask.listen<MyEvent>(priority = 1) { event ->
* player.sendMessage("Event received before the previous listener: $event")
* }
* ```
*
* @param T The type of the event to listen for.
* This should be a subclass of Event.
* @param priority The priority of the listener.
* Listeners with higher priority will be executed first.
* The Default value is 0.
* @param alwaysListen If true, the listener will be executed even if it is muted. The Default value is false.
* @param function The function to be executed when the event is posted.
* This function should take a SafeContext and an event of type T as parameters.
* @return The newly created and registered [SafeListener].
*/
inline fun <reified T : Event> Task<*>.listen(
priority: Int = 0,
alwaysListen: Boolean = false,
noinline function: SafeContext.(T) -> Unit = {},
): SafeListener<T> {
val listener = SafeListener<T>(priority, this, alwaysListen) { event ->
function(event) // ToDo: run function always on game thread
}

syncListeners.subscribe<T>(listener)

return listener
}

/**
* This function registers a new [SafeListener] for a generic [Event] type [T].
* The [transform] is executed on the same thread where the [Event] was dispatched.
Expand All @@ -142,7 +191,7 @@ class SafeListener<T : Event>(
*
* Usage:
* ```kotlin
* private val event by listenNext<MyEvent> { event ->
* private val event by listenOnce<MyEvent> { event ->
* player.sendMessage("Event received only once: $event")
* // event is stored in the value
* // event is unsubscribed after execution
Expand Down Expand Up @@ -181,51 +230,6 @@ class SafeListener<T : Event>(
return pointer
}

/**
* Registers a new [SafeListener] for a generic [Event] type [T] within the context of a [Task].
* The [function] is executed on the same thread where the [Event] was dispatched.
* The [function] will only be executed when the context satisfies certain safety conditions.
* These conditions are met when none of the following [SafeContext] properties are null:
* - [SafeContext.world]
* - [SafeContext.player]
* - [SafeContext.interaction]
* - [SafeContext.connection]
*
* Usage:
* ```kotlin
* myTask.listener<MyEvent> { event ->
* player.sendMessage("Event received: $event")
* }
*
* myTask.listener<MyEvent>(priority = 1) { event ->
* player.sendMessage("Event received before the previous listener: $event")
* }
* ```
*
* @param T The type of the event to listen for.
* This should be a subclass of Event.
* @param priority The priority of the listener.
* Listeners with higher priority will be executed first.
* The Default value is 0.
* @param alwaysListen If true, the listener will be executed even if it is muted. The Default value is false.
* @param function The function to be executed when the event is posted.
* This function should take a SafeContext and an event of type T as parameters.
* @return The newly created and registered [SafeListener].
*/
inline fun <reified T : Event> Task<*>.listener(
priority: Int = 0,
alwaysListen: Boolean = false,
noinline function: SafeContext.(T) -> Unit = {},
): SafeListener<T> {
val listener = SafeListener<T>(priority, this, alwaysListen) { event ->
function(event) // ToDo: run function always on game thread
}

syncListeners.subscribe<T>(listener)

return listener
}

/**
* Registers a new [SafeListener] for a generic [Event] type [T].
* The [function] is executed on a new thread running asynchronously to the game thread.
Expand All @@ -236,12 +240,12 @@ class SafeListener<T : Event>(
*
* Usage:
* ```kotlin
* concurrentListener<MyEvent> { event ->
* listenConcurrently<MyEvent> { event ->
* println("Concurrent event received: $event")
* // no safe access to player or world
* }
*
* concurrentListener<MyEvent>(priority = 1) { event ->
* listenConcurrently<MyEvent>(priority = 1) { event ->
* println("Concurrent event received before the previous listener: $event")
* }
* ```
Expand All @@ -251,13 +255,14 @@ class SafeListener<T : Event>(
* @param function The function to be executed when the event is posted. This function should take a SafeContext and an event of type T as parameters.
* @return The newly created and registered [SafeListener].
*/
inline fun <reified T : Event> Any.concurrentListener(
inline fun <reified T : Event> Any.listenConcurrently(
priority: Int = 0,
alwaysListen: Boolean = false,
scheduler: CoroutineDispatcher = Dispatchers.Default,
noinline function: suspend SafeContext.(T) -> Unit = {},
): SafeListener<T> {
val listener = SafeListener<T>(priority, this, alwaysListen) { event ->
runConcurrent {
runConcurrent(scheduler) {
function(event)
}
}
Expand Down
26 changes: 14 additions & 12 deletions common/src/main/kotlin/com/lambda/event/listener/UnsafeListener.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,16 +81,16 @@ class UnsafeListener<T : Event>(
* The [function] is executed on the same thread where the [Event] was dispatched.
* The execution of the [function] is independent of the safety conditions of the context.
* Use this function when you need to listen to an [Event] in a context that is not in-game.
* For only in-game related contexts, use the [SafeListener.listener] function instead.
* For only in-game related contexts, use the [SafeListener.listen] function instead.
*
* Usage:
* ```kotlin
* unsafeListener<MyEvent> { event ->
* listenUnsafe<MyEvent> { event ->
* println("Unsafe event received: $event")
* // no safe access to player or world
* }
*
* unsafeListener<MyEvent>(priority = 1) { event ->
* listenUnsafe<MyEvent>(priority = 1) { event ->
* println("Unsafe event received before the previous listener: $event")
* }
* ```
Expand All @@ -101,7 +101,7 @@ class UnsafeListener<T : Event>(
* @param function The function to be executed when the event is posted. This function should take an event of type T as a parameter.
* @return The newly created and registered [UnsafeListener].
*/
inline fun <reified T : Event> Any.unsafeListener(
inline fun <reified T : Event> Any.listenUnsafe(
priority: Int = 0,
alwaysListen: Boolean = false,
noinline function: (T) -> Unit = {},
Expand All @@ -123,7 +123,7 @@ class UnsafeListener<T : Event>(
*
* Usage:
* ```kotlin
* private val event by unsafeListenOnce<MyEvent> { event ->
* private val event by listenOnceUnsafe<MyEvent> { event ->
* println("Unsafe event received only once: $event")
* // no safe access to player or world
* // event is stored in the value
Expand All @@ -139,7 +139,7 @@ class UnsafeListener<T : Event>(
* @param transform The function used to transform the event into a value.
* @return The newly created and registered [UnsafeListener].
*/
inline fun <reified T : Event, reified E> Any.unsafeListenOnce(
inline fun <reified T : Event, reified E> Any.listenOnceUnsafe(
priority: Int = 0,
alwaysListen: Boolean = false,
noinline transform: (T) -> E? = { null },
Expand All @@ -148,7 +148,7 @@ class UnsafeListener<T : Event>(
val pointer = Pointer<E>()

val destroyable by selfReference<UnsafeListener<T>> {
UnsafeListener(priority, this@unsafeListenOnce, alwaysListen) { event ->
UnsafeListener(priority, this@listenOnceUnsafe, alwaysListen) { event ->
pointer.value = transform(event)

if (predicate(event) &&
Expand All @@ -169,19 +169,19 @@ class UnsafeListener<T : Event>(
* Registers a new [UnsafeListener] for a generic [Event] type [T].
* The [function] is executed on a new thread running asynchronously to the game thread.
* This function should only be used when the [function] performs read actions on the game data.
* For only in-game related contexts, use the [SafeListener.concurrentListener] function instead.
* For only in-game related contexts, use the [SafeListener.listenConcurrently] function instead.
*
* Caution: Using this function to write to the game data can lead to race conditions. Therefore, it is recommended
* to use this function only for read operations to avoid potential concurrency issues.
*
* Usage:
* ```kotlin
* concurrentListener<MyEvent> { event ->
* listenUnsafeConcurrently<MyEvent> { event ->
* println("Concurrent event received: $event")
* // no safe access to player or world
* }
*
* concurrentListener<MyEvent>(priority = 1) { event ->
* listenUnsafeConcurrently<MyEvent>(priority = 1) { event ->
* println("Concurrent event received before the previous listener: $event")
* }
* ```
Expand All @@ -191,12 +191,14 @@ class UnsafeListener<T : Event>(
* @param function The function to be executed when the event is posted. This function should take a SafeContext and an event of type T as parameters.
* @return The newly created and registered [UnsafeListener].
*/
inline fun <reified T : Event> Any.unsafeConcurrentListener(
inline fun <reified T : Event> Any.listenUnsafeConcurrently(
priority: Int = 0,
alwaysListen: Boolean = false,
noinline function: (T) -> Unit = {},
): UnsafeListener<T> {
val listener = UnsafeListener<T>(priority, this, alwaysListen) { event -> function(event) }
val listener = UnsafeListener<T>(priority, this, alwaysListen) { event ->
function(event)
}

EventFlow.concurrentListeners.subscribe<T>(listener)

Expand Down
4 changes: 2 additions & 2 deletions common/src/main/kotlin/com/lambda/graphics/RenderMain.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import com.lambda.Lambda.mc
import com.lambda.event.EventFlow.post
import com.lambda.event.events.RenderEvent
import com.lambda.event.events.TickEvent
import com.lambda.event.listener.SafeListener.Companion.listener
import com.lambda.event.listener.SafeListener.Companion.listen
import com.lambda.graphics.animation.Animation.Companion.exp
import com.lambda.graphics.animation.AnimationTicker
import com.lambda.graphics.buffer.FrameBuffer
Expand All @@ -46,7 +46,7 @@ object RenderMain {
private val showHud get() = mc.currentScreen == null || LambdaHudGui.isOpen

private val hudAnimation0 = with(AnimationTicker()) {
listener<TickEvent.Pre> {
listen<TickEvent.Pre> {
tick()
}

Expand Down
Loading
Loading