追加分 (01/19)
ああ… 更新された質問の内容だと、全く答えが変わってきますので、回答も更新させてください。Observable
の処理自体は非同期で行われる可能性があるので、「終了を待つ」処理が必要になります。すでにある指摘通り、 toBlocking()
もしくは、 BlockingObservable.from()
です。
コード:
public class ObservableTest {
static class S3ObjectDummy {
InputStream getObjectContent() {
return new ByteArrayInputStream(
"Arthur\nBob\nCharlie\nDick\nEdward".getBytes(StandardCharsets.UTF_8));
}
}
public static void main( String argv[]) {
Observable<S3ObjectDummy> s3object = Observable.from(
Arrays.asList(new S3ObjectDummy(), new S3ObjectDummy()));
Observable<String> strObs = s3object.flatMap(obj ->
Observable.<String>create((observer) -> {
new Thread(() -> {
try {
BufferedReader reader = new BufferedReader(new InputStreamReader(
obj.getObjectContent()));
String line;
while ((line = reader.readLine()) != null) {
observer.onNext(line + "(processed)");
}
} catch (Throwable e) {
observer.onError(e);
}
observer.onCompleted();
}).start();
})
).reduce((acc, x) -> acc + "\n" + x);
// 完了を待つ
BlockingObservable<String> blockObs = strObs.toBlocking();
// 取得
System.out.println( blockObs.first());
}
}
せっかくなのでおまけ。サンプルのコードですが、コンパイルエラーもそうですが、 Thread
を使っているところがちょっと気になりました。以下のようにできたと思います。
public class ObservableTest {
static class S3ObjectDummy {
InputStream getObjectContent() {
return new ByteArrayInputStream(
"Arthur\nBob\nCharlie\nDick\nEdward".getBytes(StandardCharsets.UTF_8));
}
}
public static void main( String argv[]) {
Observable<S3ObjectDummy> s3object = Observable.from(
Arrays.asList(new S3ObjectDummy(), new S3ObjectDummy()));
s3object.flatMap(ObservableGet::processLines)
.reduce((acc, x) -> acc + "\n" + x)
.toBlocking()
.forEach( System.out::println);
}
public static Observable<String> processLines( S3ObjectDummy obj) {
return Observable.<String>create((observer) -> {
try {
BufferedReader reader = new BufferedReader(new InputStreamReader(
obj.getObjectContent()));
String line;
while ((line = reader.readLine()) != null) {
observer.onNext(line + "(processed)");
}
} catch (Throwable e) {
observer.onError(e);
}
observer.onCompleted();
}).subscribeOn(Schedulers.newThread());
}
}
古い回答 (01/18)
前準備
質問のサンプルコードを少し整えます。
Observable<String> obs = Observable.from("hoge");
obs.subscribe((String str) -> {
System.out.println(str);
});
// => <compile error>
かっこ良く(私見)しましょう。
Observable.from(Arrays.asList("hoge").subscribe((s) -> System.out.println(s));
// Output: hoge
配列を消すために、from
は just
に変えられます。
Observable.just("hoge").subscribe((s) -> System.out.println(s));
// Output: hoge
その上で 行を分ける
Observable o = Observable.just("hoge");
o.subscribe((s) -> System.out.println(s));
// Output: hoge
...と、 ここから手をつけます。
Observable.just()
の動きを知る
just()
だけ内部では、オブジェクトを保持する Objservable が使われてます。速度稼ぐためにせこいことしてますね。
System.out.println(ScalarSynchronousObservable.create("hoge").get());
// Output: hoge
観測対象を直接保持する Observable
を作る
質問の「Observableの T を取り出す方法」をサンプルコードベースで実現するために、観測対象オブジェクトを保持するObservable
実装を作りましょうか。
class ObservableWithGet<T> extends Observable<T> {
public static final <T> ObservableWithGet<T> create(T t) { return new ObservableWithGet<T>(t); }
private final T t;
protected ObservableWithGet(final T t) {
super((Subscriber<? super T> s) -> {
s.onNext(t);
s.onCompleted();
});
this.t = t;
}
public T get() { return t; }
}
これで目的のことが実現できると思います。
ObservableWithGet o = ObservableWithGet.create("hoge");
System.out.println(o.get());
o.subscribe((s) -> System.out.println(s));
// Output: hoge\nhoge