<commit_before>package com.rxt.rxjavasample.rxjava;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;
import android.widget.Button;

import com.rxt.rxjavasample.R;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

public class MainActivity extends AppCompatActivity implements View.OnClickListener {

    private static final String TAG = "Rxjava";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        Button primary = (Button) findViewById(R.id.btn_primary);
        Button mapOperation = (Button) findViewById(R.id.btn_map);
        Button zipbtn = (Button) findViewById(R.id.btn_zip);
        Button sampleFilter = (Button) findViewById(R.id.btn_sample_filter);
        Button flowable = (Button) findViewById(R.id.btn_flowable);
        Button concat = (Button) findViewById(R.id.btn_concat);
        Button allinOne = (Button) findViewById(R.id.btn_all_in_one);
        primary.setOnClickListener(this);
        mapOperation.setOnClickListener(this);
        zipbtn.setOnClickListener(this);
        sampleFilter.setOnClickListener(this);
        flowable.setOnClickListener(this);
        concat.setOnClickListener(this);
        allinOne.setOnClickListener(this);

    }

    @Override
    public void onClick(View v) {
        switch (v.getId()) {
            case R.id.btn_primary:
                primary();
                break;
            case R.id.btn_map:
                mapTest(2);
                break;
            case R.id.btn_zip:
                zipTest();
                break;
            case R.id.btn_sample_filter:
                sampleOrFilter();
                break;
            case R.id.btn_flowable:
                flowableTest();
                break;
            case R.id.btn_concat:
                concatTest();
                break;
            case R.id.btn_all_in_one:
                test();
                break;
        }

    }

    private void test() {
        //distinct
//        Observable.create(new ObservableOnSubscribe<String>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
//                e.onNext("a");
//                e.onNext("a");
//                e.onNext("b");
//                e.onNext("b");
//                e.onNext("c");
//                e.onComplete();
//            }
//        })
//                .distinct()//去重
//                .subscribeOn(Schedulers.newThread())
//                .observeOn(AndroidSchedulers.mainThread())
//                .subscribe(new Consumer<String>() {
//                    @Override
//                    public void accept(@NonNull String s) throws Exception {
//                        Log.e(TAG, "accept: " + s);
//                    }
//                });

        //buffer
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//                        @Override
//                        public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
//                            e.onNext(1);
//                            e.onNext(2);
//                            e.onNext(3);
//                            e.onNext(4);
//                            e.onNext(5);
//                            e.onComplete();
//                        }
//                    })
//                            .buffer(3, 2)//每次取3个事件，下一次拉取事件在前一次的基础上往后推2个
//                .subscribeOn(Schedulers.io())
//                            .observeOn(AndroidSchedulers.mainThread())
//                            .subscribe(new Consumer<List<Integer>>() {
//                        @Override
//                        public void accept(@NonNull List<Integer> integers) throws Exception {
//                            Log.e(TAG, "accept: " + integers.size());
//                            for (Integer i : integers) {
//                                Log.e(TAG, "value: " + i);
//                            }
//                        }
//                });

        //timer,延时发送
//        Log.e(TAG, "startTime: ");
//        Observable.timer(2, TimeUnit.SECONDS)
//                .subscribeOn(Schedulers.newThread())
//                .observeOn(AndroidSchedulers.mainThread())
//                .subscribe(new Consumer<Long>() {
//                    @Override
//                    public void accept(@NonNull Long aLong) throws Exception {
//                        Log.e(TAG, "endTime:");
//                    }
//                });

        //interval,间隔发送long类型的数据，从0l开始
        Observable.interval(0, 2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Long>() {

                    private Disposable md;

                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        md = d;
                    }

                    @Override
                    public void onNext(@NonNull Long aLong) {
                        Log.e(TAG, "accept: " + aLong);
                        if (aLong == 5)
                            md.dispose();
                    }

                    @Override<commit_msg>rxjava操作符实际应用学习中<commit_after>package com.rxt.rxjavasample.rxjava;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;
import android.widget.Button;

import com.rxt.rxjavasample.R;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

public class MainActivity extends AppCompatActivity implements View.OnClickListener {

    private static final String TAG = "Rxjava";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        Button primary = (Button) findViewById(R.id.btn_primary);
        Button mapOperation = (Button) findViewById(R.id.btn_map);
        Button zipbtn = (Button) findViewById(R.id.btn_zip);
        Button sampleFilter = (Button) findViewById(R.id.btn_sample_filter);
        Button flowable = (Button) findViewById(R.id.btn_flowable);
        Button concat = (Button) findViewById(R.id.btn_concat);
        Button allinOne = (Button) findViewById(R.id.btn_all_in_one);
        primary.setOnClickListener(this);
        mapOperation.setOnClickListener(this);
        zipbtn.setOnClickListener(this);
        sampleFilter.setOnClickListener(this);
        flowable.setOnClickListener(this);
        concat.setOnClickListener(this);
        allinOne.setOnClickListener(this);

    }

    @Override
    public void onClick(View v) {
        switch (v.getId()) {
            case R.id.btn_primary:
                primary();
                break;
            case R.id.btn_map:
                mapTest(2);
                break;
            case R.id.btn_zip:
                zipTest();
                break;
            case R.id.btn_sample_filter:
                sampleOrFilter();
                break;
            case R.id.btn_flowable:
                flowableTest();
                break;
            case R.id.btn_concat:
                concatTest();
                break;
            case R.id.btn_all_in_one:
                test();
                break;
        }

    }

    private void test() {
        //distinct
//        Observable.create(new ObservableOnSubscribe<String>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
//                e.onNext("a");
//                e.onNext("a");
//                e.onNext("b");
//                e.onNext("b");
//                e.onNext("c");
//                e.onComplete();
//            }
//        })
//                .distinct()//去重
//                .subscribeOn(Schedulers.newThread())
//                .observeOn(AndroidSchedulers.mainThread())
//                .subscribe(new Consumer<String>() {
//                    @Override
//                    public void accept(@NonNull String s) throws Exception {
//                        Log.e(TAG, "accept: " + s);
//                    }
//                });

        //buffer
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//                        @Override
//                        public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
//                            e.onNext(1);
//                            e.onNext(2);
//                            e.onNext(3);
//                            e.onNext(4);
//                            e.onNext(5);
//                            e.onComplete();
//                        }
//                    })
//                            .buffer(3, 2)//每次取3个事件，下一次拉取事件在前一次的基础上往后推2个
//                .subscribeOn(Schedulers.io())
//                            .observeOn(AndroidSchedulers.mainThread())
//                            .subscribe(new Consumer<List<Integer>>() {
//                        @Override
//                        public void accept(@NonNull List<Integer> integers) throws Exception {
//                            Log.e(TAG, "accept: " + integers.size());
//                            for (Integer i : integers) {
//                                Log.e(TAG, "value: " + i);
//                            }
//                        }
//                });

        //timer,延时发送
//        Log.e(TAG, "startTime: ");
//        Observable.timer(2, TimeUnit.SECONDS)
//                .subscribeOn(Schedulers.newThread())
//                .observeOn(AndroidSchedulers.mainThread())
//                .subscribe(new Consumer<Long>() {
//                    @Override
//                    public void accept(@NonNull Long aLong) throws Exception {
//                        Log.e(TAG, "endTime:");
//                    }
//                });

        //interval,间隔发送long类型的数据，从0L开始, 可用于计时
        Observable.interval(0, 2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Long>() {

                    private Disposable md;

                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        md = d;
                    }

                    @Override
                    public void onNext(@NonNull Long aLong) {
                        Log.e(TAG, "accept: " + aLong);
                        if (aLong == 5)
                            md.dispose();
                    }

                    @Override
{'fraction_non_alphanumeric': 0.1121654501216545, 'fraction_numerical': 0.0025952960259529602, 'mean_word_length': 1.4622603833865815, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7323341', 'n_tokens_mistral': 3417, 'n_tokens_neox': 3211, 'n_words': 550}