Observable<T>の T を取り出す方法


RxJava なんですが、

[code]

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

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

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

2015/1/18 14:34頃追記

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

[code]

        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(); // こんなメソッドは無いが、こんな感じのことがやりたい
    
[/code]

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




## 追加分 (01/19)

ああ… 更新された質問の内容だと、全く答えが変わってきますので、回答も更新させてください。`Observable`
の処理自体は非同期で行われる可能性があるので、「終了を待つ」処理が必要になります。すでにある指摘通り、 `toBlocking()` もしくは、
`BlockingObservable.from()` です。

## コード:

[code]

    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());
        }
    }
    
[/code]

せっかくなのでおまけ。サンプルのコードですが、コンパイルエラーもそうですが、 `Thread`
を使っているところがちょっと気になりました。以下のようにできたと思います。

[code]

    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());
        }
    }
    
[/code]

## 古い回答 (01/18)

### 前準備

質問のサンプルコードを少し整えます。

[code]

    Observable<String> obs = Observable.from("hoge");
    obs.subscribe((String str) -> {
      System.out.println(str);
    });
    // => <compile error>
    
[/code]

かっこ良く（私見）しましょう。

[code]

    Observable.from(Arrays.asList("hoge").subscribe((s) -> System.out.println(s));
    // Output: hoge
    
[/code]

配列を消すために、`from` は `just` に変えられます。

[code]

    Observable.just("hoge").subscribe((s) -> System.out.println(s));
    // Output: hoge
    
[/code]

その上で 行を分ける

[code]

    Observable o = Observable.just("hoge");
    o.subscribe((s) -> System.out.println(s));
    // Output: hoge
    
[/code]

...と、 ここから手をつけます。

### `Observable.just()` の動きを知る

`just()` だけ内部では、オブジェクトを保持する Objservable が使われてます。速度稼ぐためにせこいことしてますね。

[code]

    System.out.println(ScalarSynchronousObservable.create("hoge").get());
    // Output: hoge
    
[/code]

### 観測対象を直接保持する `Observable` を作る

質問の「Observableの T
を取り出す方法」をサンプルコードベースで実現するために、観測対象オブジェクトを保持する`Observable`実装を作りましょうか。

[code]

    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; }
    }
    
[/code]

これで目的のことが実現できると思います。

[code]

    ObservableWithGet o = ObservableWithGet.create("hoge");
    System.out.println(o.get());
    o.subscribe((s) -> System.out.println(s));
    // Output: hoge\nhoge
    
[/code]




## 追加分 (01/19)

ああ… 更新された質問の内容だと、全く答えが変わってきますので、回答も更新させてください。`Observable`
の処理自体は非同期で行われる可能性があるので、「終了を待つ」処理が必要になります。すでにある指摘通り、 `toBlocking()` もしくは、
`BlockingObservable.from()` です。

## コード:

[code]

    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());
        }
    }
    
[/code]

せっかくなのでおまけ。サンプルのコードですが、コンパイルエラーもそうですが、 `Thread`
を使っているところがちょっと気になりました。以下のようにできたと思います。

[code]

    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());
        }
    }
    
[/code]

## 古い回答 (01/18)

### 前準備

質問のサンプルコードを少し整えます。

[code]

    Observable<String> obs = Observable.from("hoge");
    obs.subscribe((String str) -> {
      System.out.println(str);
    });
    // => <compile error>
    
[/code]

かっこ良く（私見）しましょう。

[code]

    Observable.from(Arrays.asList("hoge").subscribe((s) -> System.out.println(s));
    // Output: hoge
    
[/code]

配列を消すために、`from` は `just` に変えられます。

[code]

    Observable.just("hoge").subscribe((s) -> System.out.println(s));
    // Output: hoge
    
[/code]

その上で 行を分ける

[code]

    Observable o = Observable.just("hoge");
    o.subscribe((s) -> System.out.println(s));
    // Output: hoge
    
[/code]

...と、 ここから手をつけます。

### `Observable.just()` の動きを知る

`just()` だけ内部では、オブジェクトを保持する Objservable が使われてます。速度稼ぐためにせこいことしてますね。

[code]

    System.out.println(ScalarSynchronousObservable.create("hoge").get());
    // Output: hoge
    
[/code]

### 観測対象を直接保持する `Observable` を作る

質問の「Observableの T
を取り出す方法」をサンプルコードベースで実現するために、観測対象オブジェクトを保持する`Observable`実装を作りましょうか。

[code]

    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; }
    }
    
[/code]

これで目的のことが実現できると思います。

[code]

    ObservableWithGet o = ObservableWithGet.create("hoge");
    System.out.println(o.get());
    o.subscribe((s) -> System.out.println(s));
    // Output: hoge\nhoge
    
[/code]

