Add 'contains' to RxScala

Author: zsxwing

language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/RxScalaDemo.scala

@@ -25,6 +25,7 @@ import scala.language.implicitConversions
 
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue
+import org.junit.Assert.assertFalse
 import org.junit.Ignore
 import org.junit.Test
 import org.scalatest.junit.JUnitSuite
@@ -645,6 +646,14 @@ class RxScalaDemo extends JUnitSuite {
     println(m.toBlockingObservable.single)
   }
 
+  @Test def containsExample(): Unit = {
+    val o1 = List(1, 2, 3).toObservable.contains(2)
+    assertTrue(o1.toBlockingObservable.single)
+
+    val o2 = List(1, 2, 3).toObservable.contains(4)
+    assertFalse(o2.toBlockingObservable.single)
+  }
+
   @Test def retryExample1(): Unit = {
     val o : Observable[String] = List("alice", "bob", "carol").toObservable
     assertEquals(List("alice", "bob", "carol"), o.retry.toBlockingObservable.toList)

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Observable.scala

@@ -1061,6 +1061,22 @@ trait Observable[+T]
     toScalaObservable[T](asJavaObservable.cache())
   }
 
+  /**
+   * Returns an Observable that emits a Boolean that indicates whether the source Observable emitted a
+   * specified item.
+   *
+   * Note: this method uses `==` to compare elements. It's a bit different from RxJava which uses `Object.equals`.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/contains.png">
+   *
+   *@param elem the item to search for in the emissions from the source Observable
+   * @return an Observable that emits `true` if the specified item is emitted by the source Observable,
+   *         or `false` if the source Observable completes without emitting that item
+   */
+  def contains(elem: Any): Observable[Boolean] = {
+    exists(_ == elem)
+  }
+
   /**
    * Returns a a pair of a start function and an [[rx.lang.scala.Observable]], which waits until the start function is called before it begins emitting
    * items to those [[rx.lang.scala.Observer]]s that have subscribed to it.

language-adaptors/rxjava-scala/src/test/scala/rx/lang/scala/CompletenessTest.scala

@@ -71,6 +71,7 @@ class CompletenessTest extends JUnitSuite {
       "all(Func1[_ >: T, Boolean])" -> "forall(T => Boolean)",
       "buffer(Long, Long, TimeUnit)" -> "buffer(Duration, Duration)",
       "buffer(Long, Long, TimeUnit, Scheduler)" -> "buffer(Duration, Duration, Scheduler)",
+      "contains(T)" -> "contains(Any)",
       "count()" -> "length",
       "dematerialize()" -> "dematerialize(<:<[Observable[T], Observable[Notification[U]]])",
       "elementAt(Int)" -> "[use `.drop(index).first`]",