2

RxJava なんですが、

    Observable<String> s = Observable.from("hoge");
    s.subscribe((str) -> {
        System.out.println(str);
    });

この文字列 "hoge" を取り出すスマートなやり方は何かありますか?

  • インスタンス変数経由はちょっとダサい
  • final String[] 経由もいまいちダサい

2015/1/18 14:34頃追記

質問の仕方とサンプルがまずかったです。

    Observable<String> s = getS3object.flatMap(obj -> {
        return Observable.create(observer -> {
            new Thread(() -> {
                try {
                    BufferedReader reader = new BufferedReader(new InputStreamReader(obj.getObjectContent()));
                    String line;
                    while ((line = reader.readLine()) != null) {
                        observer.onNext(line);
                    }
                } catch (Throwable e) {
                    observer.onError(e);
                }
                observer.onCompleted();
            }).start();
        });
    }).reduce((acc, x) -> acc + "\n" + x);

    return s.get(); // こんなメソッドは無いが、こんな感じのことがやりたい

こんな感じのことがやりたいことです。

0

2 件の回答 2

3

追加分 (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

配列を消すために、fromjust に変えられます。

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
2
  • すみません、質問の仕方とサンプルがまずかったようです。 Observable.from("hoge") は、ストリームを色々加工した上で、一つの文字列のメッセージだけが入ってるストリームを想定していました。 Observable<String> s = getS3object.flatMap(obj -> {中略}).reduce((acc, x) -> acc + "\n" + x); return s.get(); のようなことがしたかったことになります。 Observable を ObservableWithGet に変換できれば、出来るのかなと思ったんですがやり方がよくわかりませんでした。
    – erukiti
    Commented 2015年1月18日 5:32
  • ああ、それだと全然、質問内容が変わってきますね。。。もう少し回答待つべきでした。更新しておきます。 Commented 2015年1月19日 3:11
2

s.get()は別スレッドでの処理が完了するまでブロックするという認識でよいですか?

そうであれば、こんな感じでtoBlockingが利用できると思います。

return s.toBlocking().first();

参考:
https://github.com/ReactiveX/RxJava/wiki/Blocking-Observable-Operators

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。