[RxJava] 생성 연산자 정리

 

 

[2021-04.16 업데이트]

[2020.12.01 블로그 포스팅 스터디 첫 번째 글]

RxJava 의 연산자에 대해 나중에 쉽게 찾고 활용할 수 있도록 간단한 설명과 함께 포스팅을 해보려고 합니다. ㅎㅎ

 

 

[생성 연산자] : 생성 연산자의 역할은 Observable, Single 등과 같은 데이터 흐름을 만드는 것 입니다. (just(), fromArray(), fromIterable, onNext(), onError(), onComplete() 등이 있는데 이건 제외해서 남기겠습니다.)

 

 

 

 

1. interval()

-> 일정 시간 간격으로 데이터 흐름을 생성합니다. 기본적인 스케줄러는 SchedulerSupport.COMPUTATION으로 계산 스케줄러에서 실행 됩니다.

함수 파라미터로는 

(long period, TumeUnit unit) 
(long initialDelay, long period, TimeUnit unit)

예시
Observable.interval(1000L, TimeUnit.MILLISECONDS)

 

 

---

 

 

2. timer()

->1번 interval()과 유사하지만 한 번만 실행되는 함수입니다. 일정 시간이 지난 후에 한 개의 데이터를 발행하고 omComplete() 이 발생합니다.

Observable.timer(1000L, TimeUnit.MILLISECONDS)

 

 

---

 

 

3. range(n, m)

-> 주어진 n 값 부터 m 개의 Integer 객체를 발행합니다. interval()은 Long 객체를 반환하지만 이거는 Integer !,

interval() 과의 또 하나의 차이점은 스케줄러에서 실행되지 않고 현재 스레드에서 실행됩니다.

Observable.range(1, 10) // 1에서 10까지의 숫자를 생성

 

 

 

---

 

 

4. intervalRange()

interval() 과 range()를 혼합한 함수입니다. range() 함수처럼 시작 숫자 n 부터 m 개 만큼의 값만 생성하고 onComplete() 이벤트가 발생하는데 interval() 처럼 시간 간격을 줄 수 있습니다.

 

Observable.intervalRange(1, //시작값(start)
5, //개수(count)
1000L, //초기 지연시간(initialDelay)
1000L, //시간간격(period)
TimeUnit.MILLISECONDS) //시간단위(unit)

 

 

 

 

---

 

 

5. defer()

timer()와 비슷하지만 데이터 흐름 생성을 구독자가 subscribe() 함수를 호출할 때까지 미룰 수 있습니다. (defer라는 의미 그대로 !)

밑 사진을 보면 알 수 있듯이 Observable(빨간색) 생성이 구독할 때까지 미뤄지기 때문에 최신 데이터를 얻을 수 있습니다. 기본적으로 메인 스레드에서 실행됩니다.

 

 

 

---

 

 

6. repeat()

함수 이름처럼 반복 실행을 합니다. 서버와 통신할 때 해당 서버가 잘 살아있는지 확인하는데 잘 사용합니다. (Ping, HeartBeat) 

 

Hear beat 예시

public class OkHttpHelper {
	private static OkHttpClient client = new OkHttpClient();
	public static String ERROR = "ERROR";
	
	public static String get(String url) throws IOException { 
		Request request = new Request.Builder()
		        .url(url)
		        .build();
		try {
			Response res = client.newCall(request).execute();
			return res.body().string();
		} catch (IOException e) {
			Log.e(e.getMessage());
			throw e;
		} 
	}

	public static String getT(String url) throws IOException { 
		Request request = new Request.Builder()
		        .url(url)
		        .build();
		try {
			Response res = client.newCall(request).execute();
			return res.body().string();
		} catch (IOException e) {
			Log.et(e.getMessage());
			throw e;
		} 
	}
	
	public static String getWithLog(String url) throws IOException { 
		Log.d("OkHttp call URL = " + url);
		return get(url);
	}	
}

public void heartbeatV1() { 
		CommonUtils.exampleStart();
		String serverUrl = "https://api.github.com/zen";
		
		Observable.timer(2, TimeUnit.SECONDS) 		//2초 간격으로 서버에 ping 날리기
			.map(val -> serverUrl)
			.map(OkHttpHelper::get)
			.repeat()
			.subscribe(res -> Log.it("Ping Result : " + res));	
		CommonUtils.sleep(10000);
		CommonUtils.exampleComplete();
	}

 

 

4초마다 시간 업데이트 예시

Observable.timer(4, TimeUnit.SECONDS)
                .repeat()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe {
                    _date.value = getTimestamp()
                    _time.value = getTimestamp()
                }

 

---

참고 :

RxJava 프로그래밍

reactivex.io/documentation/operators.html

ReactiveX - Operators

github.com/yudong80/reactivejava/tree/master/src/main/java/com/yudong80/reactivejava/chapter04/create

yudong80/reactivejava

위 사이트에서 예제들을 볼 수 있습니다.

 

 

 

 

 

앞으로도 시간 여유가 생기면 안드로이드, 코틀린, rxJava에 대해 포스팅하려고합니다.

댓글과 공감은 큰 힘이 됩니다. 감사합니다. !!