Small Queue improvements

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/stream/concurrent/PubSub.kt

@@ -32,7 +32,7 @@ internal interface Publish<A> {
  * Evaluates to `false` if element was not published.
  * Evaluates to `true` if element was published successfully.
  */
- suspend fun tryPublish(a: A): Boolean
+ fun tryPublish(a: A): Boolean
 }
 
 internal interface Subscribe<A, Selector> {
@@ -563,7 +563,14 @@ internal class DefaultPubSub<I, O, QS, S>(private val strategy: PubSub.Strategy<
 
  val state = atomic(initial)
 
- suspend fun <X> update(f: (PubSub.PubSubState<I, O, QS, S>) -> Pair<PubSub.PubSubState<I, O, QS, S>, suspend () -> X>): X =
+ fun <X> update(f: (PubSub.PubSubState<I, O, QS, S>) -> Pair<PubSub.PubSubState<I, O, QS, S>, X>): X =
+ state.modify { ps ->
+ val (ps1, result) = f(ps)
+ val (ps2, _) = loop(ps1) { Unit }
+ Pair(ps2, result)
+ }
+
+ suspend fun <X> modify(f: (PubSub.PubSubState<I, O, QS, S>) -> Pair<PubSub.PubSubState<I, O, QS, S>, suspend () -> X>): X =
  state.modify { ps ->
  val (ps1, result) = f(ps)
  val (ps2, action) = loop(ps1) { Unit }
@@ -681,7 +688,7 @@ internal class DefaultPubSub<I, O, QS, S>(private val strategy: PubSub.Strategy<
  ps.copy(queue = strategy.publish(i, ps.queue))
 
  override suspend fun publish(a: I) =
- update { ps ->
+ modify { ps ->
  if (strategy.accepts(a, ps.queue)) {
  val ps1 = publish_(a, ps)
  Pair(ps1, suspend { Unit })
@@ -698,17 +705,17 @@ internal class DefaultPubSub<I, O, QS, S>(private val strategy: PubSub.Strategy<
  }
  }
 
- override suspend fun tryPublish(a: I): Boolean =
+ override fun tryPublish(a: I): Boolean =
  update { ps ->
- if (!strategy.accepts(a, ps.queue)) Pair(ps, suspend { false })
+ if (!strategy.accepts(a, ps.queue)) Pair(ps, false)
  else {
  val ps1 = publish_(a, ps)
- Pair(ps1, suspend { true })
+ Pair(ps1, true)
  }
  }
 
  override suspend fun get(selector: S): O =
- update { ps ->
+ modify { ps ->
  val (ps, option) = tryGet_(selector, ps)
  when (option) {
  None -> {
@@ -726,7 +733,7 @@ internal class DefaultPubSub<I, O, QS, S>(private val strategy: PubSub.Strategy<
  }
 
  private suspend fun streamingGet(token: Token, selector: S) =
- update<O> { ps ->
+ modify<O> { ps ->
  val (ps, option) = tryGet_(selector, ps)
  when (option) {
  is Some -> Pair(ps, suspend { option.t })
@@ -743,19 +750,19 @@ internal class DefaultPubSub<I, O, QS, S>(private val strategy: PubSub.Strategy<
  }
 
  override suspend fun tryGet(selector: S): Option<O> =
- update { ps ->
+ modify { ps ->
  val (ps1, result) = tryGet_(selector, ps)
  Pair(ps1, suspend { result })
  }
 
  override suspend fun subscribe(selector: S): Boolean =
- update { ps ->
+ modify { ps ->
  val (queue, success) = strategy.subscribe(selector, ps.queue)
  Pair(ps.copy(queue = queue), suspend { success })
  }
 
  override suspend fun unsubscribe(selector: S): Unit =
- update { ps ->
+ modify { ps ->
  Pair(ps.copy(queue = strategy.unsubscribe(selector, ps.queue)), suspend { Unit })
  }
 }

arrow-fx-coroutines/src/main/kotlin/arrow/fx/coroutines/stream/concurrent/Queue.kt

@@ -39,7 +39,7 @@ interface Enqueue<A> {
  *
  * @param a `A` to enqueue
  */
- suspend fun offer1(a: A): Boolean
+ fun tryOffer1(a: A): Boolean
 }
 
 /** Provides the ability to dequeue individual elements from a `Queue`. */
@@ -97,7 +97,7 @@ interface Queue<A> : Enqueue<A>, Dequeue1<A>, Dequeue<A> {
  fun <B> imap(f: (A) -> B, g: (B) -> A): Queue<B> =
  object : Queue<B> {
  override suspend fun enqueue1(a: B) = enqueue1(g(a))
- override suspend fun offer1(a: B): Boolean = offer1(g(a))
+ override fun tryOffer1(a: B): Boolean = tryOffer1(g(a))
  override suspend fun dequeue1(): B = f(this@Queue.dequeue1())
  override suspend fun tryDequeue1(): Option<B> = this@Queue.tryDequeue1().map(f)
  override suspend fun dequeueChunk1(maxSize: Int): Chunk<B> = this@Queue.dequeueChunk1(maxSize).map(f)
@@ -111,51 +111,78 @@ interface Queue<A> : Enqueue<A>, Dequeue1<A>, Dequeue<A> {
  companion object {
 
  /** Creates a queue from the supplied strategy. */
- private suspend fun <S, A> fromStrategy(strategy: PubSub.Strategy<A, Chunk<A>, S, Int>): Queue<A> {
- val pubSub = PubSub.from(strategy)
+ private fun <S, A> fromStrategy(strategy: PubSub.Strategy<A, Chunk<A>, S, Int>): Queue<A> {
+ val pubSub = PubSub.unsafe(strategy)
  return DefaultQueue(pubSub)
  }
 
  /** Creates a queue from the supplied strategy. */
- private suspend fun <S, A> fromStrategyNoneTerminated(strategy: PubSub.Strategy<Option<A>, Option<Chunk<A>>, S, Int>): NoneTerminatedQueue<A> {
- val pubSub = PubSub.from(strategy)
+ private fun <S, A> fromStrategyNoneTerminated(strategy: PubSub.Strategy<Option<A>, Option<Chunk<A>>, S, Int>): NoneTerminatedQueue<A> {
+ val pubSub = PubSub.unsafe(strategy)
  return DefaultNoneTerminatedQueue(pubSub)
  }
 
  /** Creates a FIFO queue with no size bound. */
  suspend fun <A> unbounded(): Queue<A> =
  fromStrategy(Strategy.fifo())
 
+ fun <A> unsafeUnbounded(): Queue<A> =
+ fromStrategy(Strategy.fifo())
+
  /** Creates an unbounded FIFO queue that distributed always at max `fairSize` elements to any subscriber. */
  suspend fun <A> fairUnbounded(fairSize: Int): Queue<A> =
  fromStrategy(Strategy.fifo<A>().transformSelector { size, _ -> min(size, fairSize) })
 
+ fun <A> unsafeFairUnbounded(fairSize: Int): Queue<A> =
+ fromStrategy(Strategy.fifo<A>().transformSelector { size, _ -> min(size, fairSize) })
+
  /** Creates a FIFO queue with the specified size bound. */
  suspend fun <A> bounded(maxSize: Int): Queue<A> =
  fromStrategy(Strategy.boundedFifo(maxSize))
 
+ fun <A> unsafeBounded(maxSize: Int): Queue<A> =
+ fromStrategy(Strategy.boundedFifo(maxSize))
+
  /** Creates a FILO queue with the specified size bound. */
- suspend fun <A> boundedLife(maxSize: Int): Queue<A> =
+ suspend fun <A> boundedLifo(maxSize: Int): Queue<A> =
+ fromStrategy(Strategy.boundedLifo(maxSize))
+
+ fun <A> unsafeBoundedLifo(maxSize: Int): Queue<A> =
  fromStrategy(Strategy.boundedLifo(maxSize))
 
  /** Creates a queue which stores the last `maxSize` enqueued elements and which never blocks on enqueue. */
  suspend fun <A> circularBuffer(maxSize: Int): Queue<A> =
  fromStrategy(Strategy.circularBuffer(maxSize))
 
+ fun <A> unsafeCircularBuffer(maxSize: Int): Queue<A> =
+ fromStrategy(Strategy.circularBuffer(maxSize))
+
  /** Created a bounded queue that distributed always at max `fairSize` elements to any subscriber. */
  suspend fun <A> fairBounded(maxSize: Int, fairSize: Int): Queue<A> =
  fromStrategy(Strategy.boundedFifo<A>(maxSize).transformSelector { size, _ -> min(size, fairSize) })
 
+ fun <A> unsafeFairBounded(maxSize: Int, fairSize: Int): Queue<A> =
+ fromStrategy(Strategy.boundedFifo<A>(maxSize).transformSelector { size, _ -> min(size, fairSize) })
+
  /** Creates a queue which allows at most a single element to be enqueued at any time. */
  suspend fun <A> synchronous(): Queue<A> =
  fromStrategy(Strategy.synchronous())
 
+ fun <A> unsafeSynchronous(): Queue<A> =
+ fromStrategy(Strategy.synchronous())
+
  /** Like [synchronous], except that any enqueue of `None` will never block and cancels any dequeue operation. */
  suspend fun <A> synchronousNoneTerminated(): NoneTerminatedQueue<A> {
  val strategy = Strategy.synchronous<A>()
  val pubSub = PubSub.Strategy.closeNowOption(strategy)
  return fromStrategyNoneTerminated(pubSub)
  }
+
+ fun <A> unsafeSynchronousNoneTerminated(): NoneTerminatedQueue<A> {
+ val strategy = Strategy.synchronous<A>()
+ val pubSub = PubSub.Strategy.closeNowOption(strategy)
+ return fromStrategyNoneTerminated(pubSub)
+ }
  }
 }
 
@@ -276,7 +303,7 @@ internal class DefaultQueue<A>(private val pubSub: PubSub<A, Chunk<A>, Int>) : Q
  override suspend fun enqueue1(a: A) =
  pubSub.publish(a)
 
- override suspend fun offer1(a: A): Boolean =
+ override fun tryOffer1(a: A): Boolean =
  pubSub.tryPublish(a)
 
  override suspend fun dequeue1(): A =
@@ -309,7 +336,7 @@ internal class DefaultNoneTerminatedQueue<A>(
  override suspend fun enqueue1(a: Option<A>) =
  pubSub.publish(a)
 
- override suspend fun offer1(a: Option<A>): Boolean =
+ override fun tryOffer1(a: Option<A>): Boolean =
  pubSub.tryPublish(a)
 
  override suspend fun dequeue1(): Option<A> =