Introduce SubclassOptInRequired to the codebase

The initial stage of #3770

Marked:
* Job,
* Deferred and CompletableDeferred,
* Flow, SharedFlow, StateFlow,
* CloseableCoroutineDispatcher,
* CancellableContinuation,
* All of their `public` implementors.

kotlinx-coroutines-core/common/src/AbstractCoroutine.kt

@@ -30,6 +30,7 @@ import kotlin.coroutines.*
  *
  * @suppress **This an internal API and should not be used from general code.**
  */
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 @InternalCoroutinesApi
 public abstract class AbstractCoroutine<in T>(
     parentContext: CoroutineContext,

kotlinx-coroutines-core/common/src/Annotations.kt

@@ -90,3 +90,15 @@ public annotation class ObsoleteCoroutinesApi
             "so stable API could be provided instead"
 )
 public annotation class InternalCoroutinesApi
+
+/**
+ * Marks declarations that can be safely used, but not inherited.
+ */
+@Target(AnnotationTarget.CLASS)
+@RequiresOptIn(
+    level = RequiresOptIn.Level.WARNING, message =
+    "This is a kotlinx.coroutines API that is not intended to be inherited from. " +
+        "Either new methods may be added in the future, which would break the inheritance, " +
+        "or correctly inheriting from it requires fulfilling contracts that may change in the future."
+)
+public annotation class BrittleForInheritanceCoroutinesApi

kotlinx-coroutines-core/common/src/Builders.common.kt

