필터링 연산자

filter

• 전달 받은 데이터가 조건에 맞는지 확인한 후 결과가 true인 데이터만 통지합니다.
• 파라미터로 받는 Predicate 함수형 인터페이스에서 조건을 확인합니다.
• 연산자 함수는 메서드 체인으로 여러 개를 사용할 수 있습니다.

package com.itvillage.chapter05.chapter0502;

import com.itvillage.common.CarMaker;
import com.itvillage.common.SampleData;
import io.reactivex.Observable;

public class ObservableFilterExample02 {
    public static void main(String[] args) {
        Observable.fromIterable(SampleData.carList)
                .filter(car -> car.getCarMaker() == CarMaker.CHEVROLET)
                .filter(car -> car.getCarPrice() > 30000000)
                .subscribe(car -> System.out.println(car.getCarName()));
    }
}

 

distinct

• 이미 통지된 동일한 데이터가 있다면 이후의 동일한 데이터는 통지 하지 않습니다.

package com.itvillage.chapter05.chapter0502;

import com.itvillage.common.SampleData;
import com.itvillage.utils.LogType;
import com.itvillage.utils.Logger;
import io.reactivex.Observable;

/**
 * 이미 통지한 데이터와 같은 데이터는 제외하고 통지
 * 유일한 값을 처리하고자 할때 사용
 */
public class ObservableDistinctExample01 {
    public static void main(String[] args) {
        Observable.fromArray(SampleData.carMakersDuplicated)
                .distinct()
                .subscribe(carMaker -> Logger.log(LogType.ON_NEXT, carMaker));
    }
}

 

take

• 파라미터로 지정한 개수나 기간이 될 때까지 데이터를 통지합니다.
• 지정한 범위가 통지 데이터보다 클 경우 데이터를 모두 통지하고 완료합니다.

package com.itvillage.chapter05.chapter0502;

import com.itvillage.utils.LogType;
import com.itvillage.utils.Logger;
import io.reactivex.Observable;

/**
 * 지정한 갯수만큼 데이터를 발행
 */
public class ObservableTakeExample01 {
    public static void main(String[] args) {
        Observable.just("a", "b", "c", "d")
                .take(2)
                .subscribe(data -> Logger.log(LogType.ON_NEXT, data));
    }
}

package com.itvillage.chapter05.chapter0502;

import com.itvillage.utils.LogType;
import com.itvillage.utils.Logger;
import com.itvillage.utils.TimeUtil;
import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

/**
 * 지정한 시간동안 데이터를 계속 발행
 */
public class ObservableTakeExample02 {
    public static void main(String[] args) {
        Observable.interval(1000L, TimeUnit.MILLISECONDS)
                .take(3500L, TimeUnit.MILLISECONDS)
                .subscribe(data -> Logger.log(LogType.ON_NEXT, data));

        TimeUtil.sleep(3500L);
    }
}

 

takeUntil

• 파라미터로 지정한 조건이 true가 될 때까지 데이터를 계속 통지합니다.

package com.itvillage.chapter05.chapter0502;

import com.itvillage.common.Car;
import com.itvillage.common.SampleData;
import com.itvillage.utils.TimeUtil;
import io.reactivex.Observable;

/**
 * 파리미터로 지정한 조건이 될 때까지 데이터를 계속 발행
 */
public class ObservableTakeUntilExample01 {
    public static void main(String[] args) {
        Observable.fromIterable(SampleData.carList)
                .takeUntil((Car car) -> car.getCarName().equals("트랙스"))
                .subscribe(car -> System.out.println(car.getCarName()));

        TimeUtil.sleep(300L);
    }
}

 

• 파라미터로 지정한 Observable이 최초 데이터를 통지할 때까지 데이터를 계속 통지합니다.

package com.itvillage.chapter05.chapter0502;

import com.itvillage.utils.LogType;
import com.itvillage.utils.Logger;
import com.itvillage.utils.TimeUtil;
import io.reactivex.Observable;

import java.util.concurrent.TimeUnit;

/**
 * 파라미터로 받은 Flowable/Observable이 최초로 데이터를 발행할 때까지 계속 데이터를 발행
 * timer와 함께 사용하여 특정 시점이 되기전까지 데이터를 발행하는데 활용하기 용이
 */
public class ObservableTakeUntilExample02 {
    public static void main(String[] args) {
        Observable.interval(1000L, TimeUnit.MILLISECONDS)
                .takeUntil(Observable.timer(5500L, TimeUnit.MILLISECONDS))
                .subscribe(data -> Logger.log(LogType.ON_NEXT, data));

        TimeUtil.sleep(5500L);
    }
}

 

skip

• 파라미터로 지정한 숫자만큼 데이터를 건너뛴 후 나머지 데이터를 통지합니다.

package com.itvillage.chapter05.chapter0502;

import com.itvillage.utils.LogType;
import com.itvillage.utils.Logger;
import io.reactivex.Observable;

public class ObservableSkipExample01 {
    public static void main(String[] args) {
        Observable.range(1, 15)
                .skip(3)
                .subscribe(data -> Logger.log(LogType.ON_NEXT, data));
    }
}

 

• 파라미터로 지정한 시간 동안에는 데이터를 통지를 건너뛴 후 지정한 시간 이후 나머지 데이터를 통지합니다.

package com.itvillage.chapter05.chapter0502;

import com.itvillage.utils.LogType;
import com.itvillage.utils.Logger;
import com.itvillage.utils.TimeUtil;
import io.reactivex.Observable;

import java.util.concurrent.TimeUnit;

public class ObservableSkipExample02 {
    public static void main(String[] args) {
        Observable.interval(300L, TimeUnit.MILLISECONDS)
                .skip(1000L, TimeUnit.MILLISECONDS)
                .subscribe(data -> Logger.log(LogType.ON_NEXT, data));

        TimeUtil.sleep(3000L);
    }
}

 

 

 

[참고 자료]

• 인프런- Kevin의 알기 쉬운 RxJava
• https://reactivex.io/documentation/operators.html
• 전체 소스코드