Merge pull request #853 from benjchristensen/string-observable

StringObservable Update

Author: benjchristensen

rxjava-contrib/rxjava-string/src/main/java/rx/observables/StringObservable.java

@@ -1,12 +1,12 @@
 /**
  * Copyright 2013 Netflix, Inc.
- *
+ * 
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -15,6 +15,9 @@
  */
 package rx.observables;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
 import java.nio.ByteBuffer;
 import java.nio.CharBuffer;
 import java.nio.charset.CharacterCodingException;
@@ -27,15 +30,72 @@
 import java.util.regex.Pattern;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
-import rx.Observer;
-import rx.Subscription;
+import rx.Observable.OnSubscribe;
+import rx.Subscriber;
+import rx.operators.Operator;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 
 public class StringObservable {
+    public static Observable<byte[]> from(final InputStream i) {
+        return from(i, 8 * 1024);
+    }
+
+    public static Observable<byte[]> from(final InputStream i, final int size) {
+        return Observable.create(new OnSubscribe<byte[]>() {
+            @Override
+            public void call(Subscriber<? super byte[]> o) {
+                byte[] buffer = new byte[size];
+                try {
+                    if (o.isUnsubscribed())
+                        return;
+                    int n = 0;
+                    n = i.read(buffer);
+                    while (n != -1 && !o.isUnsubscribed()) {
+                        o.onNext(Arrays.copyOf(buffer, n));
+                        n = i.read(buffer);
+                    }
+                } catch (IOException e) {
+                    o.onError(e);
+                }
+                if (o.isUnsubscribed())
+                    return;
+                o.onCompleted();
+            }
+        });
+    }
+
+    public static Observable<String> from(final Reader i) {
+        return from(i, 8 * 1024);
+    }
+
+    public static Observable<String> from(final Reader i, final int size) {
+        return Observable.create(new OnSubscribe<String>() {
+            @Override
+            public void call(Subscriber<? super String> o) {
+                char[] buffer = new char[size];
+                try {
+                    if (o.isUnsubscribed())
+                        return;
+                    int n = 0;
+                    n = i.read(buffer);
+                    while (n != -1 && !o.isUnsubscribed()) {
+                        o.onNext(new String(buffer));
+                        n = i.read(buffer);
+                    }
+                } catch (IOException e) {
+                    o.onError(e);
+                }
+                if (o.isUnsubscribed())
+                    return;
+                o.onCompleted();
+            }
+        });
+    }
+
     /**
-     * Decodes a stream the multibyte chunks into a stream of strings that works on infinite streams and where handles when a multibyte character spans two chunks.
+     * Decodes a stream the multibyte chunks into a stream of strings that works on infinite streams
+     * and where handles when a multibyte character spans two chunks.
      * 
      * @param src
      * @param charsetName
@@ -46,7 +106,8 @@ public static Observable<String> decode(Observable<byte[]> src, String charsetNa
     }
 
     /**
-     * Decodes a stream the multibyte chunks into a stream of strings that works on infinite streams and where handles when a multibyte character spans two chunks.
+     * Decodes a stream the multibyte chunks into a stream of strings that works on infinite streams
+     * and where handles when a multibyte character spans two chunks.
      * 
      * @param src
      * @param charset
@@ -57,30 +118,31 @@ public static Observable<String> decode(Observable<byte[]> src, Charset charset)
     }
 
     /**
-     * Decodes a stream the multibyte chunks into a stream of strings that works on infinite streams and where handles when a multibyte character spans two chunks.
+     * Decodes a stream the multibyte chunks into a stream of strings that works on infinite streams
+     * and where handles when a multibyte character spans two chunks.
      * This method allows for more control over how malformed and unmappable characters are handled.
      * 
      * @param src
      * @param charsetDecoder
      * @return
      */
     public static Observable<String> decode(final Observable<byte[]> src, final CharsetDecoder charsetDecoder) {
-        return Observable.create(new OnSubscribeFunc<String>() {
+        return src.lift(new Operator<String, byte[]>() {
             @Override
-            public Subscription onSubscribe(final Observer<? super String> observer) {
-                return src.subscribe(new Observer<byte[]>() {
+            public Subscriber<? super byte[]> call(final Subscriber<? super String> o) {
+                return new Subscriber<byte[]>(o) {
                     private ByteBuffer leftOver = null;
 
                     @Override
                     public void onCompleted() {
                         if (process(null, leftOver, true))
-                            observer.onCompleted();
+                            o.onCompleted();
                     }
 
                     @Override
                     public void onError(Throwable e) {
                         if (process(null, leftOver, true))
-                            observer.onError(e);
+                            o.onError(e);
                     }
 
                     @Override
@@ -120,7 +182,7 @@ public boolean process(byte[] next, ByteBuffer last, boolean endOfInput) {
                                 cr.throwException();
                             }
                             catch (CharacterCodingException e) {
-                                observer.onError(e);
+                                o.onError(e);
                                 return false;
                             }
                         }
@@ -134,11 +196,11 @@ public boolean process(byte[] next, ByteBuffer last, boolean endOfInput) {
 
                         String string = cb.toString();
                         if (!string.isEmpty())
-                            observer.onNext(string);
+                            o.onNext(string);
 
                         return true;
                     }
-                });
+                };
             }
         });
     }
@@ -190,13 +252,14 @@ public byte[] call(String str) {
     }
 
     /**
-     * Gather up all of the strings in to one string to be able to use it as one message. Don't use this on infinite streams.
+     * Gather up all of the strings in to one string to be able to use it as one message. Don't use
+     * this on infinite streams.
      * 
      * @param src
      * @return
      */
     public static Observable<String> stringConcat(Observable<String> src) {
-        return src.aggregate(new Func2<String, String, String>() {
+        return src.reduce(new Func2<String, String, String>() {
             @Override
             public String call(String a, String b) {
                 return a + b;
@@ -218,22 +281,25 @@ public String call(String a, String b) {
      */
     public static Observable<String> split(final Observable<String> src, String regex) {
         final Pattern pattern = Pattern.compile(regex);
-        return Observable.create(new OnSubscribeFunc<String>() {
+
+        return src.lift(new Operator<String, String>() {
             @Override
-            public Subscription onSubscribe(final Observer<? super String> observer) {
-                return src.subscribe(new Observer<String>() {
+            public Subscriber<? super String> call(final Subscriber<? super String> o) {
+                return new Subscriber<String>(o) {
                     private String leftOver = null;
 
                     @Override
                     public void onCompleted() {
                         output(leftOver);
-                        observer.onCompleted();
+                        if (!o.isUnsubscribed())
+                            o.onCompleted();
                     }
 
                     @Override
                     public void onError(Throwable e) {
                         output(leftOver);
-                        observer.onError(e);
+                        if (!o.isUnsubscribed())
+                            o.onError(e);
                     }
 
                     @Override
@@ -250,24 +316,29 @@ public void onNext(String segment) {
                     }
 
                     private int emptyPartCount = 0;
+
                     /**
                      * when limit == 0 trailing empty parts are not emitted.
+                     * 
                      * @param part
                      */
                     private void output(String part) {
                         if (part.isEmpty()) {
                             emptyPartCount++;
                         }
                         else {
-                            for(; emptyPartCount>0; emptyPartCount--)
-                                observer.onNext("");
-                            observer.onNext(part);
+                            for (; emptyPartCount > 0; emptyPartCount--)
+                                if (!o.isUnsubscribed())
+                                    o.onNext("");
+                            if (!o.isUnsubscribed())
+                                o.onNext(part);
                         }
                     }
-                });
+                };
             }
         });
     }
+
     /**
      * Concatenates the sequence of values by adding a separator
      * between them and emitting the result once the source completes.
@@ -276,49 +347,55 @@ private void output(String part) {
      * {@link java.lang.String#valueOf(java.lang.Object)} calls.
      * <p>
      * For example:
+     * 
      * <pre>
-     * Observable&lt;Object> source = Observable.from("a", 1, "c");
-     * Observable&lt;String> result = join(source, ", ");
+     * Observable&lt;Object&gt; source = Observable.from(&quot;a&quot;, 1, &quot;c&quot;);
+     * Observable&lt;String&gt; result = join(source, &quot;, &quot;);
      * </pre>
      * 
      * will yield a single element equal to "a, 1, c".
      * 
-     * @param source the source sequence of CharSequence values
-     * @param separator the separator to a
+     * @param source
+     *            the source sequence of CharSequence values
+     * @param separator
+     *            the separator to a
      * @return an Observable which emits a single String value having the concatenated
      *         values of the source observable with the separator between elements
      */
     public static <T> Observable<String> join(final Observable<T> source, final CharSequence separator) {
-        return Observable.create(new OnSubscribeFunc<String>() {
-
+        return source.lift(new Operator<String, T>() {
             @Override
-            public Subscription onSubscribe(final Observer<? super String> t1) {
-                return source.subscribe(new Observer<T>() {
+            public Subscriber<T> call(final Subscriber<? super String> o) {
+                return new Subscriber<T>(o) {
                     boolean mayAddSeparator;
                     StringBuilder b = new StringBuilder();
+
                     @Override
-                    public void onNext(T args) {
-                        if (mayAddSeparator) {
-                            b.append(separator);
-                        }
-                        mayAddSeparator = true;
-                        b.append(String.valueOf(args));
+                    public void onCompleted() {
+                        String str = b.toString();
+                        b = null;
+                        if (!o.isUnsubscribed())
+                            o.onNext(str);
+                        if (!o.isUnsubscribed())
+                            o.onCompleted();
                     }
 
                     @Override
                     public void onError(Throwable e) {
                         b = null;
-                        t1.onError(e);
+                        if (!o.isUnsubscribed())
+                            o.onError(e);
                     }
 
                     @Override
-                    public void onCompleted() {
-                        String str = b.toString();
-                        b = null;
-                        t1.onNext(str);
-                        t1.onCompleted();
+                    public void onNext(Object t) {
+                        if (mayAddSeparator) {
+                            b.append(separator);
+                        }
+                        mayAddSeparator = true;
+                        b.append(String.valueOf(t));
                     }
-                });
+                };
             }
         });
     }