import rx.observables.StringObservable; //导入依赖的package包/类
@Test
public void decode() {
    byte[] bytes = new byte[16];
    Arrays.fill(bytes, (byte)65);

    byte[] bytes2 = new byte[16];
    Arrays.fill(bytes, (byte)65);
    bytes[15] = 10;

    Observable<byte[]> data = Observable.just(bytes, bytes, bytes2).repeat(500);

    StringObservable.decode(data, StandardCharsets.UTF_8)
    .observeOn(Schedulers.computation(), false, 1)
    .toBlocking()
    .subscribe(System.out::println);
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
/**
 * Creates a DocumentServiceRequest with a query.
 *
 * @param resourceType           the resource type.
 * @param relativePath           the relative URI path.
 * @param querySpec              the query.
 * @param queryCompatibilityMode the QueryCompatibilityMode mode.
 * @param headers                the request headers.
 * @return the created document service request.
 */
public static RxDocumentServiceRequest create(ResourceType resourceType,
        String relativePath,
        SqlQuerySpec querySpec,
        QueryCompatibilityMode queryCompatibilityMode,
        Map<String, String> headers) {
    OperationType operation;
    String queryText;
    switch (queryCompatibilityMode) {
    case SqlQuery:
        if (querySpec.getParameters() != null && querySpec.getParameters().size() > 0) {
            throw new IllegalArgumentException(
                    String.format("Unsupported argument in query compatibility mode '{%s}'",
                            queryCompatibilityMode.name()));
        }

        operation = OperationType.SqlQuery;
        queryText = querySpec.getQueryText();
        break;

    case Default:
    case Query:
    default:
        operation = OperationType.Query;
        queryText = querySpec.toJson();
        break;
    }

    Observable<byte[]> body = StringObservable.encode(Observable.just(queryText), StandardCharsets.UTF_8);
    return new RxDocumentServiceRequest(operation, resourceType, relativePath, body, headers);
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
@Benchmark
public String splitRxJavaStringTake5() {
    return Observable.just(lines).compose(new Transformer<String, String>() {
        @Override
        public Observable<String> call(Observable<String> o) {
            return StringObservable.split(o, "\n");
        }
    }).take(5).last().toBlocking().last();
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
@Benchmark
public String splitRxJavaString() {
    return Observable.just(lines).compose(new Transformer<String, String>() {
        @Override
        public Observable<String> call(Observable<String> o) {
            return StringObservable.split(o, "\n");
        }
    }).last().toBlocking().last();
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
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);
    }
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
public static void assertEquals(Observable<JsonToken> in, String out) {
  TestSubscriber<Object> ts = new TestSubscriber<>();

  in.compose(RxJson.toJson())
    .compose(StringObservable::stringConcat)
    .map(s -> gson.fromJson(s, Object.class))
    .subscribe(ts);

  ts.awaitTerminalEvent();

  ts.assertNoErrors();
  ts.assertCompleted();
  ts.assertValue(gson.fromJson(out, Object.class));
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
public static void assertEqualsString(Observable<JsonToken> in, String out) {
  TestSubscriber<Object> ts = new TestSubscriber<>();

  in.compose(RxJson.toJson())
    .compose(StringObservable::stringConcat)
    .subscribe(ts);

  ts.assertNoErrors();
  ts.assertCompleted();
  ts.assertValue(out);
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
@Test
public void shouldEmitComplexObjects() throws Exception {
  TestSubscriber<String> ts = new TestSubscriber<>();

  JsonArray.of(Observable.just("foo", Collections.singletonList("bar")).compose(RxJsonGson.toJsonElements()))
    .compose(RxJson.toJson())
    .compose(StringObservable::stringConcat)
    .subscribe(ts);

  ts.assertCompleted();
  ts.assertValue("[\"foo\",[\"bar\"]]");
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
private void assertEquals(Observable<JsonToken> in, String out) {
  TestSubscriber<String> ts = new TestSubscriber<>();

  in.compose(RxJson.toJson()).compose(StringObservable::stringConcat).subscribe(ts);

  ts.assertCompleted();
  ts.assertValue(out);
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
private void runFunction() {
    try {
        final String[] command = buildCommand();
        logger.debug("Executing command: {}", print(command));
        final Process process = runtime.exec(command);
        final InputStreamReader processOutputReader = new InputStreamReader(process.getInputStream());
        StringObservable.from(processOutputReader)
                .flatMap(s -> Observable.from(s.split("\n")))
                .subscribe(output);
    } catch (final IOException e) {
        logger.error("Failed to run command " + this.getClass().getName(), e);
    }
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
static String preprocess(Path hpp) throws IOException {
	Observable<String> globals_hpp = StringObservable.from(Files
			.newBufferedReader(hpp));
	Observable<String> o = StringObservable.split(globals_hpp, "\r?\n")
			.skipWhile(line -> BEGIN.matcher(line).matches() == false)
			.skip(2).map(line -> line.replaceAll("\\\\$", ""))
			.map(line -> line.replaceAll("/\\*.*?\\*/", ""))
			.filter(line -> line.matches("^[ ]+$") == false);
	String all = StringObservable.join(o, "").toBlockingObservable()
			.first();
	return all.replaceAll("EMBEDDED_ONLY\\(\\w*\\(.*?\\)\\)", "");
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
public Observable<String> observeRuntimeLog(final Pod pod) {
	final LogWatch watcher = withDefaultUserPod(pod).watchLog();
	return StringObservable.byLine(StringObservable
			.from(new InputStreamReader(watcher.getOutput())));
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
public Observable<String> observePodLog(String name) {
	LogWatch watcher = client.pods().withName(name).watchLog();
	return StringObservable.byLine(StringObservable.from(new InputStreamReader(watcher.getOutput())));
}
 

import rx.observables.StringObservable; //导入依赖的package包/类
public void testSplitOnEmptyStream() {
    assertEquals(0, (int) StringObservable.split(Observable.<String> empty(), "\n").count()
            .toBlocking().single());
}