RxJava, RxKotlin - 스케줄러를 활용하여 콜백 지옥 벗어나기

[출처 - RxJava 프로그래밍 : 리액티브 프로그래밍 기초부터 안드로이드 까지 한번에]

본 글은 'RxJava 프로그래밍 : 리액티브 프로그래밍 기초부터 안드로이드 까지 한번에' 를 학습하면서 정리한 글입니다.

 

Rx프로그래밍에는 다양한 스케줄러가 있습니다. 이 스케줄러를 활용하면 code가 훨씬 간결해집니다. 그것은 서버와 연동하는 비동기 code를 작성할 때도 마찬가지 입니다.

 

예제code를 보겠습니다. (이 code는 okhttp3 기반이므로 build.gradle에 

   implementation 'com.squareup.okhttp3:okhttp:3.2.0'

를 추가해주어야 합니다.)

Java

import okhttp3.*;

import java.io.IOException;

public class http_test {

    public static void main(String[] args){
        final String URL_README = "https://raw.githubusercontent.com/yudong80/reactivejava/master/README.md";
        final OkHttpClient client = new OkHttpClient();
        Request request = new Request.Builder()
                .url(URL_README)
                .build();
        client.newCall(request).enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                e.printStackTrace();
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
               System.out.println(response.body().string());
            }
        });
    }
}

kotlin

import okhttp3.*
import java.io.IOException

fun main(){

    val URL_README = "https://raw.githubusercontent.com/yudong80/reactivejava/master/README.md"
    val client = OkHttpClient()
    val request = Request.Builder()
        .url(URL_README)
        .build()
    client.newCall(request).enqueue(object : Callback {
        override fun onFailure(call: Call, e: IOException) {
            e.printStackTrace()
        }

        @Throws(IOException::class)
        override fun onResponse(call: Call, response: Response) {
            println(response.body().string())
        }
    })
}

결과

WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by okhttp3.internal.Platform (file:/C:/Users/PJW/.gradle/caches/modules-2/files-2.1/com.squareup.okhttp3/okhttp/3.2.0/f7873a2ebde246a45c2a8d6f3247108b4c88a879/okhttp-3.2.0.jar) to field sun.security.ssl.SSLSocketFactoryImpl.context
WARNING: Please consider reporting this to the maintainers of okhttp3.internal.Platform
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
Welcome to Java Reactive Programming!!

위아 같은 결과가 나옵니다.( WARRING 보다는 밑에 WelWelcome to Java Reactive Programming!! 가 나온거에 주목! 왜냐면 저걸 보는게 목적이었기 때문입니다.)

위의 code는 단순하게 http 통신으로 주소에 있는 데이터를 GET 하는 것입니다.

그럼 위의 code를 쫌더 확장시켜 보겠습니다.

 

Java

import okhttp3.*
import java.io.IOException

fun main(){
    val client = OkHttpClient()
    val FIRST_URL = "https://api.github.com/zen"
    val SECOND_URL = "https://raw.githubusercontent.com/yudong80/reactivejava/master/samples/callback_hell"
    val request = Request.Builder()
        .url(FIRST_URL)
        .build()
    client.newCall(request).enqueue(object : Callback {
        override fun onFailure(call: Call, e: IOException) {
            e.printStackTrace()
        }

        @Throws(IOException::class)
        override fun onResponse(call: Call, response: Response) {
            println(response.body().string())

            //add callback again
            val request = Request.Builder()
                .url(SECOND_URL)
                .build()
            client.newCall(request).enqueue(onSuccess)
        }
    })

}
private val onSuccess = object : Callback {
    override fun onFailure(call: Call, e: IOException) {
        e.printStackTrace()
    }

    @Throws(IOException::class)
    override fun onResponse(call: Call, response: Response) {
        println(response.body().string())
    }
}

kotlin

import okhttp3.*
import java.io.IOException

val FIRST_URL = "https://api.github.com/zen"
val SECOND_URL = "https://raw.githubusercontent.com/yudong80/reactivejava/master/samples/callback_hell"
fun main(){
    val client = OkHttpClient()
    val request = Request.Builder()
        .url(FIRST_URL)
        .build()
    client.newCall(request).enqueue(object : Callback {
        override fun onFailure(call: Call, e: IOException) {
            e.printStackTrace()
        }

        @Throws(IOException::class)
        override fun onResponse(call: Call, response: Response) {
            println(response.body().string())

            //add callback again
            val request = Request.Builder()
                .url(SECOND_URL)
                .build()
            client.newCall(request).enqueue(onSuccess)
        }
    })

}
private val onSuccess = object : Callback {
    override fun onFailure(call: Call, e: IOException) {
        e.printStackTrace()
    }

    @Throws(IOException::class)
    override fun onResponse(call: Call, response: Response) {
        println(response.body().string())
    }
}

결과

WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by okhttp3.internal.Platform (file:/C:/Users/PJW/.gradle/caches/modules-2/files-2.1/com.squareup.okhttp3/okhttp/3.2.0/f7873a2ebde246a45c2a8d6f3247108b4c88a879/okhttp-3.2.0.jar) to field sun.security.ssl.SSLSocketFactoryImpl.context
WARNING: Please consider reporting this to the maintainers of okhttp3.internal.Platform
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
Approachable is better than simple.
Welcome to Callback Hell!!

 

위의 FIRST_URL에서 데이터를 get하고 SECOND_URL에서 데이터를 get하는 code입니다. 여기서 문제점은 FIRST_URL의onResponse에서 SECOND_URL을 요청한다는 것입니다. 다시 말해 callback에서 요청하고 또 callback에서 그 결과를 받게 된다는 것입니다. 이런 문제는 javascript로 coding을 하다보면 많이 격는 callback 지옥이라 불리는 현상인데, callback에서 또 callback을 또 callback을 부르게 되는 현상을 말합니다.

이 현상은 code를 복잡하게 할 뿐만 아니라 유지보수도 힘들게 하는 경향이 있습니다.(이건 제 경험입니다.)

 

그래서 이부분은 다음과 같이 간단하게 처리할 수 있습니다. (CommonUtils와 CommonUtilsk는 여기서 확인 하실 수 있습니다.)

Java

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;

import java.io.IOException;

public class Callback_haven_test {
    public static final OkHttpClient client = new OkHttpClient();
    public static void main(String[] args){
        final String FIRST_URL = "https://api.github.com/zen";
        final String SECOND_URL = "https://raw.githubusercontent.com/yudong80/reactivejava/master/samples/callback_hell";
        CommonUtils.exampleStart();
        Observable<String> source = Observable.just(FIRST_URL)
                .subscribeOn(Schedulers.io())
                .map(it->get(it))
                .concatWith(Observable.just(SECOND_URL)
                        .map(it->get(it)));
        source.subscribe(it-> {
            long time = System.currentTimeMillis() - CommonUtils.startTime;
            System.out.println(CommonUtils.getThreadName() + " | " + time + " | " + "value = " + it);
        });
        CommonUtils.sleep(5000);
        CommonUtils.exampleComplete();
    }
    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) {
            System.out.println(e.getMessage());
            throw e;
        }
    }

}

Kotlin

import io.reactivex.Observable
import io.reactivex.schedulers.Schedulers
import okhttp3.OkHttpClient
import okhttp3.Request
import java.io.IOException
val client = OkHttpClient()
fun main(){

    val FIRST_URL = "https://api.github.com/zen"
    val SECOND_URL = "https://raw.githubusercontent.com/yudong80/reactivejava/master/samples/callback_hell"
    CommonUtilsk.exampleStart()
    val source = Observable.just(FIRST_URL)
        .subscribeOn(Schedulers.io())
        .map { it -> get(it) }
        .concatWith(
            Observable.just(SECOND_URL)
                .map { it -> get(it) })
    source.subscribe { it ->
        val time = System.currentTimeMillis() - CommonUtilsk.startTime
        println(CommonUtilsk.getThreadName() + " | " + time + " | " + "value = " + it)
    }
    CommonUtilsk.sleep(5000)
    CommonUtilsk.exampleComplete()
}
@Throws(IOException::class)
fun get(url: String): String {
    val request = Request.Builder()
        .url(url)
        .build()
    try {
        val res = client.newCall(request).execute()
        return res.body().string()
    } catch (e: IOException) {
        println(e.message)
        throw e
    }

}

결과

WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by okhttp3.internal.Platform (file:/C:/Users/PJW/.gradle/caches/modules-2/files-2.1/com.squareup.okhttp3/okhttp/3.2.0/f7873a2ebde246a45c2a8d6f3247108b4c88a879/okhttp-3.2.0.jar) to field sun.security.ssl.SSLSocketFactoryImpl.context
WARNING: Please consider reporting this to the maintainers of okhttp3.internal.Platform
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
RxCachedThreadScheduler-1 | 546 | value = Practicality beats purity.
RxCachedThreadScheduler-1 | 1246 | value = Welcome to Callback Hell!!

위의 code 들과 같이 더 간단하게 coding 할 수 있습니다.

위의 code는 FIRST_URL 호출과 SECOND_URL이 한눈에 보입니다. 여기서 사용된 concatWith는 concat()과 비슷한데(concat은 여기서 확인해 보세요). concat는 첫번째 Observable과 두번째 Observable을 함꼐 인자로 넣어야 하지만, concatWith는 현재의 Observable에 새로운 Observable을 결합하는 차이가 있습니다.

concatWith를 통하여 첫번째와 두번째를 한눈에 볼 수 이도록 coding 하였습니다. reqeust는 get으로 따로 만들어서 사용하고 있습니다.

 

이렇게 변경을 함으로서 순수한 비즈니스 로지고가 비동기 동작을 휘한 스레드 부분을 구별할 수 있고, 가독성을 높여 줍니다.

 

그리고 결과의 첫번쨰 value는 계속 변화합니다.

 

마지막으로 concatWith 말고 Observable을 2개를 만들고 zip 함수를 사용하여서 실행 할 수 도 있습니다.