StringObservable { public static <T> Observable<String> join(final Observable<T> source, final CharSequence separator) { return Observable.create(new OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(final Observer<? super String> t1) { return source.subscribe(new Observer<T>() { boolean mayAddSeparator; StringBuilder b = new StringBuilder(); @Override public void onNext(T args) { if (mayAddSeparator) { b.append(separator); } mayAddSeparator = true; b.append(String.valueOf(args)); } @Override public void onError(Throwable e) { b = null; t1.onError(e); } @Override public void onCompleted() { String str = b.toString(); b = null; t1.onNext(str); t1.onCompleted(); } }); } }); }  static Observable<String> decode(Observable<byte[]> src, String charsetName); static Observable<String> decode(Observable<byte[]> src, Charset charset); static Observable<String> decode(final Observable<byte[]> src, final CharsetDecoder charsetDecoder); static Observable<byte[]> encode(Observable<String> src, String charsetName); static Observable<byte[]> encode(Observable<String> src, Charset charset); static Observable<byte[]> encode(Observable<String> src, final CharsetEncoder charsetEncoder); static Observable<String> stringConcat(Observable<String> src); static Observable<String> split(final Observable<String> src, String regex); static Observable<String> join(final Observable<T> source, final CharSequence separator);  }
@Test public void testJoinMixed() { Observable<Object> source = Observable.<Object>from("a", 1, "c"); Observable<String> result = StringObservable.join(source, ", "); Observer<Object> observer = mock(Observer.class); result.subscribe(new TestObserver<Object>(observer)); verify(observer, times(1)).onNext("a, 1, c"); verify(observer, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void testJoinWithEmptyString() { Observable<String> source = Observable.from("", "b", "c"); Observable<String> result = StringObservable.join(source, ", "); Observer<Object> observer = mock(Observer.class); result.subscribe(new TestObserver<Object>(observer)); verify(observer, times(1)).onNext(", b, c"); verify(observer, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void testJoinWithNull() { Observable<String> source = Observable.from("a", null, "c"); Observable<String> result = StringObservable.join(source, ", "); Observer<Object> observer = mock(Observer.class); result.subscribe(new TestObserver<Object>(observer)); verify(observer, times(1)).onNext("a, null, c"); verify(observer, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void testJoinSingle() { Observable<String> source = Observable.from("a"); Observable<String> result = StringObservable.join(source, ", "); Observer<Object> observer = mock(Observer.class); result.subscribe(new TestObserver<Object>(observer)); verify(observer, times(1)).onNext("a"); verify(observer, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void testJoinEmpty() { Observable<String> source = Observable.empty(); Observable<String> result = StringObservable.join(source, ", "); Observer<Object> observer = mock(Observer.class); result.subscribe(new TestObserver<Object>(observer)); verify(observer, times(1)).onNext(""); verify(observer, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void testJoinThrows() { Observable<String> source = Observable.concat(Observable.just("a"), Observable.<String>error(new RuntimeException("Forced failure"))); Observable<String> result = StringObservable.join(source, ", "); Observer<Object> observer = mock(Observer.class); result.subscribe(new TestObserver<Object>(observer)); verify(observer, never()).onNext("a"); verify(observer, never()).onCompleted(); verify(observer, times(1)).onError(any(Throwable.class)); }