StringObservable { public static Observable<String> decode(Observable<byte[]> src, String charsetName) { return decode(src, Charset.forName(charsetName)); }  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 testMultibyteSpanningTwoBuffers() { Observable<byte[]> src = Observable.from(new byte[] { (byte) 0xc2 }, new byte[] { (byte) 0xa1 }); String out = StringObservable.decode(src, "UTF-8").toBlockingObservable().single(); assertEquals("\u00A1", out); }
@Test public void testMalformedAtTheEndReplace() { Observable<byte[]> src = Observable.from(new byte[] { (byte) 0xc2 }); String out = StringObservable.decode(src, "UTF-8").toBlockingObservable().single(); assertEquals("\uFFFD", out); }
@Test public void testMalformedInTheMiddleReplace() { Observable<byte[]> src = Observable.from(new byte[] { (byte) 0xc2, 65 }); String out = StringObservable.decode(src, "UTF-8").toBlockingObservable().single(); assertEquals("\uFFFDA", out); }
@Test(expected = RuntimeException.class) public void testMalformedAtTheEndReport() { Observable<byte[]> src = Observable.from(new byte[] { (byte) 0xc2 }); CharsetDecoder charsetDecoder = Charset.forName("UTF-8").newDecoder(); StringObservable.decode(src, charsetDecoder).toBlockingObservable().single(); }
@Test(expected = RuntimeException.class) public void testMalformedInTheMiddleReport() { Observable<byte[]> src = Observable.from(new byte[] { (byte) 0xc2, 65 }); CharsetDecoder charsetDecoder = Charset.forName("UTF-8").newDecoder(); StringObservable.decode(src, charsetDecoder).toBlockingObservable().single(); }
@Test public void testPropogateError() { Observable<byte[]> src = Observable.from(new byte[] { 65 }); Observable<byte[]> err = Observable.error(new IOException()); CharsetDecoder charsetDecoder = Charset.forName("UTF-8").newDecoder(); try { StringObservable.decode(Observable.concat(src, err), charsetDecoder).toList().toBlockingObservable().single(); fail(); } catch (RuntimeException e) { assertEquals(IOException.class, e.getCause().getClass()); } }
@Test public void testPropogateErrorInTheMiddleOfMultibyte() { Observable<byte[]> src = Observable.from(new byte[] { (byte) 0xc2 }); Observable<byte[]> err = Observable.error(new IOException()); CharsetDecoder charsetDecoder = Charset.forName("UTF-8").newDecoder(); try { StringObservable.decode(Observable.concat(src, err), charsetDecoder).toList().toBlockingObservable().single(); fail(); } catch (RuntimeException e) { assertEquals(MalformedInputException.class, e.getCause().getClass()); } }