4768


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]

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


CC BY-SA 3.0
2015-01-17T19:40:14.457
4764
2015-01-19T03:25:29.370
2015-01-18T05:52:39.710
3639
635
question
java
Observable<T>の T を取り出す方法


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


CC BY-SA 3.0
2015-01-18T01:43:54.420
4768
2015-01-19T03:25:29.370
2015-01-19T03:25:29.370
4978
4978
4764
answer


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

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

[code]

    return s.toBlocking().first();
    
[/code]

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


CC BY-SA 3.0
2015-01-19T02:18:51.367
4808
2015-01-19T02:18:51.367
2757
4764
answer