Rxjava(2) 变换操作

文档网址：https://mcxiaoke.gitbooks.io/rxdocs/content/

        mData.add("1");
        mData.add("2");
        mData.add("3");
        mData.add("4");
        mData.add("5");
        mData.add("6");
        mData.add("7");



        private void window(){
        /*
        ------------------
        1
        2
        3
        ------------------
        4
        5
        6
        ------------------
        7
         */
        Observable.fromIterable(mData).window(3).subscribe(new Consumer<Observable<String>>() {
            @Override
            public void accept(Observable<String> stringObservable) throws Exception {
                Log.e("qwer","------------------");
                stringObservable.subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e("qwer",s);
                    }
                });
            }
        });
    }

    private void scan(){
        /*
         1
         12
         123
         1234
         12345
         123456
         1234567
         */
        Observable.fromIterable(mData).scan(new BiFunction<String, String, String>() {
            @Override
            public String apply(String s, String s2) throws Exception {
                return s + s2;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.e("qwer", s);
            }
        });
    }

    private void map(){
        Observable.fromIterable(mData).map(new Function<String, Integer[]>() {
            @Override
            public Integer[] apply(String s) throws Exception {
                Integer d = Integer.valueOf(s);
                return new Integer[]{d, d + 100};
            }
        }).subscribe(new Observer<Integer[]>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer[] integers) {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void groupBy(){
        Observable.fromIterable(mData).groupBy(new Function<String, Integer>() {
            @Override
            public Integer apply(String s) throws Exception {
                if(Integer.valueOf(s) < 4){
                    return 10;
                }else{
                    return 11;
                }
            }
        }).subscribe(new Observer<GroupedObservable<Integer, String>>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            /*
             10  1
             10  2
             10  3
             11  4
             11  5
             11  6
             11  7
             */
            @SuppressLint("CheckResult")
            @Override
            public void onNext(GroupedObservable<Integer, String> integerStringGroupedObservable) {
                if(integerStringGroupedObservable.getKey() == 10){
                    integerStringGroupedObservable.subscribe(new Consumer<String>() {
                        @Override
                        public void accept(String s) throws Exception {
                            Log.e("qwer", 10 + "  " + s);
                        }
                    });
                }else{
                    integerStringGroupedObservable.subscribe(new Consumer<String>() {
                        @Override
                        public void accept(String s) throws Exception {
                            Log.e("qwer", 11 + "  " + s);
                        }
                    });
                }
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void flatMap(){
        Observable.fromIterable(mData).flatMap(
                new Function<String, ObservableSource<Integer>>() {
                    @Override
                    public ObservableSource<Integer> apply(String s) {
                        Integer d = Integer.valueOf(s);
                        Integer[] ds = new Integer[]{d, d + 100};
                        return Observable.fromArray(ds);

                    }
                }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.e("qwer", integer + "");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void buffer() {
        Observable.fromIterable(mData).buffer(3).subscribe(new Observer<List<String>>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(List<String> strings) {
                /**
                 3
                 3
                 1
                 */
                Log.e("qwer", strings.size() + "");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }