Keep Android live sources in reactive channel state

tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt

@@ -199,7 +199,9 @@ object NativeLib {
 
     /* Live media используют отдельную ограниченную очередь, минуя чатовый event bus. */
     fun streamOpen(id: Long, group: Long, source: Long = 0, width: Int = 0, height: Int = 0,
-                   rotation: Int = 0, audio: Boolean = true): Boolean = nativeStreamOpen(id, group, source, width, height, rotation, audio)
+                   rotation: Int = 0, audio: Boolean = true, epoch: Long = 0): Boolean =
+        nativeStreamOpen(id, group, source, width, height, rotation, audio, epoch)
+    fun streamWatch(group: Long): Boolean = nativeStreamWatch(group)
     fun streamFeed(id: Long, video: Boolean, pts: Long, key: Boolean, data: ByteArray): Int = nativeStreamFeed(id, video, pts, key, data)
     fun streamStop(id: Long) { nativeStreamStop(id) }
     fun streamPull(id: Long): ByteArray? = nativeStreamPull(id)
@@ -208,7 +210,9 @@ object NativeLib {
     fun streamOpusEncode(handle: Long, pcm: ShortArray): ByteArray? = nativeStreamOpusEncode(handle, pcm)
     fun streamOpusDecode(handle: Long, packet: ByteArray): ShortArray? = nativeStreamOpusDecode(handle, packet)
     fun streamOpusClose(handle: Long) { nativeStreamOpusClose(handle) }
-    private external fun nativeStreamOpen(id: Long, group: Long, source: Long, width: Int, height: Int, rotation: Int, audio: Boolean): Boolean
+    private external fun nativeStreamOpen(id: Long, group: Long, source: Long, width: Int, height: Int, rotation: Int,
+                                         audio: Boolean, epoch: Long): Boolean
+    private external fun nativeStreamWatch(group: Long): Boolean
     private external fun nativeStreamFeed(id: Long, video: Boolean, pts: Long, key: Boolean, data: ByteArray): Int
     private external fun nativeStreamStop(id: Long)
     private external fun nativeStreamPull(id: Long): ByteArray?

tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamSource.kt

@@ -0,0 +1,21 @@
+package com.utun.chat.data
+
+import org.json.JSONArray
+
+/** Источник идентифицируется участником и эпохой; IDs сохраняют все 64 бита. */
+data class StreamSource(val id: Long, val name: String, val epoch: Long, val self: Boolean) {
+    val displayName: String get() = name.ifEmpty { id.toULong().toString() }
+
+    companion object {
+        fun parse(json: String): List<StreamSource> {
+            val array = JSONArray(json)
+            return (0 until array.length()).map { i ->
+                val item = array.getJSONObject(i)
+                val id = item.getString("id").toULong().toLong()
+                val epoch = item.getString("epoch").toULong(16).toLong()
+                require(id != 0L && epoch != 0L) { "Zero stream identity" }
+                StreamSource(id, item.getString("name"), epoch, item.getInt("self") == 1)
+            }
+        }
+    }
+}

tools/chatgui-android/app/src/main/java/com/utun/chat/viewmodel/ChatViewModel.kt

@@ -220,6 +220,23 @@ class ChatViewModel : ViewModel() {
 
     private val _currentChannel = MutableStateFlow<Channel?>(null)
     val currentChannel: StateFlow<Channel?> = _currentChannel
+    private val _streamSources = MutableStateFlow<List<StreamSource>>(emptyList())
+    val streamSources: StateFlow<List<StreamSource>> = _streamSources
+    private val _streamSourcesFailed = MutableStateFlow(false)
+    val streamSourcesFailed: StateFlow<Boolean> = _streamSourcesFailed
+    private val _streamSelection = MutableStateFlow<StreamSource?>(null)
+    val streamSelection: StateFlow<StreamSource?> = _streamSelection
+
+    fun selectStream(source: StreamSource?) {
+        _streamSelection.value = source
+        LogManager.addLog("INFO", "stream", "selected channel=${_currentChannel.value?.id} source=${source?.id} epoch=${source?.epoch}")
+    }
+
+    private fun watchStreams() {
+        if (!NativeLib.isReady()) return
+        val group = if (_foreground.value) _currentChannel.value?.id?.toULongOrNull()?.toLong() ?: 0L else 0L
+        if (!NativeLib.streamWatch(group)) LogManager.addLog("WARN", "stream", "source watch rejected group=$group")
+    }
 
     private val _soundOn = MutableStateFlow(true)
     val soundOn: StateFlow<Boolean> = _soundOn
@@ -320,6 +337,7 @@ class ChatViewModel : ViewModel() {
 
     fun setForeground(fg: Boolean) {
         _foreground.value = fg
+        watchStreams()
         if (fg) {
             _isStuck.value = false
             if (dbReady && repo != null) {
@@ -533,8 +551,9 @@ class ChatViewModel : ViewModel() {
                 if (result == 0) refreshChannels()
             }
             4 -> { if (repo != null) { LogManager.addLog("INFO", "VM", "CHANNEL_UPDATED — refreshChannels"); refreshChannels() } else { LogManager.addLog("WARN", "VM", "CHANNEL_UPDATED skipped — repo is NULL") } }
-            12 -> LogManager.addLog("INFO", "VM", "service started")
+            12 -> { LogManager.addLog("INFO", "VM", "service started"); watchStreams() }
             13 -> {
+                _streamSources.value = emptyList(); _streamSourcesFailed.value = false; _streamSelection.value = null
                 for (id in submissions.keys.toList()) submissionResult(id, false)
                 _radioTalkState.value = RadioTalkState()
                 _radioTalking.value = null
@@ -641,6 +660,21 @@ class ChatViewModel : ViewModel() {
                 val cur = _currentChannel.value
                 if (cur != null) refreshMembers(cur.id)
             }
+            58 -> { /* STREAM_SOURCES: group:8BE, valid:1, JSON. Старый канал не меняет новую шапку. */
+                if (data == null || data.size < 11 || data[8].toInt() !in 0..1) {
+                    LogManager.addLog("ERROR", "stream", "invalid source snapshot bytes=${data?.size}"); return
+                }
+                val group = java.nio.ByteBuffer.wrap(data).order(java.nio.ByteOrder.BIG_ENDIAN).long
+                if (_currentChannel.value?.id?.toULongOrNull()?.toLong() != group || !_foreground.value) return
+                try {
+                    _streamSourcesFailed.value = data[8].toInt() == 0
+                    _streamSources.value = StreamSource.parse(String(data, 9, data.size - 9, Charsets.UTF_8))
+                    LogManager.addLog("INFO", "stream", "sources group=$group count=${_streamSources.value.size} valid=${!_streamSourcesFailed.value}")
+                } catch (e: Exception) {
+                    _streamSources.value = emptyList(); _streamSourcesFailed.value = true
+                    LogManager.addLog("ERROR", "stream", "source snapshot parse group=$group: ${e.message}")
+                }
+            }
             19 -> { if (repo != null) refreshLocalSockets() }  /* LOCAL_SOCKETS */
             23 -> { /* MEMBER_LIST: live-снимок дополняет расширенный JSON-список. */
                 if (data == null || data.size < 3) {
@@ -1215,6 +1249,7 @@ class ChatViewModel : ViewModel() {
     fun selectChannel(channel: Channel) {
         val r = repo ?: return
         if (_currentChannel.value?.id != channel.id) {
+            _streamSources.value = emptyList(); _streamSourcesFailed.value = false; _streamSelection.value = null
             _historyAnchor.value = ""
             _messageJump.value = null
             pendingLatestJump = null
@@ -1224,6 +1259,7 @@ class ChatViewModel : ViewModel() {
             memberDetailJob?.cancel(); memberDetailTarget = null; _selectedMember.value = null
         }
         _currentChannel.value = channel
+        watchStreams()
         _radioActive.value = (radioChannelId == channel.id)
         _radioTalking.value = if (_radioActive.value) _radioTalkState.value.speakers.firstOrNull() else null
         _soundOn.value = NativeLib.getSoundOnMessage(channel.id)
@@ -1783,6 +1819,7 @@ class ChatViewModel : ViewModel() {
         _channels.value = emptyList()
         _messages.value = emptyList()
         _currentChannel.value = null
+        _streamSources.value = emptyList(); _streamSelection.value = null; _streamSourcesFailed.value = false
         _generatingKeys.value = false
         _localSockets.value = emptyList()
         _channelOwner.value = false