Skip to content
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.

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

[ECO-5013] feat: Implement room lifecycle monitoring #94

Draft
wants to merge 5 commits into
base: main
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from 1 commit
Commits
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
132 changes: 132 additions & 0 deletions chat-android/src/main/java/com/ably/chat/Emitter.kt
Original file line number Diff line number Diff line change
@@ -0,0 +1,132 @@
package com.ably.chat

import java.util.TreeSet
import java.util.concurrent.LinkedBlockingQueue
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch

/**
* Kotlin Emitter interface for supplied value
* Spec: RTE1
*/
internal interface Emitter<V> {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don’t think we need this, as we’re essentially reinventing Flow here. Additionally, we’re missing some crucial points in the ScopedEmitter implementation. In its current form, listeners will be invoked non-sequentially, which can be misleading.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was hoping we could have a discussion about this eventually. I don't think we're trying to reinvent the wheel here, though the implementation might resemble shared or hot flows.

Unlike cold flows, hot flows continue emitting values even if no one is actively collecting them. This can lead to situations where the flow keeps emitting values even when the component is in the background, which could potentially cause memory leaks.

Additionally, the flow API is well-suited for use in coroutine-based functions, as both emit and collect are suspending functions. On the other hand, tryEmit can be used in non-suspending functions (like the channel.on method here), although it's generally not recommended.

Since tryEmit() is not a suspending function, it requires a buffer to store the emitted values until they are processed by subscribers. To make this work, we need to set the buffer size to unlimited, but this can affect performance and slow down other subscribers.

Another challenge is canceling flows, as there’s no guarantee that queued events will be delivered when cancellation occurs.

I'll create a separate thread to dive deeper into this and explore how the Emitter implementation can be leveraged for the existing public API to handle messages and presence events asynchronously on the Dispatchers.Default scope, as opposed to relying on the current blocking subscribers.

fun emit(value: V)
fun on(block: suspend CoroutineScope.(V) -> Unit): Subscription
fun once(block: suspend CoroutineScope.(V) -> Unit): Subscription
fun offAll()
}

/**
* ScopedEmitter is a thread-safe, non-blocking emitter implementation for Kotlin.
* It ensures that all subscribers receive events asynchronously in the same order under given scope.
*
* @param V The type of value to be emitted.
* @param subscriberScope The CoroutineScope in which the subscribers will run. Defaults to Dispatchers.Default.
* @param logger An optional logger for logging errors during event processing.
*/
internal class ScopedEmitter<V> (
private val subscriberScope: CoroutineScope = CoroutineScope(Dispatchers.Default),
private val logger: Logger? = null,
) : Emitter<V> {

// Sorted list of unique subscribers based on supplied block
private val subscribers = TreeSet<AsyncSubscriber<V>>()

// Emitter scope to make sure all subscribers receive events in same order.
// Will be automatically garbage collected once all jobs are performed.
private val sequentialScope = CoroutineScope(Dispatchers.Default.limitedParallelism(1))

val finishedProcessing: Boolean
get() = subscribers.all { it.values.isEmpty() && !it.isSubscriberRunning }

@get:Synchronized
val subscribersCount: Int
get() = subscribers.size

@Synchronized
override fun emit(value: V) {
for (subscriber in subscribers.toList()) {
subscriber.inform(value)
if (subscriber.once) {
off(subscriber)
}
}
}

private fun register(subscriber: AsyncSubscriber<V>): Subscription {
subscribers.add(subscriber)
return Subscription {
off(subscriber)
}
}

@Synchronized
override fun on(block: suspend CoroutineScope.(V) -> Unit): Subscription {
val subscriber = AsyncSubscriber(sequentialScope, subscriberScope, block, false, logger)
return register(subscriber)
}

@Synchronized
override fun once(block: suspend CoroutineScope.(V) -> Unit): Subscription {
val subscriber = AsyncSubscriber(sequentialScope, subscriberScope, block, true, logger)
return register(subscriber)
}

@Synchronized
override fun offAll() {
subscribers.clear()
}

@Synchronized
private fun off(subscriber: AsyncSubscriber<V>) {
subscribers.remove(subscriber)
}
}

private class AsyncSubscriber<V>(
private val emitterSequentialScope: CoroutineScope,
private val subscriberScope: CoroutineScope,
private val subscriberBlock: (suspend CoroutineScope.(V) -> Unit),
val once: Boolean,
private val logger: Logger? = null,
) : Comparable<V> {
val values = LinkedBlockingQueue<V>() // Accessed by both Emitter#emit and emitterSequentialScope
var isSubscriberRunning = false // Only accessed as a part of emitterSequentialScope

fun inform(value: V) {
values.add(value)
emitterSequentialScope.launch {
if (!isSubscriberRunning) {
isSubscriberRunning = true
while (values.isNotEmpty()) {
val valueTobeEmitted = values.poll()
safelyPublish(valueTobeEmitted as V) // Process sequentially, similar to core ably eventEmitter
}
isSubscriberRunning = false
}
}
}

private suspend fun safelyPublish(value: V) {
runCatching {
subscriberScope.launch {
try {
subscriberBlock(value)
} catch (t: Throwable) {
// Catching exception to avoid error propagation to parent
logger?.warn("Error processing value $value", t)
}
}.join()
}
}

override fun compareTo(other: V): Int {
// Avoid registering duplicate anonymous subscriber block with same instance id
// Common scenario when Android activity is refreshed or some app components refresh
if (other is AsyncSubscriber<*>) {
return this.subscriberBlock.hashCode().compareTo(other.subscriberBlock.hashCode())
}
return this.hashCode().compareTo(other.hashCode())
}
}
Loading
Loading