Merge pull request #1597 from GeorgiKhomeriki/master

reactivex.io: RxScala getting started examples

Author: benjchristensen

language-adaptors/rxjava-scala/src/examples/reactivex/getting-started/AsyncObservable.scala

@@ -0,0 +1,37 @@
+import rx.lang.scala.Observable
+
+object AsyncObservable extends App {
+  /**
+   * This example shows a custom Observable that does not block
+   * when subscribed to as it spawns a separate thread.
+   */
+  def customObservableNonBlocking(): Observable[String] = {
+    Observable(
+    /*
+     * This 'call' method will be invoked when the Observable is subscribed to.
+     *
+     * It spawns a thread to do it asynchronously.
+     */
+      subscriber => {
+        // For simplicity this example uses a Thread instead of an ExecutorService/ThreadPool
+        new Thread(new Runnable() {
+          def run() {
+            for (i <- 0 to 75) {
+              if (subscriber.isUnsubscribed) {
+                return
+              }
+              subscriber.onNext("value_" + i)
+            }
+            // after sending all values we complete the sequence
+            if (!subscriber.isUnsubscribed) {
+              subscriber.onCompleted()
+            }
+          }
+        }).start()
+      }
+    )
+  }
+
+  // To see output:
+  customObservableNonBlocking().subscribe(println(_))
+}

language-adaptors/rxjava-scala/src/examples/reactivex/getting-started/AsyncWiki.scala

@@ -0,0 +1,32 @@
+import java.net.URL
+import java.util.Scanner
+
+import rx.lang.scala.Observable
+
+object AsyncWiki extends App {
+  /*
+   * Fetch a list of Wikipedia articles asynchronously.
+   */
+  def fetchWikipediaArticleAsynchronously(wikipediaArticleNames: String*): Observable[String] = {
+    Observable(subscriber => {
+      new Thread(new Runnable() {
+        def run() {
+          for (articleName <- wikipediaArticleNames) {
+            if (subscriber.isUnsubscribed) {
+              return
+            }
+            val url = "http://en.wikipedia.org/wiki/" + articleName
+            val art = new Scanner(new URL(url).openStream()).useDelimiter("\\A").next()
+            subscriber.onNext(art)
+          }
+          if (!subscriber.isUnsubscribed) {
+            subscriber.onCompleted()
+          }
+        }
+      }).start()
+    })
+  }
+
+  fetchWikipediaArticleAsynchronously("Tiger", "Elephant")
+    .subscribe(art => println("--- Article ---\n" + art.substring(0, 125)))
+}

language-adaptors/rxjava-scala/src/examples/reactivex/getting-started/AsyncWikiErrorHandling.scala

@@ -0,0 +1,38 @@
+import java.net.URL
+import java.util.Scanner
+
+import rx.lang.scala.Observable
+
+object AsyncWikiErrorHandling extends App {
+  /*
+   * Fetch a list of Wikipedia articles asynchronously, with error handling.
+   */
+  def fetchWikipediaArticleAsynchronously(wikipediaArticleNames: String*): Observable[String] = {
+    Observable(subscriber => {
+      new Thread(new Runnable() {
+        def run() {
+          try {
+            for (articleName <- wikipediaArticleNames) {
+              if (subscriber.isUnsubscribed) {
+                return
+              }
+              val url = "http://en.wikipedia.org/wiki/" + articleName
+              val art = new Scanner(new URL(url).openStream()).useDelimiter("\\A").next()
+              subscriber.onNext(art)
+            }
+            if (!subscriber.isUnsubscribed) {
+              subscriber.onCompleted()
+            }
+          } catch {
+            case t: Throwable => subscriber.onError(t)
+          }
+        }
+      }).start()
+    })
+  }
+
+  fetchWikipediaArticleAsynchronously("Tiger", "Elephant")
+    .subscribe(
+      art => println("--- Article ---\n" + art.substring(0, 125)),
+      e => println("--- Error ---\n" + e.getMessage) )
+}

language-adaptors/rxjava-scala/src/examples/reactivex/getting-started/CreateFromSource.scala

@@ -0,0 +1,13 @@
+import rx.lang.scala.Observable
+
+object CreateFromSource {
+
+  def create() {
+    val o1 = Observable.just("a", "b", "c")
+
+    def list = List(5, 6, 7, 8)
+    val o2 = Observable.from(list)
+
+    val o3 = Observable.just("one object")
+  }
+}

language-adaptors/rxjava-scala/src/examples/reactivex/getting-started/Hello.scala

@@ -0,0 +1,10 @@
+import rx.lang.scala.Observable
+
+object Hello {
+
+  def hello(names: String*) {
+    Observable.from(names) subscribe { n =>
+      println(s"Hello $n!")
+    }
+  }
+}

language-adaptors/rxjava-scala/src/examples/reactivex/getting-started/SyncObservable.scala

@@ -0,0 +1,24 @@
+import rx.lang.scala.Observable
+
+object SyncObservable extends App {
+  /**
+   * This example shows a custom Observable that blocks
+   * when subscribed to (does not spawn an extra thread).
+   */
+  def customObservableBlocking(): Observable[String] = {
+    Observable(aSubscriber => {
+      for (i <- 0 to 50) {
+        if (!aSubscriber.isUnsubscribed) {
+          aSubscriber.onNext("value_" + i)
+        }
+      }
+      // after sending all values we complete the sequence
+      if (!aSubscriber.isUnsubscribed) {
+        aSubscriber.onCompleted()
+      }
+    })
+  }
+
+  // To see output:
+  customObservableBlocking().subscribe(println(_))
+}

language-adaptors/rxjava-scala/src/examples/reactivex/getting-started/Transforming.output

@@ -0,0 +1,5 @@
+onNext => value_10_xform
+onNext => value_11_xform
+onNext => value_12_xform
+onNext => value_13_xform
+onNext => value_14_xform

language-adaptors/rxjava-scala/src/examples/reactivex/getting-started/Transforming.scala

@@ -0,0 +1,13 @@
+object Transforming extends App {
+  /**
+   * Asynchronously calls 'customObservableNonBlocking' and defines
+   * a chain of operators to apply to the callback sequence.
+   */
+  def simpleComposition() {
+    AsyncObservable.customObservableNonBlocking().drop(10).take(5)
+      .map(stringValue => stringValue + "_xform")
+      .subscribe(s => println("onNext => " + s))
+  }
+
+  simpleComposition()
+}

language-adaptors/rxjava-scala/src/examples/reactivex/getting-started/hello.output

@@ -0,0 +1,3 @@
+hello("Ben", "George")
+Hello Ben!
+Hello George!