Merged with #16

Author: SalomonBrys

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

@@ -1,27 +1,46 @@
 package rx.lang.kotlin
 
 import rx.Subscriber
-import rx.observers.SerializedSubscriber
 import rx.exceptions.OnErrorNotImplementedException
+import rx.observers.SerializedSubscriber
+import java.util.ArrayList
+
+public class FunctionSubscriber<T>() : Subscriber<T>() {
+    private val onCompletedFunctions = ArrayList<() -> Unit>()
+    private val onErrorFunctions = ArrayList<(e: Throwable) -> Unit>()
+    private val onNextFunctions = ArrayList<(value: T) -> Unit>()
+    private val onStartFunctions = ArrayList<() -> Unit>()
+
+    override fun onCompleted() = onCompletedFunctions.forEach { it() }
+
+    override fun onError(e: Throwable?) = (e ?: RuntimeException("exception is unknown")).let { ex ->
+        if (onErrorFunctions.isEmpty()) {
+            throw OnErrorNotImplementedException(ex)
+        } else {
+            onErrorFunctions.forEach { it(ex) }
+        }
+    }
 
-public class FunctionSubscriber<T>(onCompletedFunction: () -> Unit, onErrorFunction: (e : Throwable) -> Unit, onNextFunction: (value : T) -> Unit, onStartFunction : () -> Unit) : Subscriber<T>() {
-    private val onCompletedFunction: () -> Unit = onCompletedFunction
-    private val onErrorFunction: (e : Throwable) -> Unit = onErrorFunction
-    private val onNextFunction: (value : T) -> Unit = onNextFunction
-    private val onStartFunction : () -> Unit = onStartFunction
+    override fun onNext(t: T) = onNextFunctions.forEach { it(t) }
 
-    override fun onCompleted() = onCompletedFunction()
+    override fun onStart() = onStartFunctions.forEach { it() }
 
-    override fun onError(e: Throwable?) = onErrorFunction(e ?: RuntimeException("exception is unknown"))
+    fun onCompleted(onCompletedFunction: () -> Unit): FunctionSubscriber<T> = copy { onCompletedFunctions.add(onCompletedFunction) }
+    fun onError(onErrorFunction: (t: Throwable) -> Unit): FunctionSubscriber<T> = copy { onErrorFunctions.add(onErrorFunction) }
+    fun onNext(onNextFunction: (t: T) -> Unit): FunctionSubscriber<T> = copy { onNextFunctions.add(onNextFunction) }
+    fun onStart(onStartFunction : () -> Unit) : FunctionSubscriber<T> = copy { onStartFunctions.add(onStartFunction) }
 
-    override fun onNext(t: T) = onNextFunction(t)
+    private fun copy(block: FunctionSubscriber<T>.() -> Unit): FunctionSubscriber<T> {
+        val newSubscriber = FunctionSubscriber<T>()
+        newSubscriber.onCompletedFunctions.addAll(onCompletedFunctions)
+        newSubscriber.onErrorFunctions.addAll(onErrorFunctions)
+        newSubscriber.onNextFunctions.addAll(onNextFunctions)
+        newSubscriber.onStartFunctions.addAll(onStartFunctions)
 
-    override fun onStart() = onStartFunction()
+        newSubscriber.block()
 
-    fun onCompleted(onCompletedFunction: () -> Unit) : FunctionSubscriber<T> = FunctionSubscriber(onCompletedFunction, this.onErrorFunction, this.onNextFunction, this.onStartFunction)
-    fun onError(onErrorFunction: (t : Throwable) -> Unit) : FunctionSubscriber<T> = FunctionSubscriber(this.onCompletedFunction, onErrorFunction, this.onNextFunction, this.onStartFunction)
-    fun onNext(onNextFunction: (t : T) -> Unit) : FunctionSubscriber<T> = FunctionSubscriber(this.onCompletedFunction, this.onErrorFunction, onNextFunction, this.onStartFunction)
-    fun onStart(onStartFunction : () -> Unit) : FunctionSubscriber<T> = FunctionSubscriber(this.onCompletedFunction, this.onErrorFunction, this.onNextFunction, onStartFunction)
+        return newSubscriber
+    }
 }
 
 public class FunctionSubscriberModifier<T>(init: FunctionSubscriber<T> = subscriber()) {
@@ -34,5 +53,5 @@ public class FunctionSubscriberModifier<T>(init: FunctionSubscriber<T> = subscri
     fun onStart(onStartFunction : () -> Unit) : Unit { subscriber = subscriber.onStart(onStartFunction) }
 }
 
-public fun <T> subscriber(): FunctionSubscriber<T> = FunctionSubscriber({}, {throw OnErrorNotImplementedException(it)}, {}, {})
-public fun <T> Subscriber<T>.synchronized() : Subscriber<T>  = SerializedSubscriber(this)
+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

@@ -1,10 +1,11 @@
 package rx.lang.kotlin
 
-import org.junit.Test as test
 import rx.Subscriber
 import rx.exceptions.OnErrorNotImplementedException
 import java.util.ArrayList
 import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+import org.junit.Test as test
 
 public class SubscribersTest {
     test fun testEmptySubscriber() {
@@ -40,7 +41,41 @@ public class SubscribersTest {
         events.clear()
     }
 
+    test(expected = javaClass<OnErrorNotImplementedException>())
+    fun testNoErrorHandlers() {
+        subscriber<Int>().onError(Exception("expected"))
+    }
+
+    test fun testErrorHandlers() {
+        var errorsCaught = 0
+
+        subscriber<Int>().
+                onError { errorsCaught++ }.
+                onError { errorsCaught++ }.
+                onError(Exception("expected"))
+
+        assertEquals(2, errorsCaught)
+    }
+
+    test fun testMultipleOnNextHandlers() {
+        var nextCaught = 0
+
+        subscriber<Int>().
+                onNext { nextCaught ++ }.
+                onNext { nextCaught ++ }.
+                onNext(1)
+
+        assertEquals(2, nextCaught)
+    }
+
+    test fun testOnStart() {
+        var started = false
+        subscriber<Int>().onStart { started = true }.onStart()
+        assertTrue(started)
+    }
+
     private fun callSubscriberMethods(hasOnError : Boolean, s: Subscriber<Int>) {
+        s.onStart()
         s.onNext(1)
         try {
             s.onError(RuntimeException())