Merge pull request #4 from davidmoten/npe-fix-1664

StringObservable.split NPE fixes as per RxJava #1664

Author: abersnaze

src/main/java/rx/observables/StringObservable.java

@@ -420,14 +420,16 @@ public Subscriber<? super String> call(final Subscriber<? super String> o) {
 
                     @Override
                     public void onCompleted() {
-                        output(leftOver);
+                        if (leftOver!=null)
+                            output(leftOver);
                         if (!o.isUnsubscribed())
                             o.onCompleted();
                     }
 
                     @Override
                     public void onError(Throwable e) {
-                        output(leftOver);
+                        if (leftOver!=null)
+                            output(leftOver);
                         if (!o.isUnsubscribed())
                             o.onError(e);
                     }

src/test/java/rx/observables/StringObservableTest.java

@@ -142,6 +142,24 @@ public void testSplitOnOh() {
         testSplit("boo:and:foo", "o", 0, "b", "", ":and:f");
     }
 
+    @Test
+    public void testSplitOnEmptyStream() {
+        assertEquals(0, (int) StringObservable.split(Observable.<String>empty(), "\n")
+                .count().toBlocking().single());
+    }
+    
+    @Test
+    public void testSplitOnStreamThatThrowsExceptionImmediately() {
+        RuntimeException ex = new RuntimeException("boo");
+        try {
+            StringObservable.split(Observable.<String>error(ex), "\n")
+                .count().toBlocking().single();
+            fail();
+        } catch (RuntimeException e) {
+            assertEquals(ex, e);
+        }
+    }
+
     public void testSplit(String str, String regex, int limit, String... parts) {
         testSplit(str, regex, 0, Observable.just(str), parts);
         for (int i = 0; i < str.length(); i++) {