Merge pull request #17 from SalomonBrys/0.x

Proposing Observable.subscribeWith { ... }

Author: MarioAriasC

src/main/kotlin/rx/lang/kotlin/observables.kt

@@ -2,6 +2,7 @@ package rx.lang.kotlin
 
 import rx.Observable
 import rx.Subscriber
+import rx.Subscription
 import rx.observables.BlockingObservable
 
 public fun <T> emptyObservable() : Observable<T> = Observable.empty()
@@ -96,3 +97,12 @@ public fun <T> Observable<T>.withIndex() : Observable<IndexedValue<T>> = lift {
  *  @returns Observable that merges all [Sequence]s produced by [body] functions
  */
 public fun <T, R> Observable<T>.flatMapSequence( body : (T) -> Sequence<R> ) : Observable<R> = flatMap { body(it).toObservable() }
+
+/**
+ * Subscribe with a subscriber that is configured inside body
+ */
+public inline fun <T> Observable<T>.subscribeWith( body : FunctionSubscriberModifier<T>.() -> Unit ) : Subscription {
+    val modifier = FunctionSubscriberModifier(subscriber<T>())
+    modifier.body()
+    return subscribe(modifier.subscriber)
+}

src/main/kotlin/rx/lang/kotlin/subscribers.kt

@@ -43,5 +43,15 @@ public class FunctionSubscriber<T>() : Subscriber<T>() {
     }
 }
 
+public class FunctionSubscriberModifier<T>(init: FunctionSubscriber<T> = subscriber()) {
+    public var subscriber: FunctionSubscriber<T> = init
+        private set
+
+    fun onCompleted(onCompletedFunction: () -> Unit) : Unit { subscriber = subscriber.onCompleted(onCompletedFunction) }
+    fun onError(onErrorFunction: (t : Throwable) -> Unit) : Unit { subscriber = subscriber.onError(onErrorFunction) }
+    fun onNext(onNextFunction: (t : T) -> Unit) : Unit { subscriber = subscriber.onNext(onNextFunction) }
+    fun onStart(onStartFunction : () -> Unit) : Unit { subscriber = subscriber.onStart(onStartFunction) }
+}
+
 public fun <T> subscriber(): FunctionSubscriber<T> = FunctionSubscriber()
 public fun <T> Subscriber<T>.synchronized(): Subscriber<T> = SerializedSubscriber(this)

src/test/kotlin/rx/lang/kotlin/SubscribersTest.kt

@@ -86,4 +86,52 @@ public class SubscribersTest {
         }
         s.onCompleted()
     }
-}
+
+    test fun testSubscribeWith() {
+        val completeObservable = observable<Int> {
+            it.onNext(1)
+            it.onCompleted()
+        }
+
+        val events = ArrayList<String>()
+
+        completeObservable.subscribeWith {
+            onNext { events.add("onNext($it)") }
+        }
+
+        assertEquals(listOf("onNext(1)"), events)
+        events.clear()
+
+        completeObservable.subscribeWith {
+            onNext { events.add("onNext($it)") }
+            onCompleted { events.add("onCompleted") }
+        }
+
+        assertEquals(listOf("onNext(1)", "onCompleted"), events)
+        events.clear()
+
+        val errorObservable = observable<Int> {
+            it.onNext(1)
+            it.onError(RuntimeException())
+        }
+
+        errorObservable.subscribeWith {
+            onNext { events.add("onNext($it)") }
+            onError { events.add("onError(${it.javaClass.getSimpleName()})") }
+        }
+
+        assertEquals(listOf("onNext(1)", "onError(RuntimeException)"), events)
+        events.clear()
+
+        try {
+            errorObservable.subscribeWith {
+                onNext { events.add("onNext($it)") }
+            }
+        } catch (t: Throwable) {
+            events.add("catch(${t.javaClass.getSimpleName()})")
+        }
+
+        assertEquals(listOf("onNext(1)", "catch(OnErrorNotImplementedException)"), events)
+        events.clear()
+    }
+}