feat: Entity getAsFlow (closes #108)

feat: Basic support for removing observers after adding

geary-core/src/commonMain/kotlin/com/mineinabyss/geary/datatypes/Entity.kt

@@ -4,14 +4,25 @@ import com.mineinabyss.geary.annotations.optin.DangerousComponentOperation
 import com.mineinabyss.geary.components.EntityName
 import com.mineinabyss.geary.datatypes.family.family
 import com.mineinabyss.geary.engine.Engine
-import com.mineinabyss.geary.helpers.*
+import com.mineinabyss.geary.helpers.NO_COMPONENT
+import com.mineinabyss.geary.helpers.component
+import com.mineinabyss.geary.helpers.componentId
+import com.mineinabyss.geary.helpers.componentIdWithNullable
 import com.mineinabyss.geary.modules.Geary
 import com.mineinabyss.geary.modules.relationOf
+import com.mineinabyss.geary.observers.entity.observe
+import com.mineinabyss.geary.observers.entity.removeObserver
 import com.mineinabyss.geary.observers.events.OnAdd
+import com.mineinabyss.geary.observers.events.OnEntityRemoved
+import com.mineinabyss.geary.observers.events.OnRemove
+import com.mineinabyss.geary.observers.events.OnSet
 import com.mineinabyss.geary.systems.accessors.AccessorOperations
 import com.mineinabyss.geary.systems.accessors.RelationWithData
-import org.koin.core.Koin
-import org.koin.core.KoinApplication
+import com.mineinabyss.geary.systems.query.query
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.Channel.Factory.CONFLATED
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
 import kotlin.reflect.KClass
 
 typealias GearyEntity = Entity
@@ -181,13 +192,44 @@ class Entity(val id: EntityId, val world: Geary) {
         requireSameWorldAs(base)
         world.write.extendFor(id, base.id)
     }
-    
+
     /** Removes a [prefab] from this entity.  */
     fun removePrefab(prefab: Entity) {
         requireSameWorldAs(prefab)
         remove(Relation.of(comp.instanceOf, prefab.id).id)
     }
 
+    /**
+     * Get a component as a [Flow], updates to the component will be emitted, including `null` when the component is removed.
+     *
+     * The flow stops when the entity is removed.
+     */
+    inline fun <reified T : Any> getAsFlow(): Flow<T?> = with(world) {
+        flow {
+            val updates = Channel<T?>(CONFLATED)
+            updates.trySend(get<T>())
+            val onSetObserver = observe<OnSet>().involving<T>().exec(query<T>()) { (comp) ->
+                updates.trySend(comp)
+            }
+            val onRemoveObserver = observe<OnRemove>().involving<T>().exec(query<T>()) { (comp) ->
+                updates.trySend(null)
+            }
+            val onEntityRemoved = observe<OnEntityRemoved>().exec {
+                updates.close()
+            }
+
+            try {
+                for (update in updates) {
+                    emit(update)
+                }
+            } finally {
+                removeObserver(onSetObserver)
+                removeObserver(onRemoveObserver)
+                removeObserver(onEntityRemoved)
+            }
+        }
+    }
+
     // Relations
 
     /** Gets the data stored under the relation of kind [K] and target [T]. */

geary-core/src/commonMain/kotlin/com/mineinabyss/geary/observers/EventToObserversMap.kt

@@ -10,11 +10,19 @@ class EventToObserversMap(
 ) {
     private val eventToObserverMap = LongSparseArray<ObserverList>()
 
+    val isEmpty get() = eventToObserverMap.isEmpty()
+
     fun addObserver(observer: Observer) {
         observer.listenToEvents.forEach { event ->
             eventToObserverMap.getOrPut(event.toLong()) { ObserverList(records) }.add(observer)
         }
     }
 
+    fun removeObserver(observer: Observer) {
+        observer.listenToEvents.forEach { event ->
+            eventToObserverMap[event.toLong()]?.remove(observer)
+        }
+    }
+
     operator fun get(event: ComponentId): ObserverList? = eventToObserverMap[event.toLong()]
 }

geary-core/src/commonMain/kotlin/com/mineinabyss/geary/observers/ObserverList.kt

@@ -23,6 +23,14 @@ class ObserverList(
         }
     }
 
