laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-08 23:54:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Adds a try/catch to make sure this procedure unsubscribes from the req

Browse Source
This commit is contained in:
Vitor Pamplona
2026-03-26 11:10:45 -04:00
parent 3776f65179
commit 958a460ea9
1 changed files with 37 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/service/broadcast/BroadcastTracker.kt
+37 -34
View File
@@ -142,55 +142,58 @@ class BroadcastTracker {
}
}
client.subscribe(subscription)
try {
client.subscribe(subscription)
val finalBroadcast =
coroutineScope {
val resultCollector =
async {
val receivedRelays = mutableSetOf<NormalizedRelayUrl>()
var currentBroadcast = broadcast
val finalBroadcast =
coroutineScope {
val resultCollector =
async {
val receivedRelays = mutableSetOf<NormalizedRelayUrl>()
var currentBroadcast = broadcast
withTimeoutOrNull(TIMEOUT_SECONDS * 1000) {
while (receivedRelays.size < relays.size) {
val response = resultChannel.receive()
withTimeoutOrNull(TIMEOUT_SECONDS * 1000) {
while (receivedRelays.size < relays.size) {
val response = resultChannel.receive()
// Skip if already received (don't override success)
if (response.relay in receivedRelays) continue
// Skip if already received (don't override success)
if (response.relay in receivedRelays) continue
receivedRelays.add(response.relay)
currentBroadcast = currentBroadcast.withResult(response.relay, response.result)
receivedRelays.add(response.relay)
currentBroadcast = currentBroadcast.withResult(response.relay, response.result)
// Update active broadcasts with new progress
_activeBroadcasts.update { list ->
list.map { if (it.id == trackingId) currentBroadcast else it }.toImmutableList()
// Update active broadcasts with new progress
_activeBroadcasts.update { list ->
list.map { if (it.id == trackingId) currentBroadcast else it }.toImmutableList()
}
}
}
// Mark remaining relays as timeout
relays.filter { it !in receivedRelays }.forEach { relay ->
currentBroadcast = currentBroadcast.withResult(relay, RelayResult.Timeout)
}
currentBroadcast
}
// Mark remaining relays as timeout
relays.filter { it !in receivedRelays }.forEach { relay ->
currentBroadcast = currentBroadcast.withResult(relay, RelayResult.Timeout)
}
// Send after setting up listener
client.send(event, relays)
currentBroadcast
}
resultCollector.await()
}
// Send after setting up listener
client.send(event, relays)
resultChannel.close()
resultCollector.await()
// Remove from active, emit to completed
_activeBroadcasts.update { list ->
list.map { if (it.id == trackingId) finalBroadcast else it }.toImmutableList()
}
client.unsubscribe(subscription)
resultChannel.close()
// Remove from active, emit to completed
_activeBroadcasts.update { list ->
list.map { if (it.id == trackingId) finalBroadcast else it }.toImmutableList()
Log.d(TAG, "Broadcast $trackingId complete: ${finalBroadcast.successCount}/${finalBroadcast.totalRelays} success")
} finally {
client.unsubscribe(subscription)
}
Log.d(TAG, "Broadcast $trackingId complete: ${finalBroadcast.successCount}/${finalBroadcast.totalRelays} success")
}
/**