Test asyncstreams to work with twitter futures

Author: danslapman

build.sbt

@@ -16,8 +16,8 @@ libraryDependencies ++= Seq(
   "org.scalaz" %% "scalaz-core" % "7.2.15",
   "io.monix" %% "monix-eval" % versions("monix") % Test,
   "io.monix" %% "monix-scalaz-72" % versions("monix") % Test,
-  //"com.twitter" %% "util-core" % "6.43.0" % Test,
-  //"io.catbird" %% "catbird-util" % "0.14.0" % Test, //cats instances for util-core
-  //"com.codecommit" %% "shims-core" % "1.0-b0e5152" % Test,
+  "com.twitter" %% "util-core" % "6.45.0" % Test,
+  "io.catbird" %% "catbird-util" % "0.15.0" % Test,
+  "me.jeffshaw.harmony" %% "harmony_cats0-9-0_scalaz7-2" % "1.1" % Test,
   "org.scalatest" %% "scalatest" % "3.0.3" % Test
 )

readme.md

@@ -1,8 +1,6 @@
 asyncstreams [![Release](https://jitpack.io/v/danslapman/asyncstreams.svg)](https://jitpack.io/#danslapman/asyncstreams)
 =========
 
-**Note: 0.4 release is outdated, use master-SNAPSHOT for now**
-
 asyncstreams is a monadic asynchronous stream library. It allows you to write stateful asynchronous algorithms
 that emits elements into a stream:
 
@@ -22,6 +20,7 @@ See more examples in tests.
 
 asyncstreams is tested to work with:
 - standard scala futures
+- twitter futures
 - monix tasks
 
 asyncstreams is available via jitpack:

src/test/scala/asyncstreams/monixTask/AsyncStreamTests.scala renamed to src/test/scala/asyncstreams/monixTask/AsyncStreamTestsWithMonixTask.scala

@@ -13,14 +13,12 @@ import scala.concurrent.Await
 import scala.concurrent.duration._
 import scalaz.syntax.monadPlus._
 
-class AsyncStreamTests extends FunSuite with Matchers {
+class AsyncStreamTestsWithMonixTask extends FunSuite with Matchers {
   import TaskZeroError.ze
   private implicit val scheduler = Scheduler.fixedPool("monix", 4)
   private def makeInfStream = AsyncStream.unfold[Task, Int](0)(_ + 1)
   private def wait[T](f: Task[T], d: FiniteDuration = 5.seconds): T = Await.result(f.runAsync, d)
 
-  Task.never
-
   test("composition operator") {
     val s = 1 ~:: 2 ~:: 3 ~:: AsyncStream.asyncNil[Task, Int]
     wait(s.to[List]) shouldBe List(1, 2, 3)

src/test/scala/asyncstreams/stdFuture/AsyncStreamTests.scala renamed to src/test/scala/asyncstreams/stdFuture/AsyncStreamTestsWithStandardFuture.scala

@@ -14,7 +14,7 @@ import scala.concurrent.{Await, ExecutionContext, Future}
 import scalaz.std.scalaFuture._
 import scalaz.syntax.monadPlus._
 
-class AsyncStreamTests extends FunSuite with Matchers {
+class AsyncStreamTestsWithStandardFuture extends FunSuite with Matchers {
   private implicit val executor = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(4))
 
   private def makeInfStream = AsyncStream.unfold(0)(_ + 1)

src/test/scala/asyncstreams/twitterFuture/AsyncStreamTestsWithTwitterFuture.scala

@@ -0,0 +1,86 @@
+package asyncstreams.twitterFuture
+
+import asyncstreams.Utils._
+import asyncstreams.{ASImpl, AsyncStream}
+import asyncstreams.Implicits.MonadErrorInstances._
+import com.twitter.util.{Await, Future}
+import io.catbird.util.FutureInstances
+import org.scalatest.{FunSuite, Matchers}
+import harmony.toscalaz.typeclass.MonadErrorConverter._
+
+import scala.collection.mutable.ArrayBuffer
+import scalaz.syntax.monadPlus._
+
+class AsyncStreamTestsWithTwitterFuture extends FunSuite with Matchers with FutureInstances {
+  import TwFutureZeroError.ze
+
+  private def wait[T](f: Future[T]): T = Await.result(f)
+  private def makeInfStream = AsyncStream.unfold(0)(_ + 1)
+
+  test("composition operator") {
+    val s = 1 ~:: 2 ~:: 3 ~:: AsyncStream.asyncNil[Future, Int]
+    wait(s.to[List]) shouldBe List(1, 2, 3)
+  }
+
+  test("foldLeft") {
+    val s2 = List(2, 3).toAS[Future]
+    val f = implicitly[ASImpl[Future]].collectLeft(s2)(List[Int]())((list, el) => el :: list)
+    wait(f) shouldBe List(3, 2)
+  }
+
+  test("concatenation") {
+    val s1 = List(0, 1).toAS
+    val s2 = List(2, 3).toAS
+    val f = s1 <+> s2
+    wait(f.to[List]) shouldBe List(0, 1, 2, 3)
+  }
+
+  test("working as monad") {
+    val s1 = List(0, 1).toAS
+    val s2 = List(2, 3).toAS
+
+    val res = for {
+      v1 <- s1
+      v2 <- s2
+    } yield v1 * v2
+
+    wait(res.to[List]) shouldBe List(0, 0, 2, 3)
+  }
+
+  test("takeWhile") {
+    val r = makeInfStream.takeWhile(_ < 4)
+    wait(r.to[List]) shouldBe List(0, 1, 2, 3)
+  }
+
+  test("take") {
+    val r = makeInfStream.take(3)
+    wait(r.to[List]) shouldBe List(0, 1, 2)
+  }
+
+  test("folding large stream should not crash") {
+    val r = makeInfStream.takeWhile(_ < 1000000)
+    wait(r.to[List]) shouldBe (0 to 999999)
+  }
+
+  test("foreach") {
+    val stream = makeInfStream.take(10)
+    val buffer = ArrayBuffer[Int]()
+    val task = stream.foreach(i => buffer += i)
+    Await.ready(task)
+    buffer.to[List] shouldBe 0 :: 1 :: 2 :: 3 :: 4 :: 5 :: 6 :: 7 :: 8 :: 9 :: Nil
+  }
+
+  test("foreachF") {
+    val stream = makeInfStream.take(10)
+    val buffer = ArrayBuffer[Int]()
+    val task = stream.foreachF(i => Future(buffer += i))
+    Await.ready(task)
+    buffer.to[List] shouldBe 0 :: 1 :: 2 :: 3 :: 4 :: 5 :: 6 :: 7 :: 8 :: 9 :: Nil
+  }
+
+  test("flatten") {
+    val stream = Vector.range(0, 1000000).grouped(10).to[Vector].toAS
+    val flatStream = stream.flatten
+    wait(flatStream.to[Vector]) shouldBe Vector.range(0, 1000000)
+  }
+}

src/test/scala/asyncstreams/twitterFuture/TwFutureZeroError.scala

@@ -0,0 +1,12 @@
+package asyncstreams.twitterFuture
+
+import asyncstreams.ZeroError
+import com.twitter.util.Future
+import io.catbird.util.FutureInstances
+import harmony.toscalaz.typeclass.MonadErrorConverter._
+
+object TwFutureZeroError extends FutureInstances {
+  implicit val ze = new ZeroError[Throwable, Future] {
+    override val zeroElement: Throwable = new NoSuchElementException
+  }
+}