Add flow support for elements

playback/core/src/main/kotlin/element/ElementsContainer.kt

@@ -1,12 +1,22 @@
 package org.jellyfin.playback.core.element
 
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.map
 import java.util.concurrent.ConcurrentHashMap
 
 /**
  * Container to hold elements identified with an [ElementKey].
  */
 open class ElementsContainer {
 	private val elements = ConcurrentHashMap<ElementKey<*>, Any?>()
+	private val updateFlow = MutableSharedFlow<ElementKey<*>>(
+		replay = 1,
+		extraBufferCapacity = 1,
+		onBufferOverflow = BufferOverflow.DROP_OLDEST,
+	)
 
 	fun <T : Any> get(key: ElementKey<T>): T = getOrNull(key)
 		?: error("No element found for key $key.")
@@ -18,9 +28,17 @@ open class ElementsContainer {
 
 	fun <T : Any> put(key: ElementKey<T>, value: T) {
 		elements[key] = value
+		updateFlow.tryEmit(key)
 	}
 
 	fun <T : Any> remove(key: ElementKey<T>) {
 		elements.remove(key)
+		updateFlow.tryEmit(key)
+	}
+
+	fun <T : Any> getFlow(key: ElementKey<T>): Flow<T?> {
+		return updateFlow
+			.map { getOrNull(key) }
+			.distinctUntilChanged()
 	}
 }

playback/core/src/main/kotlin/element/delegates.kt

@@ -1,5 +1,7 @@
 package org.jellyfin.playback.core.element
 
+import kotlinx.coroutines.flow.Flow
+import kotlin.properties.ReadOnlyProperty
 import kotlin.properties.ReadWriteProperty
 import kotlin.reflect.KProperty
 
@@ -38,3 +40,10 @@ fun <T : Any> requiredElement(
 		thisRef.put(key, value)
 	}
 }
+
+/**
+ * Delegate for the flow of an element.
+ */
+fun <T : Any> elementFlow(
+	key: ElementKey<T>,
+) = ReadOnlyProperty<ElementsContainer, Flow<T?>> { thisRef, _ -> thisRef.getFlow(key) }

playback/core/src/main/kotlin/mediastream/QueueEntryMediaStream.kt

@@ -2,6 +2,7 @@ package org.jellyfin.playback.core.mediastream
 
 import org.jellyfin.playback.core.element.ElementKey
 import org.jellyfin.playback.core.element.element
+import org.jellyfin.playback.core.element.elementFlow
 import org.jellyfin.playback.core.queue.QueueEntry
 
 private val mediaStreamKey = ElementKey<PlayableMediaStream>("MediaStream")
@@ -10,3 +11,8 @@ private val mediaStreamKey = ElementKey<PlayableMediaStream>("MediaStream")
  * Get or set the [MediaStream] for this [QueueEntry].
  */
 var QueueEntry.mediaStream by element(mediaStreamKey)
+
+/**
+ * Get the [MediaStream] flow for this [QueueEntry].
+ */
+val QueueEntry.mediaStreamFlow by elementFlow(mediaStreamKey)

playback/core/src/main/kotlin/mediastream/normalizationGainElement.kt

@@ -2,6 +2,7 @@ package org.jellyfin.playback.core.mediastream
 
 import org.jellyfin.playback.core.element.ElementKey
 import org.jellyfin.playback.core.element.element
+import org.jellyfin.playback.core.element.elementFlow
 import org.jellyfin.playback.core.queue.QueueEntry
 
 private val normalizationGainKey = ElementKey<Float>("NormalizationGain")
@@ -11,3 +12,9 @@ private val normalizationGainKey = ElementKey<Float>("NormalizationGain")
  * apply a gain to the audio output. The normalization gain must target a loudness of -23LUFS.
  */
 var QueueEntry.normalizationGain by element(normalizationGainKey)
+
+/**
+ * Get the flow of [normalizationGain].
+ * @see normalizationGain
+ */
+val QueueEntry.normalizationGainFlow by elementFlow(normalizationGainKey)

playback/core/src/main/kotlin/queue/QueueEntryMetadata.kt

@@ -1,6 +1,7 @@
 package org.jellyfin.playback.core.queue
 
 import org.jellyfin.playback.core.element.ElementKey
+import org.jellyfin.playback.core.element.elementFlow
 import org.jellyfin.playback.core.element.requiredElement
 import java.time.LocalDate
 import kotlin.time.Duration
@@ -42,3 +43,9 @@ private val metadataKey = ElementKey<QueueEntryMetadata>("QueueEntryMetadata")
  * Get or set the [QueueEntryMetadata] for this [QueueEntry]. Defaults to [QueueEntryMetadata.Empty].
  */
 var QueueEntry.metadata by requiredElement(metadataKey) { QueueEntryMetadata.Empty }
+
+/**
+ * Get the flow of [metadata].
+ * @see metadata
+ */
+val QueueEntry.metadataFlow by elementFlow(metadataKey)

playback/jellyfin/src/main/kotlin/queue/baseItemElement.kt

@@ -2,6 +2,7 @@ package org.jellyfin.playback.jellyfin.queue
 
 import org.jellyfin.playback.core.element.ElementKey
 import org.jellyfin.playback.core.element.element
+import org.jellyfin.playback.core.element.elementFlow
 import org.jellyfin.playback.core.queue.QueueEntry
 import org.jellyfin.sdk.model.api.BaseItemDto
 
@@ -11,3 +12,8 @@ private val baseItemKey = ElementKey<BaseItemDto>("BaseItemDto")
  * Get or set the [BaseItemDto] for this [QueueEntry].
  */
 var QueueEntry.baseItem by element(baseItemKey)
+
+/**
+ * Get the [BaseItemDto] flow for this [QueueEntry].
+ */
+val QueueEntry.baseItemFlow by elementFlow(baseItemKey)

playback/jellyfin/src/main/kotlin/queue/mediaSourceIdElement.kt

@@ -2,6 +2,7 @@ package org.jellyfin.playback.jellyfin.queue
 
 import org.jellyfin.playback.core.element.ElementKey
 import org.jellyfin.playback.core.element.element
+import org.jellyfin.playback.core.element.elementFlow
 import org.jellyfin.playback.core.queue.QueueEntry
 
 private val mediaSourceIdKey = ElementKey<String>("MediaSource")
@@ -11,3 +12,9 @@ private val mediaSourceIdKey = ElementKey<String>("MediaSource")
  * behavior.
  */
 var QueueEntry.mediaSourceId by element(mediaSourceIdKey)
+
+/**
+ * Get the flow of [mediaSourceId].
+ * @see mediaSourceId
+ */
+val QueueEntry.mediaSourceIdFlow by elementFlow(mediaSourceIdKey)