#36 rxjava: filtering

RxJava/practice/src/main/java/org/example/ex09/ObservableDistinctEx01.java

@@ -0,0 +1,18 @@
+package org.example.ex09;
+
+import io.reactivex.rxjava3.core.Observable;
+import org.example.common.SampleData;
+import org.example.utils.LogType;
+import org.example.utils.Logger;
+
+/**
+ * 이미 통지한 데이터와 같은 데이터는 제외하고 통지
+ * 유일한 값을 처리하고자 할때 사용
+ */
+public class ObservableDistinctEx01 {
+    public static void main(String[] args) {
+        Observable.fromArray(SampleData.carMakersDuplicated)
+                .distinct()
+                .subscribe(carMaker -> Logger.log(LogType.ON_NEXT, carMaker));
+    }
+}

RxJava/practice/src/main/java/org/example/ex09/ObservableDistinctEx02.java

@@ -0,0 +1,16 @@
+package org.example.ex09;
+
+import io.reactivex.rxjava3.core.Observable;
+import org.example.common.CarMaker;
+import org.example.common.SampleData;
+import org.example.utils.LogType;
+import org.example.utils.Logger;
+
+public class ObservableDistinctEx02 {
+    public static void main(String[] args) {
+        Observable.fromArray(SampleData.carMakersDuplicated)
+                .distinct()
+                .filter(carMaker -> carMaker == CarMaker.SSANGYOUNG)
+                .subscribe(carMaker -> Logger.log(LogType.ON_NEXT, carMaker));
+    }
+}

RxJava/practice/src/main/java/org/example/ex09/ObservableDistinctEx03.java

@@ -0,0 +1,19 @@
+package org.example.ex09;
+
+import io.reactivex.rxjava3.core.Observable;
+import org.example.common.Car;
+import org.example.common.SampleData;
+import org.example.utils.LogType;
+import org.example.utils.Logger;
+
+/**
+ * 객체의 특정 필드를 기준으로 distinct 하는 예제
+ */
+public class ObservableDistinctEx03 {
+    public static void main(String[] args) {
+        Observable.fromIterable(SampleData.carList)
+                .distinct(Car::getCarMaker)
+                .subscribe(car -> Logger.log(LogType.ON_NEXT, car.getCarName()));
+
+    }
+}

RxJava/practice/src/main/java/org/example/ex09/ObservableFilterEx01.java

@@ -0,0 +1,15 @@
+package org.example.ex09;
+
+import io.reactivex.rxjava3.core.Observable;
+import org.example.common.CarMaker;
+import org.example.common.SampleData;
+import org.example.utils.LogType;
+import org.example.utils.Logger;
+
+public class ObservableFilterEx01 {
+    public static void main(String[] args) {
+        Observable.fromIterable(SampleData.carList)
+                .filter(car -> car.getCarMaker() == CarMaker.CHEVROLET)
+                .subscribe(car -> Logger.log(LogType.ON_NEXT, car.getCarMaker() + " : " + car.getCarName()));
+    }
+}

RxJava/practice/src/main/java/org/example/ex09/ObservableFilterEx02.java

@@ -0,0 +1,14 @@
+package org.example.ex09;
+
+import io.reactivex.rxjava3.core.Observable;
+import org.example.common.CarMaker;
+import org.example.common.SampleData;
+
+public class ObservableFilterEx02 {
+    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()));
+    }
+}

RxJava/practice/src/main/java/org/example/ex09/ObservableSkipEx01.java

@@ -0,0 +1,13 @@
+package org.example.ex09;
+
+import io.reactivex.rxjava3.core.Observable;
+import org.example.utils.LogType;
+import org.example.utils.Logger;
+
+public class ObservableSkipEx01 {
+    public static void main(String[] args) {
+        Observable.range(1, 15)
+                .skip(3)
+                .subscribe(data -> Logger.log(LogType.ON_NEXT, data));
+    }
+}

RxJava/practice/src/main/java/org/example/ex09/ObservableSkipEx02.java

@@ -0,0 +1,18 @@
+package org.example.ex09;
+
+import io.reactivex.rxjava3.core.Observable;
+import org.example.utils.LogType;
+import org.example.utils.Logger;
+import org.example.utils.TimeUtil;
+
+import java.util.concurrent.TimeUnit;
+
+public class ObservableSkipEx02 {
+    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);
+    }
+}

RxJava/practice/src/main/java/org/example/ex09/ObservableTakeEx01.java

@@ -0,0 +1,16 @@
+package org.example.ex09;
+
+import io.reactivex.rxjava3.core.Observable;
+import org.example.utils.LogType;
+import org.example.utils.Logger;
+
+/**
+ * 지정한 갯수만큼 데이터를 발행
+ */
+public class ObservableTakeEx01 {
+    public static void main(String[] args) {
+        Observable.just("a", "b", "c", "d")
+                .take(2)
+                .subscribe(data -> Logger.log(LogType.ON_NEXT, data));
+    }
+}

RxJava/practice/src/main/java/org/example/ex09/ObservableTakeEx02.java

@@ -0,0 +1,21 @@
+package org.example.ex09;
+
+import io.reactivex.rxjava3.core.Observable;
+import org.example.utils.LogType;
+import org.example.utils.Logger;
+import org.example.utils.TimeUtil;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 지정한 시간동안 데이터를 계속 발행
+ */
+public class ObservableTakeEx02 {
+    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);
+    }
+}

RxJava/practice/src/main/java/org/example/ex09/ObservableTakeUntilEx01.java

@@ -0,0 +1,19 @@
+package org.example.ex09;
+
+import io.reactivex.rxjava3.core.Observable;
+import org.example.common.Car;
+import org.example.common.SampleData;
+import org.example.utils.TimeUtil;
+
+/**
+ * 파리미터로 지정한 조건이 될 때까지 데이터를 계속 발행
+ */
+public class ObservableTakeUntilEx01 {
+    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);
+    }
+}

RxJava/practice/src/main/java/org/example/ex09/ObservableTakeUntilEx02.java

@@ -0,0 +1,22 @@
+package org.example.ex09;
+
+import io.reactivex.rxjava3.core.Observable;
+import org.example.utils.LogType;
+import org.example.utils.Logger;
+import org.example.utils.TimeUtil;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 파라미터로 받은 Flowable/Observable이 최초로 데이터를 발행할 때까지 계속 데이터를 발행
+ * timer와 함께 사용하여 특정 시점이 되기전까지 데이터를 발행하는데 활용하기 용이
+ */
+public class ObservableTakeUntilEx02 {
+    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);
+    }
+}