Merge pull request #1297 from everpeace/rxjava-scalaz

[RxScala] rxjava-scalaz: providing some type class instances.

Author: benjchristensen

rxjava-contrib/rxjava-scalaz/README.md

@@ -0,0 +1,73 @@
+# rxjava-scalaz
+This provides some useful type class instances for `Observable`.  Therefore you can apply scalaz's fancy operators to `Observable`.
+
+Provided type class instances are `Monoid`, `Monad`, `MonadPlus`, `Traverse`, `Foldable`, etc.
+
+For QuickStart, please refer to [RxScalazDemo](./src/test/scala/rx/lang/scala/scalaz/examples/RxScalazDemo.scala).
+
+## How to use
+
+```scala
+import scalaz._, Scalaz._
+import rx.lang.scala.Observable
+import rx.lang.scala.scalaz._
+
+Observable.items(1, 2) |+| Observable.items(3, 4)             // == Observable.items(1 2 3 4)
+Observable.items(1, 2) ∘ {_ + 1}                              // == Observable.items(2, 3)
+(Observable.items(1, 2) |@| Observable.items(3, 4)) {_ + _}   // == Observable.items(4, 5, 5, 6)
+1.η[Observable]                                              // == Observable.items(1)
+(Observable.items(3) >>= {(i: Int) => Observable.items(i + 1)}) // Observable.items(4)
+```
+
+Some other useful operators are available.  Please see below for details.
+
+## Provided Typeclass Instances
+### Monoid
+`Observable` obviously forms a monoid interms of  [`concat`](https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#concat).
+
+```scala
+(Observable.items(1, 2) |+| Observable.items(3, 4)) === Observable.items(1, 2, 3, 4)
+(Observable.items(1, 2) ⊹ Observable.items(3, 4)) === Observable.items(1, 2, 3, 4)
+mzero[Observable[Int]] === Observable.empty
+```
+
+### Monad, MonadPlus
+Essentially, `Observable` is similar to `Stream`. So, `Observable` can be a Stream-like `Monad` and can be a `MonadPlus` as well as `Monoid`.  Of course, `Observable` can be also `Functor` and `Applicative`.
+
+```scala
+// Functor operators
+(Observable.items(1, 2) ∘ {_ + 1}) === Observable.items(2, 3)
+(Observable.items(1, 2) >| 5) === Observable.items(5, 5)
+Observable.items(1, 2).fpair === Observable.items((1, 1), (2, 2))
+Observable.items(1, 2).fproduct {_ + 1} === Observable.items((1, 2), (2, 3))
+Observable.items(1, 2).strengthL("x") === Observable.items(("x", 1), ("x", 2))
+Observable.items(1, 2).strengthR("x") === Observable.items((1, "x"), (2, "x"))
+Functor[Observable].lift {(_: Int) + 1}(Observable.items(1, 2)) === Observable.items(2, 3)
+
+// Applicative operators
+1.point[Observable] === Observable.items(1)
+1.η[Observable] === Observable.items(1)
+(Observable.items(1, 2) |@| Observable.items(3, 4)) {_ + _} === Observable.items(4, 5, 5, 6)
+(Observable.items(1) <*> {(_: Int) + 1}.η[Observable]) === Observable.items(2)
+Observable.items(1) <* Observable.items(2) === Observable.items(1)
+Observable.items(1) *> Observable.items(2) === Observable.items(2)
+
+// Monad and MonadPlus operators
+(Observable.items(3) >>= {(i: Int) => Observable.items(i + 1)}) === Observable.items(4)
+Observable.items(3) >> Observable.items(2) === Observable.items(2)
+Observable.items(Observable.items(1, 2), Observable.items(3, 4)).μ === Observable.items(1, 2, 3, 4)
+Observable.items(1, 2) <+> Observable.items(3, 4) === Observable.items(1, 2, 3, 4)
+```
+
+### Traverse and Foldable
+`Observable` can be `Traverse` and `Foldable` as well as `Stream`.  This means you can fold `Observable` instance to single value.
+
+```scala
+Observable.items(1, 2, 3).foldMap {_.toString} === "123"
+Observable.items(1, 2, 3).foldLeftM(0)((acc, v) => (acc * v).some) === 6.some
+Observable.items(1, 2, 3).suml === 6
+Observable.items(1, 2, 3).∀(_ > 3) === true
+Observable.items(1, 2, 3).∃(_ > 3) === false
+Observable.items(1, 2, 3).traverse(x => (x + 1).some) === Observable.items(2, 3, 4).some
+Observable.items(1.some, 2.some).sequence === Observable.items(1, 2).some
+```

rxjava-contrib/rxjava-scalaz/build.gradle

@@ -0,0 +1,55 @@
+apply plugin: 'scala'
+apply plugin: 'osgi'
+
+tasks.withType(ScalaCompile) {
+  scalaCompileOptions.fork = true
+  scalaCompileOptions.unchecked = true
+  scalaCompileOptions.setAdditionalParameters(['-feature'])
+
+  configure(scalaCompileOptions.forkOptions) {
+    memoryMaximumSize = '1g'
+    jvmArgs = ['-XX:MaxPermSize=512m']
+  }
+}
+
+sourceSets {
+  main {
+        scala {
+            srcDir 'src/main/scala'
+        }
+    }
+    test {
+        scala {
+            srcDir 'src/test/scala'
+        }
+    }
+}
+
+dependencies {
+    compile project(':rxjava-core')
+    compile project(':language-adaptors:rxjava-scala')
+    compile 'org.scalaz:scalaz-core_2.10:7.0.4'
+
+    testCompile 'org.scalaz:scalaz-scalacheck-binding_2.10:7.0.4'
+    testCompile 'org.typelevel:scalaz-specs2_2.10:0.1.2'
+    testCompile 'junit:junit-dep:4.10'
+}
+
+tasks.compileScala {
+    classpath = classpath + (configurations.compile + configurations.provided)
+}
+
+task repl(type: Exec) {
+    commandLine 'scala', '-cp', sourceSets.test.runtimeClasspath.asPath
+    standardInput = System.in
+}
+
+jar {
+    manifest {
+        name = 'rxjava-scalaz'
+        instruction 'Bundle-Vendor', 'Netflix'
+        instruction 'Bundle-DocURL', 'https://github.com/Netflix/RxJava'
+        instruction 'Import-Package', '!org.junit,!junit.framework,!org.mockito.*,!org.scalatest.*,*'
+        instruction 'Fragment-Host', 'com.netflix.rxjava.core'
+    }
+}

rxjava-contrib/rxjava-scalaz/src/main/scala/rx/lang/scala/scalaz/package.scala

@@ -0,0 +1,67 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.lang.scala
+
+import scala.language.higherKinds
+import _root_.scalaz._
+import _root_.scalaz.Tags.{Zip => TZip}
+
+/**
+ * This package object provides some type class instances for Observable.
+ */
+package object scalaz {
+
+  // Monoid
+  implicit def observableMonoid[A] = new Monoid[Observable[A]] {
+    override def zero: Observable[A] = Observable.empty
+    override def append(f1: Observable[A], f2: => Observable[A]): Observable[A] = f1 ++ f2
+  }
+
+  implicit val observableInstances = new MonadPlus[Observable] with Zip[Observable]
+    with IsEmpty[Observable] with Traverse[Observable] {
+
+    // Monad
+    override def point[A](a: => A) = Observable.items(a)
+    override def bind[A, B](oa: Observable[A])(f: (A) => Observable[B]) = oa.flatMap(f)
+
+    // MonadPlus
+    override def empty[A]: Observable[A] = observableMonoid[A].zero
+    override def plus[A](a: Observable[A], b: => Observable[A]): Observable[A] = observableMonoid[A].append(a, b)
+
+    // Zip
+    override def zip[A, B](a: => Observable[A], b: => Observable[B]): Observable[(A, B)] = a zip b
+
+    // IsEmpty (NOTE: This method is blocking call)
+    override def isEmpty[A](fa: Observable[A]): Boolean = fa.isEmpty.toBlocking.first
+
+    // Traverse (NOTE: This method is blocking call)
+    override def traverseImpl[G[_], A, B](fa: Observable[A])(f: (A) => G[B])(implicit G: Applicative[G]): G[Observable[B]] = {
+      val seed: G[Observable[B]] = G.point(Observable.empty)
+      fa.foldLeft(seed) {
+        (ys, x) => G.apply2(ys, f(x))((bs, b) => bs :+ b)
+      }.toBlocking.first
+    }
+  }
+
+  // Observable can be ZipList like applicative functor.
+  // However, due to https://github.com/scalaz/scalaz/issues/338,
+  // This instance doesn't have 'implicit' modifier.
+  val observableZipApplicative: Applicative[({ type λ[α] = Observable[α] @@ TZip })#λ] = new Applicative[({ type λ[α] = Observable[α] @@ TZip })#λ] {
+    def point[A](a: => A) = TZip(Observable.items(a).repeat)
+    def ap[A, B](oa: => Observable[A] @@ TZip)(of: => Observable[A => B] @@ TZip) = TZip(of.zip(oa) map {fa => fa._1(fa._2)})
+  }
+
+}

rxjava-contrib/rxjava-scalaz/src/test/scala/rx/lang/scala/scalaz/ImplicitsForTest.scala

@@ -0,0 +1,45 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.lang.scala.scalaz
+
+import scalaz._
+import rx.lang.scala.Observable
+import org.scalacheck.Arbitrary
+import scala.concurrent.{Await, Promise}
+import scala.concurrent.duration.Duration
+
+/**
+ * This object provides implicits for tests.
+ */
+object ImplicitsForTest {
+
+  // Equality based on sequenceEqual() method.
+  implicit def observableEqual[A](implicit eqA: Equal[A]) = new Equal[Observable[A]]{
+    def equal(a1: Observable[A], a2: Observable[A]) = {
+      val p = Promise[Boolean]
+      val sub = a1.sequenceEqual(a2).firstOrElse(false).subscribe(v => p.success(v))
+      try {
+        Await.result(p.future, Duration.Inf)
+      } finally {
+        sub.unsubscribe()
+      }
+    }
+  }
+
+  implicit def observableArbitrary[A](implicit a: Arbitrary[A], array: Arbitrary[Array[A]]): Arbitrary[Observable[A]]
+  = Arbitrary(for (arr <- array.arbitrary) yield Observable.items(arr:_*))
+
+}

rxjava-contrib/rxjava-scalaz/src/test/scala/rx/lang/scala/scalaz/ObservableEqualSpec.scala

@@ -0,0 +1,42 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.lang.scala.scalaz
+
+import scalaz._
+import Scalaz._
+import scalaz.scalacheck.ScalazProperties._
+import rx.lang.scala.Observable
+import org.specs2.scalaz.Spec
+import org.specs2.runner.JUnitRunner
+import org.junit.runner.RunWith
+
+/**
+ * Even though Equal[Observable[A]] instance is only for tests.
+ * However, we should test Equal[Observable[A]] instance,
+ * Because the result of whole test is based on this instance.
+ */
+@RunWith(classOf[JUnitRunner])
+class ObservableEqualSpec extends Spec{
+
+  import rx.lang.scala.scalaz._
+  import ImplicitsForTest._
+
+  "Observable" should {
+    "satisfies equal laws" in {
+      checkAll(equal.laws[Observable[Int]])
+    }
+  }
+}

rxjava-contrib/rxjava-scalaz/src/test/scala/rx/lang/scala/scalaz/ObservableIsEmptySpec.scala

@@ -0,0 +1,37 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.lang.scala.scalaz
+
+import scalaz._
+import Scalaz._
+import scalaz.scalacheck.ScalazProperties._
+import rx.lang.scala.Observable
+import org.specs2.scalaz.Spec
+import org.specs2.runner.JUnitRunner
+import org.junit.runner.RunWith
+
+@RunWith(classOf[JUnitRunner])
+class ObservableIsEmptySpec extends Spec{
+
+  import rx.lang.scala.scalaz._
+  import ImplicitsForTest._
+
+  "Observable" should {
+    "satisfies isEmpty laws" in {
+      checkAll(isEmpty.laws[Observable])
+    }
+  }
+}

rxjava-contrib/rxjava-scalaz/src/test/scala/rx/lang/scala/scalaz/ObservableMonadSpec.scala

@@ -0,0 +1,40 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.lang.scala.scalaz
+
+import scalaz._
+import Scalaz._
+import scalaz.scalacheck.ScalazProperties._
+import rx.lang.scala.Observable
+import org.specs2.scalaz.Spec
+import org.specs2.runner.JUnitRunner
+import org.junit.runner.RunWith
+
+@RunWith(classOf[JUnitRunner])
+class ObservableMonadSpec extends Spec {
+
+  import rx.lang.scala.scalaz._
+  import ImplicitsForTest._
+
+  "Observable" should {
+    "satisfies monad laws" in {
+      checkAll(monad.laws[Observable])
+    }
+    "satisfies monadplus laws" in {
+      checkAll(monadPlus.strongLaws[Observable])
+    }
+  }
+}