An Observer for Events, simplifying the pattern of checking if the Event's content has already been handled.

EventObserver.kt

/**
 * An [Observer] for [Event]s, simplifying the pattern of checking if the [Event]'s content has
 * already been handled.
 *
 * [onEventUnhandledContent] is *only* called if the [Event]'s contents has not been handled.
 */
class EventObserver<T>(private val onEventUnhandledContent: (T) -> Unit) : Observer<Event<T>> {
    override fun onChanged(event: Event<T>?) {
        event?.getContentIfNotHandled()?.let { value ->
            onEventUnhandledContent(value)
        }
    }
}

I've implemented this concept a bit neater with thread-safety in Kotlin. It might be helpful.

import java.util.concurrent.atomic.AtomicBoolean

/**
 * @author aminography
 */
class OneTimeEvent<T>(
    private val value: T
) {

    private val isConsumed = AtomicBoolean(false)

    internal fun getValue(): T? =
        if (isConsumed.compareAndSet(false, true)) value
        else null
}

fun <T> T.toOneTimeEvent() =
    OneTimeEvent(this)

fun <T> OneTimeEvent<T>.consume(block: (T) -> Unit): T? =
    getValue()?.also(block)

First, when you want to post a value on LiveData, use toOneTimeEvent() extension function to wrap it with the OneTimeEvent:

yourLiveData.postValue(yourObject.toOneTimeEvent())

Second, when you are observing on the LiveData, use consume { } function on the delivered value to gain the feature of OneTimeEvent. You'll be sure that the block of consume { } will be executed only once, among all Observers.

yourLiveData.observe(this, Observer {
            it.consume { yourObject->
                // TODO: do whatever with 'yourObject'
            }
        })

crossinline

This doesn't seem readable. Certainly not to beginners.

There is nothing wrong with crossinline, although I must admit it was one of those things I've asked like 4 people for a month before I understood what it does, having seen code written by Jake Wharton where he uses it to wrap a multi-method abstract class to receive optional implementations through the constructor, and invoking said lambdas from overridden methods of an inline anonymous object implementation.

Just to actually mention an approach that might help resolve some issues, see https://medium.com/@Zhuinden/simplifying-jetpack-navigation-between-top-level-destinations-using-dagger-hilt-3d918721d91e

Thanks @aminography, very neat implementation, it's maybe not for total beginners but it's no rockets science for who's has some knowledge of thread safety... and the boilerplate is quite reduced.
I've a question though, I'm not an expert in Kotlin, why did you implement the following as an extension function and not directly as pasrt of the OneTimeEvent class?

fun <T> OneTimeEvent<T>.consume(block: (T) -> Unit): T? =
    getValue()?.also(block)

Thanks @aminography, very neat implementation, it's maybe not for total beginners but it's no rockets science for who's has some knowledge of thread safety... and the boilerplate is quite reduced.
I've a question though, I'm not an expert in Kotlin, why did you implement the following as an extension function and not directly as pasrt of the OneTimeEvent class?

fun <T> OneTimeEvent<T>.consume(block: (T) -> Unit): T? =
    getValue()?.also(block)

You're welcome :)
I'd written the extension functions in a separate file before I wrote them here. So, you can put the consume into the OneTimeEvent class, no problem.

For those working with Kotlin Flow I have created an extension function, onEachEvent, using the Event wrapper above. You may find the full explanation and code here.

OnEachEvent.kt

/**
 * Returns a flow which performs the given [action] on each value of the original flow's [Event].
 */
public fun <T> Flow<Event<T?>>.onEachEvent(action: suspend (T) -> Unit): Flow<T> = transform { value ->
    value.getContentIfNotHandled()?.let {
        action(it)
        return@transform emit(it)
    }
}

I think if you're using a Flow that is not a StateFlow, you don't need Event wrappers at all, as your events are not replayed on subscription anyway.

@Zhuinden, In the use case in Coinverse a MutableStateFlow is used in the Fragment in order to emit intents in the Model-View-Intent pattern. In the ViewModel, the intent is observed as a Flow which is non-mutable so that the event is unidirectional.

I observed when triggering the selectContent intent that opens content in the feed, and navigating to a different screen, that upon returning, the intent is emitting a second time reopening the past content selected. After refactoring with the Event pattern this has resolved the issue.

This is good to know for Flow.

as your events are not replayed on subscription (with Flow)

When implementing MutableStateFlow to update intents, the Event pattern has utility as Roman Elizarov also summarizes.

If you use StateFlow to deliver events to your views (as opposed to the delivering a state), you must use an event wrapper class just as suggested by the article you've linked to: https://medium.com/androiddevelopers/livedata-with-snackbar-navigation-and-other-events-the-singleliveevent-case-ac2622673150
With the event wrapper you can ensure that every event is handled at most once.

The full code is under the refactor/mvi branch which will be merged to root.

FeedView.kt

@ExperimentalCoroutinesApi
interface FeedView {
    /**
     * Intent to select content from the feed
     *
     * @return A flow that emits the content to select from the feed
     */
    fun selectContent(): Flow<Event<SelectContent?>>
}

FeedFragment.kt

// MutableStateFlow passed into Adapter
val selectContent: MutableStateFlow<Event<SelectContent?>> = MutableStateFlow(Event(null)),

FeedAdapter.kt

private fun createOnClickListener(content: Content, position: Int) = OnClickListener { view ->
        when (view.id) {
            preview, contentTypeLogo -> intent.selectContent.value =
                    Event(SelectContent(content, position))
        }
    }

