Add publishLast to RxScala

Author: zsxwing

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

@@ -1012,4 +1012,40 @@ class RxScalaDemo extends JUnitSuite {
     subscription.unsubscribe()
   }
 
+  def createAHotObservable: Observable[String] = {
+    var first = true
+    Observable[String] {
+      subscriber =>
+        if (first) {
+          subscriber.onNext("1st: First")
+          subscriber.onNext("1st: Last")
+          first = false
+        }
+        else {
+          subscriber.onNext("2nd: First")
+          subscriber.onNext("2nd: Last")
+        }
+        subscriber.onCompleted()
+    }
+  }
+
+  @Test def withoutPublishLastExample() {
+    val hot = createAHotObservable
+    hot.takeRight(1).subscribe(n => println(s"subscriber 1 gets $n"))
+    hot.takeRight(1).subscribe(n => println(s"subscriber 2 gets $n"))
+  }
+
+  @Test def publishLastExample() {
+    val hot = createAHotObservable
+    val o = hot.publishLast
+    o.subscribe(n => println(s"subscriber 1 gets $n"))
+    o.subscribe(n => println(s"subscriber 2 gets $n"))
+    o.connect
+  }
+
+  @Test def publishLastExample2() {
+    val hot = createAHotObservable
+    val o = hot.publishLast(co => co ++ co) // "++" subscribes "co" twice
+    o.subscribe(n => println(s"subscriber gets $n"))
+  }
 }

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

@@ -1512,6 +1512,40 @@ trait Observable[+T]
     toScalaObservable[R](thisJava.publish(fJava, initialValue))
   }
 
+  /**
+   * Returns a [[ConnectableObservable]] that emits only the last item emitted by the source Observable.
+   * A [[ConnectableObservable]] resembles an ordinary Observable, except that it does not begin emitting items
+   * when it is subscribed to, but only when its `connect` method is called.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/publishLast.png">
+   *
+   * @return a [[ConnectableObservable]] that emits only the last item emitted by the source Observable
+   */
+  def publishLast: ConnectableObservable[T] = {
+    new ConnectableObservable[T](asJavaObservable.publishLast())
+  }
+
+  /**
+   * Returns an Observable that emits an item that results from invoking a specified selector on the last item
+   * emitted by a [[ConnectableObservable]] that shares a single subscription to the source Observable.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/publishLast.f.png">
+   *
+   * @param selector a function that can use the multicasted source sequence as many times as needed, without
+   *                 causing multiple subscriptions to the source Observable. Subscribers to the source will only
+   *                 receive the last item emitted by the source.
+   * @return an Observable that emits an item that is the result of invoking the selector on a [[ConnectableObservable]]
+   *         that shares a single subscription to the source Observable
+   */
+  def publishLast[R](selector: Observable[T] => Observable[R]): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val fJava = new rx.functions.Func1[rx.Observable[T], rx.Observable[R]]() {
+      override def call(jo: rx.Observable[T]): rx.Observable[R] =
+        selector(toScalaObservable[T](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+    }
+    toScalaObservable[R](thisJava.publishLast(fJava))
+  }
+
   // TODO add Scala-like aggregate function
 
   /**