+    fun remove(observer: Observer) {
+        if (observer.involvedComponents.size == 0) {
+            involved2Observer[0L]?.remove(observer)
+        } else observer.involvedComponents.forEach { componentId ->
+            involved2Observer[componentId.toLong()]?.remove(observer)
+        }
+    }
+
     inline fun forEach(involvedComp: ComponentId, entity: EntityId, exec: (Observer, Archetype, row: Int) -> Unit) {
         involved2Observer[0L]?.forEach {
             records.runOn(entity) { archetype, row -> exec(it, archetype, row) }

geary-core/src/commonMain/kotlin/com/mineinabyss/geary/observers/entity/EntityObserver.kt

@@ -5,7 +5,6 @@ import com.mineinabyss.geary.datatypes.ComponentId
 import com.mineinabyss.geary.datatypes.GearyEntity
 import com.mineinabyss.geary.helpers.componentId
 import com.mineinabyss.geary.helpers.entity
-import com.mineinabyss.geary.modules.ArchetypeEngineModule
 import com.mineinabyss.geary.observers.EventToObserversMap
 import com.mineinabyss.geary.observers.Observer
 import com.mineinabyss.geary.observers.builders.*
@@ -18,6 +17,9 @@ inline fun <reified T : Any> GearyEntity.observeWithData(): ObserverEventsBuilde
     return observeWithData(world.componentId<T>())
 }
 
+/**
+ * Attaches an observer to fire on events emitted on this entity and its instances.
+ */
 fun GearyEntity.attachObserver(observer: Observer) {
     val observerEntity = world.entity {
         // TODO avoid cast
@@ -29,6 +31,20 @@ fun GearyEntity.attachObserver(observer: Observer) {
     addRelation<Observer>(observerEntity)
 }
 
+/**
+ * Removes an entity observer that was previously attached via [attachObserver]
+ */
+fun GearyEntity.removeObserver(observer: Observer) = with(world) {
+    getRelations<Observer, Any?>().forEach {
+        val observerEntity = it.target.toGeary()
+        val map = observerEntity.get<EventToObserversMap>()
+        if (map != null) {
+            map.removeObserver(observer)
+            if (map.isEmpty) removeRelation<Observer>(observerEntity)
+        }
+    }
+}
+
 fun GearyEntity.observe(vararg events: ComponentId): ObserverEventsBuilder<ObserverContext> {
     return ObserverWithoutData(events.toList(), world, ::attachObserver)
 }

geary-core/src/jvmTest/kotlin/com/mineinabyss/geary/observers/EntityGetAsFlowTest.kt

@@ -0,0 +1,54 @@
+package com.mineinabyss.geary.observers
+
+import com.mineinabyss.geary.helpers.entity
+import com.mineinabyss.geary.test.GearyTest
+import io.kotest.matchers.collections.shouldContainExactly
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+
+class EntityGetAsFlowTest : GearyTest() {
+    @Test
+    fun `getAsFlow should correctly listen to entity updates`() = runTest {
+        val entity = entity()
+        val collected = mutableListOf<Int>()
+
+        backgroundScope.launch(UnconfinedTestDispatcher(testScheduler)) {
+            entity.getAsFlow<Int>().collect {
+                if(it != null) collected.add(it)
+                else collected.add(0)
+            }
+        }
+        entity.set(1)
+        entity.set("other component")
+        entity.set(2)
+        entity.remove<Int>()
+        entity.set(3)
+
+        collected shouldContainExactly listOf(0, 1, 2, 0, 3)
+    }
+
+    @Test
+    fun `getAsFlow should unregister itself when cancelled`()  = runTest {
+        val entity = entity()
+        val collected = mutableListOf<Int>()
+
+        backgroundScope.launch(UnconfinedTestDispatcher(testScheduler)) {
+            entity.getAsFlow<Int>().collect {
+                if(it != null) collected.add(it)
+                else collected.add(0)
+            }
+            val collecting = launch(UnconfinedTestDispatcher(testScheduler)) {
+                entity.getAsFlow<Int>().collect {
+                    if(it != null) collected.add(it)
+                    else collected.add(0)
+                }
+            }
+            entity.set(1)
+            collecting.cancel()
+            entity.set(2)
+            collected shouldContainExactly listOf(0, 1)
+        }
+    }
+}