@@ -88,6 +88,7 @@ public fun <T> CoroutineScope.async(
     return coroutine
 }
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 @Suppress("UNCHECKED_CAST")
 private open class DeferredCoroutine<T>(
     parentContext: CoroutineContext,
@@ -182,6 +183,7 @@ public suspend inline operator fun <T> CoroutineDispatcher.invoke(
 
 // --------------- implementation ---------------
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private open class StandaloneCoroutine(
     parentContext: CoroutineContext,
     active: Boolean

kotlinx-coroutines-core/common/src/CancellableContinuation.kt

@@ -41,6 +41,7 @@ import kotlin.coroutines.intrinsics.*
  *    +-----------+
  * ```
  */
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 public interface CancellableContinuation<in T> : Continuation<T> {
     /**
      * Returns `true` when this continuation is active -- it has not completed or cancelled yet.

kotlinx-coroutines-core/common/src/CancellableContinuationImpl.kt

@@ -25,6 +25,7 @@ internal val RESUME_TOKEN = Symbol("RESUME_TOKEN")
 /**
  * @suppress **This is unstable API and it is subject to change.**
  */
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 @PublishedApi
 internal open class CancellableContinuationImpl<in T>(
     final override val delegate: Continuation<T>,

kotlinx-coroutines-core/common/src/CloseableCoroutineDispatcher.kt

@@ -6,11 +6,9 @@ package kotlinx.coroutines
  * associated with the current dispatcher.
  * Examples of closeable dispatchers are dispatchers backed by `java.lang.Executor` and
  * by `kotlin.native.Worker`.
- *
- * **The `CloseableCoroutineDispatcher` class is not stable for inheritance in 3rd party libraries**, as new methods
- * might be added to this interface in the future, but is stable for use.
  */
 @ExperimentalCoroutinesApi
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 public expect abstract class CloseableCoroutineDispatcher() : CoroutineDispatcher {
 
     /**

kotlinx-coroutines-core/common/src/CompletableDeferred.kt

@@ -15,10 +15,8 @@ import kotlinx.coroutines.selects.*
  *
  * All functions on this interface are **thread-safe** and can
  * be safely invoked from concurrent coroutines without external synchronization.
- *
- * **The `CompletableDeferred` interface is not stable for inheritance in 3rd party libraries**,
- * as new methods might be added to this interface in the future, but is stable for use.
  */
+@SubclassOptInRequired(markerClass = BrittleForInheritanceCoroutinesApi::class)
 public interface CompletableDeferred<T> : Deferred<T> {
     /**
      * Completes this deferred value with a given [value]. The result is `true` if this deferred was
@@ -73,6 +71,7 @@ public fun <T> CompletableDeferred(value: T): CompletableDeferred<T> = Completab
 /**
  * Concrete implementation of [CompletableDeferred].
  */
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 @Suppress("UNCHECKED_CAST")
 private class CompletableDeferredImpl<T>(
     parent: Job?

kotlinx-coroutines-core/common/src/CompletableJob.kt

@@ -10,6 +10,7 @@ package kotlinx.coroutines
  * **The `CompletableJob` interface is not stable for inheritance in 3rd party libraries**,
  * as new methods might be added to this interface in the future, but is stable for use.
  */
+@SubclassOptInRequired(markerClass = BrittleForInheritanceCoroutinesApi::class)
 public interface CompletableJob : Job {
     /**
      * Completes this job. The result is `true` if this job was completed as a result of this invocation and

kotlinx-coroutines-core/common/src/Deferred.kt

@@ -26,10 +26,8 @@ import kotlinx.coroutines.selects.*
  *
  * All functions on this interface and on all interfaces derived from it are **thread-safe** and can
  * be safely invoked from concurrent coroutines without external synchronization.
- *
- * **`Deferred` interface and all its derived interfaces are not stable for inheritance in 3rd party libraries**,
- * as new methods might be added to this interface in the future, but is stable for use.
  */
+@SubclassOptInRequired(markerClass = BrittleForInheritanceCoroutinesApi::class)
 public interface Deferred<out T> : Job {
 
     /**

kotlinx-coroutines-core/common/src/Job.kt

@@ -99,12 +99,8 @@ import kotlin.jvm.*
  *
  * All functions on this interface and on all interfaces derived from it are **thread-safe** and can
  * be safely invoked from concurrent coroutines without external synchronization.
- *
- * ### Not stable for inheritance
- *
- * **`Job` interface and all its derived interfaces are not stable for inheritance in 3rd party libraries**,
- * as new methods might be added to this interface in the future, but is stable for use.
  */
+@SubclassOptInRequired(markerClass = BrittleForInheritanceCoroutinesApi::class)
 public interface Job : CoroutineContext.Element {
     /**
      * Key for [Job] instance in the coroutine context.
@@ -404,6 +400,7 @@ public fun interface DisposableHandle {
  */
 @InternalCoroutinesApi
 @Deprecated(level = DeprecationLevel.ERROR, message = "This is internal API and may be removed in the future releases")
+@SubclassOptInRequired(markerClass = BrittleForInheritanceCoroutinesApi::class)
 public interface ChildJob : Job {
     /**
      * Parent is cancelling its child by invoking this method.
@@ -423,6 +420,7 @@ public interface ChildJob : Job {
  */
 @InternalCoroutinesApi
 @Deprecated(level = DeprecationLevel.ERROR, message = "This is internal API and may be removed in the future releases")
+@SubclassOptInRequired(markerClass = BrittleForInheritanceCoroutinesApi::class)
 public interface ParentJob : Job {
     /**
      * Child job is using this method to learn its cancellation cause when the parent cancels it with [ChildJob.parentCancelled].

kotlinx-coroutines-core/common/src/JobSupport.kt

@@ -20,6 +20,7 @@ import kotlin.jvm.*
  * @param active when `true` the job is created in _active_ state, when `false` in _new_ state. See [Job] for details.
  * @suppress **This is unstable API and it is subject to change.**
  */
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 @Deprecated(level = DeprecationLevel.ERROR, message = "This is internal API and may be removed in the future releases")
 public open class JobSupport constructor(active: Boolean) : Job, ChildJob, ParentJob {
     final override val key: CoroutineContext.Key<*> get() = Job
@@ -1409,6 +1410,7 @@ private class Empty(override val isActive: Boolean) : Incomplete {
     override fun toString(): String = "Empty{${if (isActive) "Active" else "New" }}"
 }
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 @PublishedApi // for a custom job in the test module
 internal open class JobImpl(parent: Job?) : JobSupport(true), CompletableJob {
     init { initParentJob(parent) }

kotlinx-coroutines-core/common/src/NonCancellable.kt

@@ -21,6 +21,7 @@ import kotlin.coroutines.*
  * when the parent is cancelled, the whole parent-child relation between parent and child is severed.
  * The parent will not wait for the child's completion, nor will be cancelled when the child crashed.
  */
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 @Suppress("DeprecatedCallableAddReplaceWith")
 public object NonCancellable : AbstractCoroutineContextElement(Job), Job {
 

kotlinx-coroutines-core/common/src/channels/Broadcast.kt

@@ -127,6 +127,7 @@ public fun <E> CoroutineScope.broadcast(
     return coroutine
 }
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private open class BroadcastCoroutine<E>(
     parentContext: CoroutineContext,
     protected val _channel: BroadcastChannel<E>,

kotlinx-coroutines-core/common/src/channels/ChannelCoroutine.kt

@@ -3,6 +3,7 @@ package kotlinx.coroutines.channels
 import kotlinx.coroutines.*
 import kotlin.coroutines.*
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 internal open class ChannelCoroutine<E>(
     parentContext: CoroutineContext,
     protected val _channel: Channel<E>,

kotlinx-coroutines-core/common/src/flow/Builders.kt

@@ -52,6 +52,7 @@ import kotlinx.coroutines.flow.internal.unsafeFlow as flow
 public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)
 
 // Named anonymous object
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {
     override suspend fun collectSafely(collector: FlowCollector<T>) {
         collector.block()
@@ -137,6 +138,7 @@ public fun <T> flowOf(value: T): Flow<T> = flow {
  */
 public fun <T> emptyFlow(): Flow<T> = EmptyFlow
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private object EmptyFlow : Flow<Nothing> {
     override suspend fun collect(collector: FlowCollector<Nothing>) = Unit
 }
@@ -303,6 +305,7 @@ public fun <T> channelFlow(@BuilderInference block: suspend ProducerScope<T>.()
 public fun <T> callbackFlow(@BuilderInference block: suspend ProducerScope<T>.() -> Unit): Flow<T> = CallbackFlowBuilder(block)
 
 // ChannelFlow implementation that is the first in the chain of flow operations and introduces (builds) a flow
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private open class ChannelFlowBuilder<T>(
     private val block: suspend ProducerScope<T>.() -> Unit,
     context: CoroutineContext = EmptyCoroutineContext,

kotlinx-coroutines-core/common/src/flow/Channels.kt

@@ -87,6 +87,7 @@ public fun <T> ReceiveChannel<T>.consumeAsFlow(): Flow<T> = ChannelAsFlow(this,
  * However, additional [buffer] calls cause a separate buffering channel to be created and that is where
  * the context might play a role, because it is used by the producing coroutine.
  */
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private class ChannelAsFlow<T>(
     private val channel: ReceiveChannel<T>,
     private val consume: Boolean,

kotlinx-coroutines-core/common/src/flow/Flow.kt

@@ -173,6 +173,7 @@ import kotlin.coroutines.*
  * These implementations ensure that the context preservation property is not violated, and prevent most
  * of the developer mistakes related to concurrency, inconsistent flow dispatchers, and cancellation.
  */
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 public interface Flow<out T> {
 
     /**
@@ -218,6 +219,7 @@ public interface Flow<out T> {
  * ```
  */
 @ExperimentalCoroutinesApi
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 public abstract class AbstractFlow<T> : Flow<T>, CancellableFlow<T> {
 
     public final override suspend fun collect(collector: FlowCollector<T>) {

kotlinx-coroutines-core/common/src/flow/SharedFlow.kt

@@ -119,6 +119,7 @@ import kotlin.jvm.*
  * might be added to this interface in the future, but is stable for use.
  * Use the `MutableSharedFlow(replay, ...)` constructor function to create an implementation.
  */
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 public interface SharedFlow<out T> : Flow<T> {
     /**
      * A snapshot of the replay cache.
@@ -170,6 +171,7 @@ public interface SharedFlow<out T> : Flow<T> {
  * might be added to this interface in the future, but is stable for use.
  * Use the `MutableSharedFlow(...)` constructor function to create an implementation.
  */
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 public interface MutableSharedFlow<T> : SharedFlow<T>, FlowCollector<T> {
     /**
      * Emits a [value] to this shared flow, suspending on buffer overflow.
@@ -309,6 +311,7 @@ internal class SharedFlowSlot : AbstractSharedFlowSlot<SharedFlowImpl<*>>() {
     }
 }
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 internal open class SharedFlowImpl<T>(
     private val replay: Int,
     private val bufferCapacity: Int,

kotlinx-coroutines-core/common/src/flow/StateFlow.kt

@@ -130,6 +130,7 @@ import kotlin.coroutines.*
  * might be added to this interface in the future, but is stable for use.
  * Use the `MutableStateFlow(value)` constructor function to create an implementation.
  */
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 public interface StateFlow<out T> : SharedFlow<T> {
     /**
      * The current value of this state flow.
@@ -151,6 +152,7 @@ public interface StateFlow<out T> : SharedFlow<T> {
  * might be added to this interface in the future, but is stable for use.
  * Use the `MutableStateFlow()` constructor function to create an implementation.
  */
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 public interface MutableStateFlow<T> : StateFlow<T>, MutableSharedFlow<T> {
     /**
      * The current value of this state flow.
@@ -305,6 +307,7 @@ private class StateFlowSlot : AbstractSharedFlowSlot<StateFlowImpl<*>>() {
     }
 }
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private class StateFlowImpl<T>(
     initialState: Any // T | NULL
 ) : AbstractSharedFlow<StateFlowSlot>(), MutableStateFlow<T>, CancellableFlow<T>, FusibleFlow<T> {

kotlinx-coroutines-core/common/src/flow/internal/AbstractSharedFlow.kt

@@ -1,5 +1,6 @@
 package kotlinx.coroutines.flow.internal
 
+import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.internal.*
@@ -113,6 +114,7 @@ internal abstract class AbstractSharedFlow<S : AbstractSharedFlowSlot<*>> : Sync
  *
  * To avoid that (especially in a more complex scenarios), we do not conflate subscription updates.
  */
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private class SubscriptionCountStateFlow(initialValue: Int) : StateFlow<Int>,
     SharedFlowImpl<Int>(1, Int.MAX_VALUE, BufferOverflow.DROP_OLDEST)
 {

kotlinx-coroutines-core/common/src/flow/internal/ChannelFlow.kt

@@ -16,6 +16,7 @@ internal fun <T> Flow<T>.asChannelFlow(): ChannelFlow<T> =
  *
  * @suppress **This an internal API and should not be used from general code.**
  */
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 @InternalCoroutinesApi
 public interface FusibleFlow<T> : Flow<T> {
     /**
@@ -38,6 +39,7 @@ public interface FusibleFlow<T> : Flow<T> {
  *
  * @suppress **This an internal API and should not be used from general code.**
  */
+@SubclassOptInRequired(BrittleForInheritanceCoroutinesApi::class)
 @InternalCoroutinesApi
 public abstract class ChannelFlow<T>(
     // upstream context
@@ -133,6 +135,7 @@ public abstract class ChannelFlow<T>(
 }
 
 // ChannelFlow implementation that operates on another flow before it
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 internal abstract class ChannelFlowOperator<S, T>(
     @JvmField protected val flow: Flow<S>,
     context: CoroutineContext,

kotlinx-coroutines-core/common/src/flow/internal/Merge.kt

@@ -34,6 +34,7 @@ internal class ChannelFlowTransformLatest<T, R>(
     }
 }
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 internal class ChannelFlowMerge<T>(
     private val flow: Flow<Flow<T>>,
     private val concurrency: Int,
@@ -73,6 +74,7 @@ internal class ChannelFlowMerge<T>(
     override fun additionalToStringProps(): String = "concurrency=$concurrency"
 }
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 internal class ChannelLimitedFlowMerge<T>(
     private val flows: Iterable<Flow<T>>,
     context: CoroutineContext = EmptyCoroutineContext,

kotlinx-coroutines-core/common/src/flow/internal/SafeCollector.common.kt

@@ -100,6 +100,7 @@ internal tailrec fun Job?.transitiveCoroutineParent(collectJob: Job?): Job? {
  * An analogue of the [flow] builder that does not check the context of execution of the resulting flow.
  * Used in our own operators where we trust the context of invocations.
  */
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 @PublishedApi
 internal inline fun <T> unsafeFlow(@BuilderInference crossinline block: suspend FlowCollector<T>.() -> Unit): Flow<T> {
     return object : Flow<T> {

kotlinx-coroutines-core/common/src/flow/operators/Context.kt

@@ -259,6 +259,7 @@ public fun <T> Flow<T>.cancellable(): Flow<T> =
 /**
  * Internal marker for flows that are [cancellable].
  */
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 internal interface CancellableFlow<out T> : Flow<T>
 
 /**

kotlinx-coroutines-core/common/src/flow/operators/Distinct.kt

@@ -3,6 +3,7 @@
 
 package kotlinx.coroutines.flow
 
+import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.internal.*
 import kotlin.jvm.*
 
@@ -58,6 +59,7 @@ private fun <T> Flow<T>.distinctUntilChangedBy(
     else -> DistinctFlowImpl(this, keySelector, areEquivalent)
 }
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private class DistinctFlowImpl<T>(
     private val upstream: Flow<T>,
     @JvmField val keySelector: (T) -> Any?,

kotlinx-coroutines-core/common/src/flow/operators/Share.kt

@@ -363,6 +363,7 @@ public fun <T> MutableSharedFlow<T>.asSharedFlow(): SharedFlow<T> =
 public fun <T> MutableStateFlow<T>.asStateFlow(): StateFlow<T> =
     ReadonlyStateFlow(this, null)
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private class ReadonlySharedFlow<T>(
     flow: SharedFlow<T>,
     @Suppress("unused")
@@ -372,6 +373,7 @@ private class ReadonlySharedFlow<T>(
         fuseSharedFlow(context, capacity, onBufferOverflow)
 }
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private class ReadonlyStateFlow<T>(
     flow: StateFlow<T>,
     @Suppress("unused")
@@ -397,6 +399,7 @@ private class ReadonlyStateFlow<T>(
 public fun <T> SharedFlow<T>.onSubscription(action: suspend FlowCollector<T>.() -> Unit): SharedFlow<T> =
     SubscribedSharedFlow(this, action)
 
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 private class SubscribedSharedFlow<T>(
     private val sharedFlow: SharedFlow<T>,
     private val action: suspend FlowCollector<T>.() -> Unit

kotlinx-coroutines-core/common/src/internal/Scopes.kt

@@ -8,6 +8,7 @@ import kotlin.jvm.*
 /**
  * This is a coroutine instance that is created by [coroutineScope] builder.
  */
+@OptIn(BrittleForInheritanceCoroutinesApi::class)
 internal open class ScopeCoroutine<in T>(
     context: CoroutineContext,
     @JvmField val uCont: Continuation<T> // unintercepted continuation