Introduce SharedFlow and sharing operators

[elizarov]

Summary of changes:

• SharedFlow, MutableSharedFlow and its constructor.
• StateFlow implements SharedFlow.
• SharedFlow.onSubscription operator, clarified docs in other onXxx operators.
• BufferOverflow strategy in kotlinx.coroutines.channels package.
• shareIn and stateIn operators and SharingStarted strategies for them.
• SharedFlow.flowOn error lint (up from StateFlow).
• Precise cancellable() operator fusion.
• Precise distinctUntilChanged() operator fusion.
• StateFlow.compareAndSet function.
• asStateFlow and asSharedFlow read-only view functions.
• Consistently clarified docs on cold vs hot flows.
• Future deprecation notice for BroadcastChannel, ConflatedBroadcastChannel, broadcast, and broadcastIn.
• Channel(...) constructor function has onBufferOverflow parameter.
• buffer(...) operator has onBufferOverflow parameter.
• shareIn/stateIn buffer and overflow strategy are configured via upstream buffer operators.
• shareIn/stateIn fuse with upstream flowOn for more efficient execution.
• conflate() is implemented as buffer(onBufferOverflow=KEEP_LATEST), non-suspending strategies are reasonably supported with 0 and default capacities.
• Added reactive operator migration hints.
• WhileSubscribed with kotlin.time.Duration params

Fixes #2034
Fixes #2047

[elizarov]

I just noticed this sentence for the first time. Do I understand correctly that this means that if the buffer overflow mode is DROP_OLDEST or SUSPEND, it will effectively act like be DROP_LATEST until the first subscriber subscribes?

Yes, just can think of it this way. The correct mental model, though, is that until the first subscriber subscribers there's always an internal very-fast subscriber that consumes all subscribed messages immediately (so that emit never suspends) so that all the other user-created subscriptions just see replay most recently emitted values.

[elizarov]

This PR is now complete, squashed, and rebased onto the develop branch.

[zach-klippenstein]

The correct mental model, though, is that until the first subscriber subscribers there's always an internal very-fast subscriber that consumes all subscribed messages immediately (so that emit never suspends) so that all the other user-created subscriptions just see replay most recently emitted values.

That's clear and makes sense for the SUSPEND case, but I'm more confused about DROP_OLDEST now. If the overflow mode is DROP_OLDEST and more than replay items are emitted before the first subscriber subscribes, are the oldest ones or the latest ones dropped?

[zach-klippenstein]

What does "supported" mean here? If replay == 0, is this value just ignored? DROP_LATEST and DROP_OLDEST seem like they would both mean emitters would never suspend but otherwise be equivalent, but SUSPEND could still suspend emitters.

[elizarov]

It means that IllegalArgumentException is thrown if you try to do replay = 0 and onBufferOverflow != SUSPEND

[zach-klippenstein]

I see the general "This function throws [IllegalArgumentException] on unsupported values of parameters of combinations thereof" warning at above, but I think it would be more clear to explicitly put your reply to my comment in the docs. It's a little more verbose, but given the complexity of the interactions of the various values of these parameters I think it would be worth it.

[elizarov]

That's clear and makes sense for the SUSPEND case, but I'm more confused about DROP_OLDEST now. If the overflow mode is DROP_OLDEST and more than replay items are emitted before the first subscriber subscribes, are the oldest ones or the latest ones dropped?

onBufferOverflow happens only when there is a buffer. There is a buffer only when there are slow subscribers that cannot keep up. No slow subscribers, no buffer, no overflow, does not matter what overflow strategy is set at all.

[1zaman]

While looking through the documentation, I came across quite a few minor grammatical errors. If you like I can point them out here, but maybe it can benefit from some proofreading.

[elizarov]

While looking through the documentation, I came across quite a few minor grammatical errors. If you like I can point them out here, but maybe it can benefit from some proofreading.

You can directly point to places in the code via GitHub interface (and "Files Changed" tab) and even provide suggested replacements.

[1zaman]

Here is the grammar review of the documentation and comments. Most issues are quite minor, but there are a lot of them. Many are probably subjective though, so feel free to change or ignore as seems appropriate.

I have provided suggested replacements, but did not make any adjustments for line wrapping, which might also need to be accounted for when implementing the changes.

[p-lr]

Any chance to have this reviewed by @qwwdfsad and merged before Kotlin 1.4 is released? ;)
I'm super excited by those changes.

[qwwdfsad]

👍

[qwwdfsad]

oldState is volatile and can be read without a lock.
Its value can be changed during a comparison, but it's inseparable from an arbitrary interleaving of two updates.
Tho oldValue still has to be re-read under the lock, so maybe it's not worth it

[qwwdfsad]

Last minute change: let's mark ConflatedBroacastChannel and BroadcastChannel as obsolete to completely deprecate them in the future

[prithivraj]

This is amazing!
Will this be part of 1.4.0?
When can we start using it?

[LouisCAD]

@prithivraj Look at the label, that will answer your first question.
Regarding when… I think time will tell. Also, there'll be a "Coroutines Update" talk later today here: https://kotlinlang.org/lp/event-14/, you might be interested.