FeedViewModel.kt

fun bindIntents(view: FeedView) {
    view.selectContent().onEachEvent { selectContent ->
        selectContent(selectContent)
    }.launchIn(coroutineScope)
}

private fun selectContent(selectContent: SelectContent) {
    when (selectContent.content.contentType) {
        ARTICLE -> repository.getAudiocast(selectContent).onEach { resource ->
            when (resource.status) {
                LOADING -> state.value = OpenContent(...)
                SUCCESS -> state.value = OpenContent(...)
                Status.ERROR -> state.value = OpenContent(...)
            }
        }.launchIn(coroutineScope)
        YOUTUBE -> state.value = OpenContent(...)
        NONE -> throw IllegalArgumentException("contentType expected, contentType is 'NONE'")
    }
}

Because one-off events shouldn't even be in a MutableStateFlow.

I hear that a LinkedListChannel exposed as a Flow would have the same effect, but without the magic.

Kotlin/kotlinx.coroutines@345458b

Thank you for the recommendation! I implemented a LinkedListChannel to test it out and it works as expected to emit the one-time events.

The tradeoff is using the Channel introduces the need to launch the coroutine to utilize the Channel send and receive functions which creates a layer of nesting with either lifecycleScope.launch and viewModelScope.launch whereas with the Event, the value can be emitted with .value and returned with the extension function onEachEvent that tracks whether the event has been handled or not. This strategy also requires passing an additional argument into the Adapter, coroutineScope.

LinkedListChannel

FeedView.kt

@ExperimentalCoroutinesApi
interface FeedView {
    /**
     * Intent to select content from the feed
     *
     * @return A flow that emits the content to select from the feed
     */
     fun selectContent(): Channel<SelectContent?>
}

FeedFragment.kt

// LinkedListChannel passed into Adapter
val selectContent: Channel<SelectContent?> = Channel(UNLIMITED)

FeedAdapter.kt

private fun createOnClickListener(content: Content, position: Int) = OnClickListener { view ->
        when (view.id) {
            preview, contentTypeLogo ->
                coroutineScope.launch {
                    intent.selectContent.send(SelectContent(content, position))
                }
        }
    }

FeedViewModel.kt

fun bindIntents(view: FeedView) {
    viewModelScope.launch {
            view.selectContent().receive()?.let {
                selectContent(it)
            }
     }
}

Can't you just emit the event using offer() though? That way you don't need to create a CoroutineScope.

The SendChannel<T>.offer() documentation says this:

Immediately adds the specified [element] to this channel, if this doesn't violate its capacity restrictions,
and returns true. Otherwise, just returns false. This is a synchronous variant of [send] which backs off
in situations when send suspends.

and the LinkedListChannel, say this:

Channel with linked-list buffer of a unlimited capacity (limited only by available memory).
Sender to this channel never suspends and [offer] always returns true.

I've been using the event-emitter library from Zhuinden, but since I'm already using coroutines. I've been trying to implement the same behavior using only coroutines.

Thanks @JoseAlcerreca.

It's also possible to create an extension to MutableLiveData to directly post an Event:

fun <T> MutableLiveData<Event<T>>.postEvent(content: T) {
    postValue(Event(content))
}

And to replace usage with:

mutableLiveData.postValue(Event(data)) => mutableLiveData.postEvent(data)

Am I wrong or can't EventObserver handle null events? i.e. new Event<>(null). I'm not in Kotlin but when I got it right it?.getContentIfNotHandled()?.let(onEventUnhandledContent)

onEventUnhandledContent

is only called if event != null and the return value != null so with new Event<>(null)

onEventUnhandledContent

is never called?!

I've tweaked some code a bit to allow nulling the content. I'm new to kotlin so please correct me if there are any issues

open class Event<out T>(private val content: T) {
    var hasBeenHandled = false
        private set // Allow external read but not write

    fun handleContent() : T {
        hasBeenHandled = true
        return content
    }

    /**
     * Returns the content, even if it's already been handled.
     */
    fun peekContent(): T = content
}

class EventObserver<T>(private val onEventUnhandledContent: (T) -> Unit) : Observer<Event<T>> {
    override fun onChanged(event: Event<T>?) {
        event?.let {
            if (it.hasBeenHandled) return
            onEventUnhandledContent(it.handleContent())
        }
    }
}

I use https://github.com/Zhuinden/live-event now

I hear that a LinkedListChannel exposed as a Flow would have the same effect, but without the magic.

As you are probably aware it also needs a custom collector to work reliably.

I use https://github.com/Zhuinden/live-event now

This is fine solution too, supporting multiple observers and multiple queued events, but with a small catch: only one observer receives these queued events.

@gmk57 this is true. Theoretically it'd be possible to either expose setPaused from command-queue and/or provide a "minimumSubscriberCount" (working name) that would make it so that under 3 subscribers the EventEmitter is paused

inline fun <T> LiveData<Event<T>>.observeEvent(owner: LifecycleOwner, crossinline onEventUnhandledContent: (T) -> Unit) {
    observe(owner, Observer { it?.getContentIfNotHandled()?.let(onEventUnhandledContent) })
}

I finished it like this. It seems to work correctly. What's your opinion?

inline fun LiveData<Event>.observeEvent(owner: LifecycleOwner, crossinline onEventUnhandledContent: (T) -> Unit) {
observe(owner) { it?.getContentIfNotHandled()?.let(onEventUnhandledContent) }
}