Queue new API + docs

arrow-benchmarks-fx/src/jmh/kotlin/arrow/benchmarks/Queue.kt

@@ -1,6 +1,5 @@
 package arrow.benchmarks
 
-import arrow.fx.ConcurrentQueue
 import arrow.fx.ForIO
 import arrow.fx.IO
 import arrow.fx.IOOf
@@ -9,7 +8,6 @@ import arrow.fx.extensions.io.concurrent.concurrent
 import arrow.fx.extensions.io.functor.unit
 import arrow.fx.extensions.io.monad.flatMap
 import arrow.fx.fix
-import arrow.fx.internal.CancellableQueue
 import org.openjdk.jmh.annotations.Benchmark
 import org.openjdk.jmh.annotations.CompilerControl
 import org.openjdk.jmh.annotations.Fork
@@ -34,12 +32,10 @@ open class Queue {
  var size: Int = 0
 
  var ConcurQueue by Delegates.notNull<Queue<ForIO, Int>>()
- var CancelQueue by Delegates.notNull<Queue<ForIO, Int>>()
 
  @Setup(Level.Trial)
  fun createQueues(): Unit {
- ConcurQueue = ConcurrentQueue.empty<ForIO, Int>(IO.concurrent()).fix().unsafeRunSync()
- CancelQueue = CancellableQueue.empty<ForIO, Int>(IO.concurrent()).fix().unsafeRunSync()
+ ConcurQueue = Queue.unbounded<ForIO, Int>(IO.concurrent()).fix().unsafeRunSync()
  }
 
  fun <A> IOOf<A>.repeat(n: Int): IO<A> =
@@ -52,7 +48,4 @@ open class Queue {
 
  @Benchmark
  fun concurrentQueue(): Unit = loop(ConcurQueue)
-
- @Benchmark
- fun cancellableQueue(): Unit = loop(CancelQueue)
 }

arrow-docs/docs/queue/README.md

@@ -162,52 +162,3 @@ fun main(args: Array<String>) {
  println(result.unsafeRunSync())
 }
 ```
-
-### Shutting down
-
-A `Queue` also has the ability to be `shutdown`, interrupting any future calls to `take` or `offer` with a
-`QueueShutdown` error and cancelling any suspended fibers waiting to `take` or `offer`.
-
-```kotlin:ank:playground
-import arrow.fx.*
-import arrow.fx.extensions.fx
-import arrow.fx.extensions.io.concurrent.concurrent
-
-fun main(args: Array<String>) {
- val result =
- //sampleStart
- IO.fx {
- val q = !Queue.bounded<ForIO, Int>(10, IO.concurrent())
- val t = !q.take().fork()
- !q.shutdown()
- !t.join() // Attempting to `join` after `shutdown` results in a `QueueShutdown` error
- }
- //sampleEnd
- println(result.attempt().unsafeRunSync())
-}
-```
-
-Consumers of the `Queue` can also track the event of a shutdown by calling `awaitShutdown` to receive a suspended
-`Concurrent<F>` that will resume once the `Queue` has been shutdown.
-
-```kotlin:ank:playground
-import arrow.fx.*
-import arrow.fx.extensions.fx
-import arrow.fx.extensions.io.concurrent.concurrent
-
-fun main(args: Array<String>) {
- val result =
- //sampleStart
- IO.fx {
- val q = !Queue.bounded<ForIO, Int>(10, IO.concurrent())
- val onShutdown = !q.awaitShutdown().fork()
- !q.offer(42)
- val fortyTwo = !q.take()
- !q.shutdown()
- !onShutdown.join()
- fortyTwo
- }
- //sampleEnd
- println(result.unsafeRunSync())
-}
-```

arrow-fx-test/src/main/kotlin/arrow/fx/test/laws/Law.kt

@@ -1,5 +1,13 @@
 package arrow.fx.test.laws
 
+import arrow.core.extensions.either.eq.eq
+import arrow.fx.IO
+import arrow.fx.IOOf
+import arrow.fx.fix
+import arrow.fx.extensions.io.applicative.applicative
+import arrow.fx.extensions.io.concurrent.waitFor
+import arrow.fx.typeclasses.Duration
+import arrow.fx.typeclasses.seconds
 import arrow.core.test.generators.tuple2
 import arrow.core.test.generators.tuple3
 import arrow.core.test.generators.tuple4
@@ -14,6 +22,18 @@ import io.kotlintest.shouldNot
 fun <A> A.equalUnderTheLaw(b: A, eq: Eq<A>): Boolean =
  shouldBeEq(b, eq).let { true }
 
+fun <A> IOOf<A>.equalUnderTheLaw(b: IOOf<A>, EQA: Eq<A> = Eq.any(), timeout: Duration = 5.seconds): Boolean =
+ (this should object : Matcher<IOOf<A>> {
+ override fun test(value: IOOf<A>): Result =
+ arrow.core.Either.eq(Eq.any(), EQA).run {
+ IO.applicative().mapN(value.fix().attempt(), b.fix().attempt()) { (a, b) ->
+ Result(a.eqv(b), "Expected: $b but found: $a", "$b and $a should be equal")
+ }
+ .waitFor(timeout)
+ .unsafeRunSync()
+ }
+ }).let { true }
+
 fun <A> A.shouldBeEq(b: A, eq: Eq<A>): Unit = this should matchUnderEq(eq, b)
 
 fun <A> A.shouldNotBeEq(b: A, eq: Eq<A>): Unit = this shouldNot matchUnderEq(eq, b)