reactive/kotlinx-coroutines-reactive/src/Publish.kt - platform/external/kotlinx.coroutines - Git at Google

 /*
  * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 package kotlinx.coroutines.reactive

 import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
 import kotlinx.coroutines.selects.*
 import kotlinx.coroutines.sync.*
 import org.reactivestreams.*
 import kotlin.coroutines.*

 /**
  * Creates a cold reactive [Publisher] that runs a given [block] in a coroutine.
  *
  * Every time the returned flux is subscribed, it starts a new coroutine in the specified [context].
  * The coroutine emits (via [Subscriber.onNext]) values with [send][ProducerScope.send],
  * completes (via [Subscriber.onComplete]) when the coroutine completes or channel is explicitly closed, and emits
  * errors (via [Subscriber.onError]) if the coroutine throws an exception or closes channel with a cause.
  * Unsubscribing cancels the running coroutine.
  *
  * Invocations of [send][ProducerScope.send] are suspended appropriately when subscribers apply back-pressure and to
  * ensure that [onNext][Subscriber.onNext] is not invoked concurrently.
  *
  * Coroutine context can be specified with [context] argument.
  * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [Dispatchers.Default] is
  * used.
  *
  * **Note: This is an experimental api.** Behaviour of publishers that work as children in a parent scope with respect
  *        to cancellation and error handling may change in the future.
  *
  * @throws IllegalArgumentException if the provided [context] contains a [Job] instance.
  */
 public fun <T> publish(
     context: CoroutineContext = EmptyCoroutineContext,
     @BuilderInference block: suspend ProducerScope<T>.() -> Unit
 ): Publisher<T> {
     require(context[Job] === null) { "Publisher context cannot contain job in it." +
             "Its lifecycle should be managed via subscription. Had $context" }
     return publishInternal(GlobalScope, context, DEFAULT_HANDLER, block)
 }

 /** @suppress For internal use from other reactive integration modules only */
 @InternalCoroutinesApi
 public fun <T> publishInternal(
     scope: CoroutineScope, // support for legacy publish in scope
     context: CoroutineContext,
     exceptionOnCancelHandler: (Throwable, CoroutineContext) -> Unit,
     block: suspend ProducerScope<T>.() -> Unit
 ): Publisher<T> = Publisher { subscriber ->
     // specification requires NPE on null subscriber
     if (subscriber == null) throw NullPointerException("Subscriber cannot be null")
     val newContext = scope.newCoroutineContext(context)
     val coroutine = PublisherCoroutine(newContext, subscriber, exceptionOnCancelHandler)
     subscriber.onSubscribe(coroutine) // do it first (before starting coroutine), to avoid unnecessary suspensions
     coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
 }

 private const val CLOSED = -1L    // closed, but have not signalled onCompleted/onError yet
 private const val SIGNALLED = -2L  // already signalled subscriber onCompleted/onError
 private val DEFAULT_HANDLER: (Throwable, CoroutineContext) -> Unit = { t, ctx -> if (t !is CancellationException) handleCoroutineException(ctx, t) }

 /** @suppress */
 @Suppress("CONFLICTING_JVM_DECLARATIONS", "RETURN_TYPE_MISMATCH_ON_INHERITANCE")
 @InternalCoroutinesApi
 public class PublisherCoroutine<in T>(
     parentContext: CoroutineContext,
     private val subscriber: Subscriber<T>,
     private val exceptionOnCancelHandler: (Throwable, CoroutineContext) -> Unit
 ) : AbstractCoroutine<Unit>(parentContext, false, true), ProducerScope<T>, Subscription {
     override val channel: SendChannel<T> get() = this

     private val _nRequested = atomic(0L) // < 0 when closed (CLOSED or SIGNALLED)

     @Volatile
     private var cancelled = false // true after Subscription.cancel() is invoked

     override val isClosedForSend: Boolean get() = !isActive
     override fun close(cause: Throwable?): Boolean = cancelCoroutine(cause)
     override fun invokeOnClose(handler: (Throwable?) -> Unit): Nothing =
         throw UnsupportedOperationException("PublisherCoroutine doesn't support invokeOnClose")

     // Mutex is locked when either nRequested == 0 or while subscriber.onXXX is being invoked
     private val mutex: Mutex = Mutex(locked = true)

     @Suppress("UNCHECKED_CAST", "INVISIBLE_MEMBER")
     override val onSend: SelectClause2<T, SendChannel<T>> get() = SelectClause2Impl(
         clauseObject = this,
         regFunc = PublisherCoroutine<*>::registerSelectForSend as RegistrationFunction,
         processResFunc = PublisherCoroutine<*>::processResultSelectSend as ProcessResultFunction
     )

     @Suppress("UNCHECKED_CAST", "UNUSED_PARAMETER")
     private fun registerSelectForSend(select: SelectInstance<*>, element: Any?) {
         // Try to acquire the mutex and complete in the registration phase.
         if (mutex.tryLock()) {
             select.selectInRegistrationPhase(Unit)
             return
         }
         // Start a new coroutine that waits for the mutex, invoking `trySelect(..)` after that.
         // Please note that at the point of the `trySelect(..)` invocation the corresponding
         // `select` can still be in the registration phase, making this `trySelect(..)` bound to fail.
         // In this case, the `onSend` clause will be re-registered, which alongside with the mutex
         // manipulation makes the resulting solution obstruction-free.
         launch {
             mutex.lock()
             if (!select.trySelect(this@PublisherCoroutine, Unit)) {
                 mutex.unlock()
             }
         }
     }

     @Suppress("RedundantNullableReturnType", "UNUSED_PARAMETER", "UNCHECKED_CAST")
     private fun processResultSelectSend(element: Any?, selectResult: Any?): Any? {
         doLockedNext(element as T)?.let { throw it }
         return this@PublisherCoroutine
     }

     override fun trySend(element: T): ChannelResult<Unit> =
         if (!mutex.tryLock()) {
             ChannelResult.failure()
         } else {
             when (val throwable = doLockedNext(element)) {
                 null -> ChannelResult.success(Unit)
                 else -> ChannelResult.closed(throwable)
             }
         }

     public override suspend fun send(element: T) {
         mutex.lock()
         doLockedNext(element)?.let { throw it }
     }

     /*
      * This code is not trivial because of the following properties:
      * 1. It ensures conformance to the reactive specification that mandates that onXXX invocations should not
      *    be concurrent. It uses Mutex to protect all onXXX invocation and ensure conformance even when multiple
      *    coroutines are invoking `send` function.
      * 2. Normally, `onComplete/onError` notification is sent only when coroutine and all its children are complete.
      *    However, nothing prevents `publish` coroutine from leaking reference to it send channel to some
      *    globally-scoped coroutine that is invoking `send` outside of this context. Without extra precaution this may
      *    lead to `onNext` that is concurrent with `onComplete/onError`, so that is why signalling for
      *    `onComplete/onError` is also done under the same mutex.
      * 3. The reactive specification forbids emitting more elements than requested, so `onNext` is forbidden until the
      *    subscriber actually requests some elements. This is implemented by the mutex being locked when emitting
      *    elements is not permitted (`_nRequested.value == 0`).
      */

     /**
      * Attempts to emit a value to the subscriber and, if back-pressure permits this, unlock the mutex.
      *
      * Requires that the caller has locked the mutex before this invocation.
      *
      * If the channel is closed, returns the corresponding [Throwable]; otherwise, returns `null` to denote success.
      *
      * @throws NullPointerException if the passed element is `null`
      */
     private fun doLockedNext(elem: T): Throwable? {
         if (elem == null) {
             unlockAndCheckCompleted()
             throw NullPointerException("Attempted to emit `null` inside a reactive publisher")
         }
         /** This guards against the case when the caller of this function managed to lock the mutex not because some
          * elements were requested--and thus it is permitted to call `onNext`--but because the channel was closed.
          *
          * It may look like there is a race condition here between `isActive` and a concurrent cancellation, but it's
          * okay for a cancellation to happen during `onNext`, as the reactive spec only requires that we *eventually*
          * stop signalling the subscriber. */
         if (!isActive) {
             unlockAndCheckCompleted()
             return getCancellationException()
         }
         // notify the subscriber
         try {
             subscriber.onNext(elem)
         } catch (cause: Throwable) {
             /** The reactive streams spec forbids the subscribers from throwing from [Subscriber.onNext] unless the
              * element is `null`, which we check not to be the case. Therefore, we report this exception to the handler
              * for uncaught exceptions and consider the subscription cancelled, as mandated by
              * https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.3/README.md#2.13.
              *
              * Some reactive implementations, like RxJava or Reactor, are known to throw from [Subscriber.onNext] if the
              * execution encounters an exception they consider to be "fatal", like [VirtualMachineError] or
              * [ThreadDeath]. Us using the handler for the undeliverable exceptions to signal "fatal" exceptions is
              * inconsistent with RxJava and Reactor, which attempt to bubble the exception up the call chain as soon as
              * possible. However, we can't do much better here, as simply throwing from all methods indiscriminately
              * would violate the contracts we place on them. */
             cancelled = true
             val causeDelivered = close(cause)
             unlockAndCheckCompleted()
             return if (causeDelivered) {
                 // `cause` is the reason this channel is closed
                 cause
             } else {
                 // Someone else closed the channel during `onNext`. We report `cause` as an undeliverable exception.
                 exceptionOnCancelHandler(cause, context)
                 getCancellationException()
             }
         }
         // now update nRequested
         while (true) { // lock-free loop on nRequested
             val current = _nRequested.value
             if (current < 0) break // closed from inside onNext => unlock
             if (current == Long.MAX_VALUE) break // no back-pressure => unlock
             val updated = current - 1
             if (_nRequested.compareAndSet(current, updated)) {
                 if (updated == 0L) {
                     // return to keep locked due to back-pressure
                     return null
                 }
                 break // unlock if updated > 0
             }
         }
         unlockAndCheckCompleted()
         return null
     }

     private fun unlockAndCheckCompleted() {
        /*
         * There is no sense to check completion before doing `unlock`, because completion might
         * happen after this check and before `unlock` (see `signalCompleted` that does not do anything
         * if it fails to acquire the lock that we are still holding).
         * We have to recheck `isCompleted` after `unlock` anyway.
         */
         mutex.unlock()
         // check isCompleted and try to regain lock to signal completion
         if (isCompleted && mutex.tryLock()) {
             doLockedSignalCompleted(completionCause, completionCauseHandled)
         }
     }

     // assert: mutex.isLocked() & isCompleted
     private fun doLockedSignalCompleted(cause: Throwable?, handled: Boolean) {
         try {
             if (_nRequested.value == SIGNALLED)
                 return
             _nRequested.value = SIGNALLED // we'll signal onError/onCompleted (the final state, so no CAS needed)
             // Specification requires that after the cancellation is requested we eventually stop calling onXXX
             if (cancelled) {
                 // If the parent failed to handle this exception, then we must not lose the exception
                 if (cause != null && !handled) exceptionOnCancelHandler(cause, context)
                 return
             }
             if (cause == null) {
                 try {
                     subscriber.onComplete()
                 } catch (e: Throwable) {
                     handleCoroutineException(context, e)
                 }
             } else {
                 try {
                     // This can't be the cancellation exception from `cancel`, as then `cancelled` would be `true`.
                     subscriber.onError(cause)
                 } catch (e: Throwable) {
                     if (e !== cause) {
                         cause.addSuppressed(e)
                     }
                     handleCoroutineException(context, cause)
                 }
             }
         } finally {
             mutex.unlock()
         }
     }

     override fun request(n: Long) {
         if (n <= 0) {
             // Specification requires to call onError with IAE for n <= 0
             cancelCoroutine(IllegalArgumentException("non-positive subscription request $n"))
             return
         }
         while (true) { // lock-free loop for nRequested
             val cur = _nRequested.value
             if (cur < 0) return // already closed for send, ignore requests, as mandated by the reactive streams spec
             var upd = cur + n
             if (upd < 0 || n == Long.MAX_VALUE)
                 upd = Long.MAX_VALUE
             if (cur == upd) return // nothing to do
             if (_nRequested.compareAndSet(cur, upd)) {
                 // unlock the mutex when we don't have back-pressure anymore
                 if (cur == 0L) {
                     /** In a sense, after a successful CAS, it is this invocation, not the coroutine itself, that owns
                      * the lock, given that `upd` is necessarily strictly positive. Thus, no other operation has the
                      * right to lower the value on [_nRequested], it can only grow or become [CLOSED]. Therefore, it is
                      * impossible for any other operations to assume that they own the lock without actually acquiring
                      * it. */
                     unlockAndCheckCompleted()
                 }
                 return
             }
         }
     }

     // assert: isCompleted
     private fun signalCompleted(cause: Throwable?, handled: Boolean) {
         while (true) { // lock-free loop for nRequested
             val current = _nRequested.value
             if (current == SIGNALLED) return // some other thread holding lock already signalled cancellation/completion
             check(current >= 0) // no other thread could have marked it as CLOSED, because onCompleted[Exceptionally] is invoked once
             if (!_nRequested.compareAndSet(current, CLOSED)) continue // retry on failed CAS
             // Ok -- marked as CLOSED, now can unlock the mutex if it was locked due to backpressure
             if (current == 0L) {
                 doLockedSignalCompleted(cause, handled)
             } else {
                 // otherwise mutex was either not locked or locked in concurrent onNext... try lock it to signal completion
                 if (mutex.tryLock()) doLockedSignalCompleted(cause, handled)
                 // Note: if failed `tryLock`, then `doLockedNext` will signal after performing `unlock`
             }
             return // done anyway
         }
     }

     override fun onCompleted(value: Unit) {
         signalCompleted(null, false)
     }

     override fun onCancelled(cause: Throwable, handled: Boolean) {
         signalCompleted(cause, handled)
     }

     override fun cancel() {
         // Specification requires that after cancellation publisher stops signalling
         // This flag distinguishes subscription cancellation request from the job crash
         cancelled = true
         super.cancel(null)
     }
 }

 @Deprecated(
     message = "CoroutineScope.publish is deprecated in favour of top-level publish",
     level = DeprecationLevel.HIDDEN,
     replaceWith = ReplaceWith("publish(context, block)")
 ) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0. Binary compatibility with Spring
 public fun <T> CoroutineScope.publish(
     context: CoroutineContext = EmptyCoroutineContext,
     @BuilderInference block: suspend ProducerScope<T>.() -> Unit
 ): Publisher<T> = publishInternal(this, context, DEFAULT_HANDLER, block)
	/*
	* Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
	*/
	package kotlinx.coroutines.reactive

	import kotlinx.atomicfu.*
	import kotlinx.coroutines.*
	import kotlinx.coroutines.channels.*
	import kotlinx.coroutines.selects.*
	import kotlinx.coroutines.sync.*
	import org.reactivestreams.*
	import kotlin.coroutines.*

	/**
	* Creates a cold reactive [Publisher] that runs a given [block] in a coroutine.
	*
	* Every time the returned flux is subscribed, it starts a new coroutine in the specified [context].
	* The coroutine emits (via [Subscriber.onNext]) values with [send][ProducerScope.send],
	* completes (via [Subscriber.onComplete]) when the coroutine completes or channel is explicitly closed, and emits
	* errors (via [Subscriber.onError]) if the coroutine throws an exception or closes channel with a cause.
	* Unsubscribing cancels the running coroutine.
	*
	* Invocations of [send][ProducerScope.send] are suspended appropriately when subscribers apply back-pressure and to
	* ensure that [onNext][Subscriber.onNext] is not invoked concurrently.
	*
	* Coroutine context can be specified with [context] argument.
	* If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [Dispatchers.Default] is
	* used.
	*
	* Note: This is an experimental api. Behaviour of publishers that work as children in a parent scope with respect
	* to cancellation and error handling may change in the future.
	*
	* @throws IllegalArgumentException if the provided [context] contains a [Job] instance.
	*/
	public fun <T> publish(
	context: CoroutineContext = EmptyCoroutineContext,
	@BuilderInference block: suspend ProducerScope<T>.() -> Unit
	): Publisher<T> {
	require(context[Job] === null) { "Publisher context cannot contain job in it." +
	"Its lifecycle should be managed via subscription. Had $context" }
	return publishInternal(GlobalScope, context, DEFAULT_HANDLER, block)
	}

	/** @suppress For internal use from other reactive integration modules only */
	@InternalCoroutinesApi
	public fun <T> publishInternal(
	scope: CoroutineScope, // support for legacy publish in scope
	context: CoroutineContext,
	exceptionOnCancelHandler: (Throwable, CoroutineContext) -> Unit,
	block: suspend ProducerScope<T>.() -> Unit
	): Publisher<T> = Publisher { subscriber ->
	// specification requires NPE on null subscriber
	if (subscriber == null) throw NullPointerException("Subscriber cannot be null")
	val newContext = scope.newCoroutineContext(context)
	val coroutine = PublisherCoroutine(newContext, subscriber, exceptionOnCancelHandler)
	subscriber.onSubscribe(coroutine) // do it first (before starting coroutine), to avoid unnecessary suspensions
	coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
	}

	private const val CLOSED = -1L // closed, but have not signalled onCompleted/onError yet
	private const val SIGNALLED = -2L // already signalled subscriber onCompleted/onError
	private val DEFAULT_HANDLER: (Throwable, CoroutineContext) -> Unit = { t, ctx -> if (t !is CancellationException) handleCoroutineException(ctx, t) }

	/** @suppress */
	@Suppress("CONFLICTING_JVM_DECLARATIONS", "RETURN_TYPE_MISMATCH_ON_INHERITANCE")
	@InternalCoroutinesApi
	public class PublisherCoroutine<in T>(
	parentContext: CoroutineContext,
	private val subscriber: Subscriber<T>,
	private val exceptionOnCancelHandler: (Throwable, CoroutineContext) -> Unit
	) : AbstractCoroutine<Unit>(parentContext, false, true), ProducerScope<T>, Subscription {
	override val channel: SendChannel<T> get() = this

	private val _nRequested = atomic(0L) // < 0 when closed (CLOSED or SIGNALLED)

	@Volatile
	private var cancelled = false // true after Subscription.cancel() is invoked

	override val isClosedForSend: Boolean get() = !isActive
	override fun close(cause: Throwable?): Boolean = cancelCoroutine(cause)
	override fun invokeOnClose(handler: (Throwable?) -> Unit): Nothing =
	throw UnsupportedOperationException("PublisherCoroutine doesn't support invokeOnClose")

	// Mutex is locked when either nRequested == 0 or while subscriber.onXXX is being invoked
	private val mutex: Mutex = Mutex(locked = true)

	@Suppress("UNCHECKED_CAST", "INVISIBLE_MEMBER")
	override val onSend: SelectClause2<T, SendChannel<T>> get() = SelectClause2Impl(
	clauseObject = this,
	regFunc = PublisherCoroutine<*>::registerSelectForSend as RegistrationFunction,
	processResFunc = PublisherCoroutine<*>::processResultSelectSend as ProcessResultFunction
	)

	@Suppress("UNCHECKED_CAST", "UNUSED_PARAMETER")
	private fun registerSelectForSend(select: SelectInstance<*>, element: Any?) {
	// Try to acquire the mutex and complete in the registration phase.
	if (mutex.tryLock()) {
	select.selectInRegistrationPhase(Unit)
	return
	}
	// Start a new coroutine that waits for the mutex, invoking `trySelect(..)` after that.
	// Please note that at the point of the `trySelect(..)` invocation the corresponding
	// `select` can still be in the registration phase, making this `trySelect(..)` bound to fail.
	// In this case, the `onSend` clause will be re-registered, which alongside with the mutex
	// manipulation makes the resulting solution obstruction-free.
	launch {
	mutex.lock()
	if (!select.trySelect(this@PublisherCoroutine, Unit)) {
	mutex.unlock()
	}
	}
	}

	@Suppress("RedundantNullableReturnType", "UNUSED_PARAMETER", "UNCHECKED_CAST")
	private fun processResultSelectSend(element: Any?, selectResult: Any?): Any? {
	doLockedNext(element as T)?.let { throw it }
	return this@PublisherCoroutine
	}

	override fun trySend(element: T): ChannelResult<Unit> =
	if (!mutex.tryLock()) {
	ChannelResult.failure()
	} else {
	when (val throwable = doLockedNext(element)) {
	null -> ChannelResult.success(Unit)
	else -> ChannelResult.closed(throwable)
	}
	}

	public override suspend fun send(element: T) {
	mutex.lock()
	doLockedNext(element)?.let { throw it }
	}

	/*
	* This code is not trivial because of the following properties:
	* 1. It ensures conformance to the reactive specification that mandates that onXXX invocations should not
	* be concurrent. It uses Mutex to protect all onXXX invocation and ensure conformance even when multiple
	* coroutines are invoking `send` function.
	* 2. Normally, `onComplete/onError` notification is sent only when coroutine and all its children are complete.
	* However, nothing prevents `publish` coroutine from leaking reference to it send channel to some
	* globally-scoped coroutine that is invoking `send` outside of this context. Without extra precaution this may
	* lead to `onNext` that is concurrent with `onComplete/onError`, so that is why signalling for
	* `onComplete/onError` is also done under the same mutex.
	* 3. The reactive specification forbids emitting more elements than requested, so `onNext` is forbidden until the
	* subscriber actually requests some elements. This is implemented by the mutex being locked when emitting
	* elements is not permitted (`_nRequested.value == 0`).
	*/

	/**
	* Attempts to emit a value to the subscriber and, if back-pressure permits this, unlock the mutex.
	*
	* Requires that the caller has locked the mutex before this invocation.
	*
	* If the channel is closed, returns the corresponding [Throwable]; otherwise, returns `null` to denote success.
	*
	* @throws NullPointerException if the passed element is `null`
	*/
	private fun doLockedNext(elem: T): Throwable? {
	if (elem == null) {
	unlockAndCheckCompleted()
	throw NullPointerException("Attempted to emit `null` inside a reactive publisher")
	}
	/** This guards against the case when the caller of this function managed to lock the mutex not because some
	* elements were requested--and thus it is permitted to call `onNext`--but because the channel was closed.
	*
	* It may look like there is a race condition here between `isActive` and a concurrent cancellation, but it's
	* okay for a cancellation to happen during `onNext`, as the reactive spec only requires that we eventually
	* stop signalling the subscriber. */
	if (!isActive) {
	unlockAndCheckCompleted()
	return getCancellationException()
	}
	// notify the subscriber
	try {
	subscriber.onNext(elem)
	} catch (cause: Throwable) {
	/** The reactive streams spec forbids the subscribers from throwing from [Subscriber.onNext] unless the
	* element is `null`, which we check not to be the case. Therefore, we report this exception to the handler
	* for uncaught exceptions and consider the subscription cancelled, as mandated by
	* https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.3/README.md#2.13.
	*
	* Some reactive implementations, like RxJava or Reactor, are known to throw from [Subscriber.onNext] if the
	* execution encounters an exception they consider to be "fatal", like [VirtualMachineError] or
	* [ThreadDeath]. Us using the handler for the undeliverable exceptions to signal "fatal" exceptions is
	* inconsistent with RxJava and Reactor, which attempt to bubble the exception up the call chain as soon as
	* possible. However, we can't do much better here, as simply throwing from all methods indiscriminately
	* would violate the contracts we place on them. */
	cancelled = true
	val causeDelivered = close(cause)
	unlockAndCheckCompleted()
	return if (causeDelivered) {
	// `cause` is the reason this channel is closed
	cause
	} else {
	// Someone else closed the channel during `onNext`. We report `cause` as an undeliverable exception.
	exceptionOnCancelHandler(cause, context)
	getCancellationException()
	}
	}
	// now update nRequested
	while (true) { // lock-free loop on nRequested
	val current = _nRequested.value
	if (current < 0) break // closed from inside onNext => unlock
	if (current == Long.MAX_VALUE) break // no back-pressure => unlock
	val updated = current - 1
	if (_nRequested.compareAndSet(current, updated)) {
	if (updated == 0L) {
	// return to keep locked due to back-pressure
	return null
	}
	break // unlock if updated > 0
	}
	}
	unlockAndCheckCompleted()
	return null
	}

	private fun unlockAndCheckCompleted() {
	/*
	* There is no sense to check completion before doing `unlock`, because completion might
	* happen after this check and before `unlock` (see `signalCompleted` that does not do anything
	* if it fails to acquire the lock that we are still holding).
	* We have to recheck `isCompleted` after `unlock` anyway.
	*/
	mutex.unlock()
	// check isCompleted and try to regain lock to signal completion
	if (isCompleted && mutex.tryLock()) {
	doLockedSignalCompleted(completionCause, completionCauseHandled)
	}
	}

	// assert: mutex.isLocked() & isCompleted
	private fun doLockedSignalCompleted(cause: Throwable?, handled: Boolean) {
	try {
	if (_nRequested.value == SIGNALLED)
	return
	_nRequested.value = SIGNALLED // we'll signal onError/onCompleted (the final state, so no CAS needed)
	// Specification requires that after the cancellation is requested we eventually stop calling onXXX
	if (cancelled) {
	// If the parent failed to handle this exception, then we must not lose the exception
	if (cause != null && !handled) exceptionOnCancelHandler(cause, context)
	return
	}
	if (cause == null) {
	try {
	subscriber.onComplete()
	} catch (e: Throwable) {
	handleCoroutineException(context, e)
	}
	} else {
	try {
	// This can't be the cancellation exception from `cancel`, as then `cancelled` would be `true`.
	subscriber.onError(cause)
	} catch (e: Throwable) {
	if (e !== cause) {
	cause.addSuppressed(e)
	}
	handleCoroutineException(context, cause)
	}
	}
	} finally {
	mutex.unlock()
	}
	}

	override fun request(n: Long) {
	if (n <= 0) {
	// Specification requires to call onError with IAE for n <= 0
	cancelCoroutine(IllegalArgumentException("non-positive subscription request $n"))
	return
	}
	while (true) { // lock-free loop for nRequested
	val cur = _nRequested.value
	if (cur < 0) return // already closed for send, ignore requests, as mandated by the reactive streams spec
	var upd = cur + n
	if (upd < 0 \|\| n == Long.MAX_VALUE)
	upd = Long.MAX_VALUE
	if (cur == upd) return // nothing to do
	if (_nRequested.compareAndSet(cur, upd)) {
	// unlock the mutex when we don't have back-pressure anymore
	if (cur == 0L) {
	/** In a sense, after a successful CAS, it is this invocation, not the coroutine itself, that owns
	* the lock, given that `upd` is necessarily strictly positive. Thus, no other operation has the
	* right to lower the value on [_nRequested], it can only grow or become [CLOSED]. Therefore, it is
	* impossible for any other operations to assume that they own the lock without actually acquiring
	* it. */
	unlockAndCheckCompleted()
	}
	return
	}
	}
	}

	// assert: isCompleted
	private fun signalCompleted(cause: Throwable?, handled: Boolean) {
	while (true) { // lock-free loop for nRequested
	val current = _nRequested.value
	if (current == SIGNALLED) return // some other thread holding lock already signalled cancellation/completion
	check(current >= 0) // no other thread could have marked it as CLOSED, because onCompleted[Exceptionally] is invoked once
	if (!_nRequested.compareAndSet(current, CLOSED)) continue // retry on failed CAS
	// Ok -- marked as CLOSED, now can unlock the mutex if it was locked due to backpressure
	if (current == 0L) {
	doLockedSignalCompleted(cause, handled)
	} else {
	// otherwise mutex was either not locked or locked in concurrent onNext... try lock it to signal completion
	if (mutex.tryLock()) doLockedSignalCompleted(cause, handled)
	// Note: if failed `tryLock`, then `doLockedNext` will signal after performing `unlock`
	}
	return // done anyway
	}
	}

	override fun onCompleted(value: Unit) {
	signalCompleted(null, false)
	}

	override fun onCancelled(cause: Throwable, handled: Boolean) {
	signalCompleted(cause, handled)
	}

	override fun cancel() {
	// Specification requires that after cancellation publisher stops signalling
	// This flag distinguishes subscription cancellation request from the job crash
	cancelled = true
	super.cancel(null)
	}
	}

	@Deprecated(
	message = "CoroutineScope.publish is deprecated in favour of top-level publish",
	level = DeprecationLevel.HIDDEN,
	replaceWith = ReplaceWith("publish(context, block)")
	) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0. Binary compatibility with Spring
	public fun <T> CoroutineScope.publish(
	context: CoroutineContext = EmptyCoroutineContext,
	@BuilderInference block: suspend ProducerScope<T>.() -> Unit
	): Publisher<T> = publishInternal(this, context, DEFAULT_HANDLER, block)