Pair renamed to Step to avoid clashing with Predef.Pair

Author: danslapman

build.sbt

@@ -1,6 +1,6 @@
 name := "asyncstreams"
 
-version := "0.3"
+version := "0.3.5"
 
 scalaVersion := "2.11.8"
 

readme.md

@@ -23,7 +23,7 @@ asyncstreams is available via jitpack:
 ```
     resolvers += "jitpack" at "https://jitpack.io"
 
-    libraryDependencies += "com.github.danslapman" %% "asyncstreams" % "0.3"
+    libraryDependencies += "com.github.danslapman" %% "asyncstreams" % "0.3.5"
 ```
 
 asyncstreams is based on [scala-async](https://github.com/iboltaev/scala-async) ideas.

src/main/scala/asyncstreams/AsyncStream.scala

@@ -6,14 +6,14 @@ import scala.collection.generic.CanBuildFrom
 import scala.concurrent.{ExecutionContext, Future}
 import scala.language.higherKinds
 
-case class AsyncStream[A](data: Future[Pair[A, AsyncStream[A]]]) {
+case class AsyncStream[A](data: Future[Step[A, AsyncStream[A]]]) {
   import AsyncStream._
 
   def foldLeft[B](start: B)(f: (B, A) => B)(implicit executor: ExecutionContext): Future[B] = {
-    def impl(d: Future[Pair[A, AsyncStream[A]]], acc: Future[B]): Future[B] =
+    def impl(d: Future[Step[A, AsyncStream[A]]], acc: Future[B]): Future[B] =
       d.flatMap {
         case END => acc
-        case pair => impl(pair.second.data, acc map (b => f(b, pair.first)))
+        case step => impl(step.rest.data, acc map (b => f(b, step.value)))
       }
 
     impl(data, Future(start))
@@ -26,16 +26,16 @@ case class AsyncStream[A](data: Future[Pair[A, AsyncStream[A]]]) {
   def takeWhile(p: A => Boolean)(implicit executor: ExecutionContext): AsyncStream[A] =
     new AsyncStream[A](data map {
       case END => END
-      case pair if !p(pair.first) => END
-      case pair => Pair(pair.first, pair.second.takeWhile(p))
+      case step if !p(step.value) => END
+      case step => Step(step.value, step.rest.takeWhile(p))
     })
 
 
   def take(n: Int)(implicit executor: ExecutionContext): AsyncStream[A] =
     if (n <= 0) nil
     else AsyncStream(data.map {
       case END => END
-      case p => Pair(p.first, p.second.take(n - 1))
+      case p => Step(p.value, p.rest.take(n - 1))
     })
 
   def foreach[U](f: (A) => U)(implicit executor: ExecutionContext): Future[Unit] =
@@ -45,12 +45,12 @@ case class AsyncStream[A](data: Future[Pair[A, AsyncStream[A]]]) {
     foldLeft(Future(()))((fu: Future[Unit], a: A) => fu.flatMap(_ => f(a)).map(_ => ())).flatMap(u => u)
 
   def flatten[B](implicit asIterable: A => GenIterable[B], executor: ExecutionContext): AsyncStream[B] = {
-    val streamChunk = (p: Pair[A, AsyncStream[A]]) =>
-      concat(generate(asIterable(p.first))(it => if (it.nonEmpty) Future(it.head, it.tail) else ENDF), p.second.flatten)
+    val streamChunk = (p: Step[A, AsyncStream[A]]) =>
+      concat(generate(asIterable(p.value))(it => if (it.nonEmpty) Future(it.head, it.tail) else ENDF), p.rest.flatten)
 
     AsyncStream(data.flatMap {
       case END => ENDF
-      case pair => streamChunk(pair).data
+      case step => streamChunk(step).data
     })
   }
 }
@@ -59,18 +59,18 @@ case class AsyncStream[A](data: Future[Pair[A, AsyncStream[A]]]) {
 object AsyncStream {
   def nil[A](implicit executor: ExecutionContext): AsyncStream[A] = AsyncStream(ENDF)
   def single[A](item: A)(implicit executor: ExecutionContext): AsyncStream[A] =
-    AsyncStream(Future(Pair(item, nil[A])))
+    AsyncStream(Future(Step(item, nil[A])))
 
   def generate[S, A](start: S)(gen: S => Future[(A, S)])(implicit executor: ExecutionContext): AsyncStream[A] =
     AsyncStream(gen(start).map {
       case END => END
-      case (el, rest) => Pair(el, generate(rest)(gen))
+      case (el, rest) => Step(el, generate(rest)(gen))
     })
 
   def concat[A](s1: AsyncStream[A], s2: AsyncStream[A])(implicit executor: ExecutionContext): AsyncStream[A] =
     new AsyncStream[A](s1.data.flatMap {
       case END => s2.data
-      case p => Future(Pair(p.first, concat(p.second, s2)))
+      case step => Future(Step(step.value, concat(step.rest, s2)))
     })
 }
 

src/main/scala/asyncstreams/AsyncStreamMonad.scala

@@ -16,8 +16,8 @@ class AsyncStreamMonad(implicit executor: ExecutionContext) extends MonadPlus[As
     AsyncStream(
       ma.data.flatMap {
         case END => ENDF
-        case pair => f(pair.first).data.map { pair2 =>
-          Pair(pair2.first, concat(pair2.second, bind(pair.second)(f)))
+        case step => f(step.value).data.map { step2 =>
+          Step(step2.value, concat(step2.rest, bind(step.rest)(f)))
         }
       }
     )
@@ -32,19 +32,19 @@ trait AsyncStreamMonadFunctions {
     })
 
   def isEmpty[A, S](stream: AsyncStream[A])(implicit ex: ExecutionContext): FState[S, Boolean] =
-    FState(s => stream.data.map(pair => (pair eq END, s)))
+    FState(s => stream.data.map(step => (step eq END, s)))
 
   def isEmpty[A, S : FStateMonad](f: S => AsyncStream[A])(implicit fsm: FStateMonad[S], ex: ExecutionContext): FState[S, Boolean] =
     fsm.fcondS((s: S) => isEmpty(f(s)))
 
   def notEmpty[A, S](stream: AsyncStream[A])(implicit ex: ExecutionContext): FState[S, Boolean] =
-    FState(s => stream.data map (pair => (!(pair eq END), s)))
+    FState(s => stream.data map (step => (!(step eq END), s)))
 
   def notEmpty[A, S](f: S => AsyncStream[A])(implicit fsm: FStateMonad[S], ex: ExecutionContext): FState[S, Boolean] =
     fsm.fcondS(s => notEmpty(f(s)))
 
   def get[A, S](stream: AsyncStream[A])(implicit ex: ExecutionContext): FState[S, (A, AsyncStream[A])] =
-    FState(s => stream.data.map(pair => ((pair.first, pair.second), s)))
+    FState(s => stream.data.map(step => ((step.value, step.rest), s)))
 
   def generateS[S,A](start: S)(gen: FState[S, A])(implicit ex: ExecutionContext) =
     AsyncStream.generate(start)(gen.func)

src/main/scala/asyncstreams/Pair.scala

@@ -0,0 +1,10 @@
+package asyncstreams
+
+class Step[A, B](fp: A, sp: => B) {
+  val value = fp
+  lazy val rest = sp
+}
+
+object Step {
+  def apply[A, B](value:  A, rest: => B) = new Step[A, B](value, rest)
+}