비동기적 코드를 작성하는 것은 정말 어렵습니다. 그러나 Functional reactive 프로그래밍이라면 변수를 다루는 것만큼이나 쉽게 closure를 다룰 수 있어서 코드를 깔끔하게 작성할 수 있습니다. 새로운 라이브러리인 RxSwift는 이벤트 구동형 앱의 유지성과 가독성을 높이고 골칫거리와 버그를 줄여줍니다. Max Alexander가 소개하는 functional reactive programming의 기초와 기능을 확인해 보세요.

소개 (0:00)

최근 Rx에 대한 관심이 뜨겁습니다. Rx는 Observable<Element> 인터페이스로 표현되는 포괄적인 추상화이며, RxSwift는 Rx의 Swift 버전입니다. 어렵게 느껴질 수도 있지만, 쉽게 설명해보겠습니다.

이 코드는 어쩌다가 이런 복잡한 모양이 됐을까요?

Alamofire.request(.POST, "login", parameters: ["username": "max", "password": "insanity"])
  .responseJSON(completionHandler: { (firedResponse) -> Void in
    Alamofire.request(.GET, "myUserInfo" + firedResponse.result.value)
      .responseJSON(completionHandler: { myUserInfoResponse in
        Alamofire.request(.GET, "friendList" + myUserInfoResponse.result.value)
          .responseJSON(completionHandler: { friendListResponse in
            Alamofire.request(.GET, "blockedUsers" + friendListResponse.result.value)
              .responseJSON(completionHandler: {

              })
            })
          })
    Alamofire.request(.GET, "myUserAcccount" + firedResponse.result.value)
      .responseJSON(completionHandler: {
      })
    })

Alamofire 요청을 위한 코드입니다. 잘못된 점이 보이나요? Alamofire란 AFNetworking 처럼 HTTP 요청을 하는 코드입니다. 블록 기반으로 얽혀서 계속 인덴트가 증가하는 코드는 그 결과를 예측할 수 없다는 문제가 있습니다. 네트워크 통신이 실패할 경우는 어떨까요? 에러 핸들러가 있지만 예외를 어디서 처리해야 할지 알 수 없습니다. Rx는 이런 상황을 해결해 줍니다.

기초적인 설명 (2:26)

새로운 이벤트가 있으면 collection도 있죠. [1, 2, 3, 4, 5, 6]와 같은 배열 혹은 리스트의 모양입니다. Swift 방식이라면 filter를 활용해 짝수만 거를 수 있습니다.

[1, 2, 3, 4, 5, 6].filter{ $0 % 2 == 0 }

5를 곱한 후 다시 배열로 만들려면 어떻게 할까요?

[1, 2, 3, 4, 5, 6].map{ $0 * 5 }

전체 합을 구하려면요?

[1, 2, 3, 4, 5, 6].reduce(0, +)

의미가 잘 나타나는 코드입니다. for loop을 돌리지도 않고 중간 상태를 저장하지 않아도 되죠. 거의 Scala나 Haskel과 같은 방식입니다. 오직 배열만을 사용하는 앱은 거의 없습니다. 사용자들이 이미지를 다운받거나 네트워크 통신을 하고 친구를 추가하는 활동을 좋아하므로 인터넷은 필수이고, IO를 아주 많이 사용하게 됩니다. 따라서 기기 메모리를 넘어선 영역 즉, 사용자 상호작용, 다른 기기, 카메라, 디스크 등과 여러 요소와 소통해야 합니다. 이런 비동기적인 활동은 실패할 수 있고 문제 발생 가능성도 높습니다.

Rx 권리장전 (4:09)

이쯤에서 Rx 권리장전을 소개해 드리겠습니다. Rx 권리장전은 이렇게 선언합니다.

우리는 비동기 이벤트를 마치 iterable 컬렉션처럼 간단하게 다룰 권리가 있다.

정말 마음에 쏙 드는 문구죠?

Observables (4:25)

Rx에서 배열 대신 Observables를 생각해 봅시다. Observables은 안전한 형변환이 가능한 이벤트로 다른 종류의 데이터를 넣고 뺄 수 있습니다. 현재 RxSwift는 Beta 3 버전으로 쉽게 설치할 수 있죠. 단지 RxSwift를 적용하기만 하면 됩니다.

pod 'RxSwift', '~> 2.0.0-beta.3'
import RxSwift

just를 사용하면 간단하게 Observable을 생성할 수 있습니다. 어떤 타입의 변수를 넣어도 해당 타입의 Observable을 반환해주죠.

just(1)  //Observable<Int>

배열을 넣고 이벤트를 하나씩 꺼내려면 어떻게 할까요?

[1,2,3,4,5,6].toObservable()  //Observable<Int>

Observable<Int> 타입을 반환받을 수 있습니다.

만약 S3에 업로드하거나 로컬 데이터베이스에 데이터를 저장하는 API가 있다면 다음과 같은 모습일 겁니다.

create { (observer: AnyObserver<AuthResponse>) -> Disposable in


  return AnonymousDisposable {

  }
}

create를 호출하면 블록을 돌려주죠. 이 블록이 Observable를 만들어주는데, 이는 누군가 이 이벤트를 구독할 것이라는 뜻입니다. 당장은 AnonymousDisposable을 무시해도 됩니다. 그 다음 두 줄이 Observable를 만들도록 API를 처리할 곳입니다.

Alamofire와 비슷한 코드를 만들어 봤습니다.

create { (observer: AnyObserver<AuthResponse>) -> Disposable in

  let request = MyAPI.get(url, ( (result, error) -> {
    if let err = error {
      observer.onError(err);
    }
    else if let authResponse = result {
      observer.onNext(authResponse);
      observer.onComplete();
    }
  })
  return AnonymousDisposable {
    request.cancel()
  }
}

로그인을 하거나 GET 요청을 해서 결과와 에러 값을 담은 콜백을 받습니다. 다른 클라이언트 SDK의 코드이기 때문에 API를 고칠 수는 없지만, 반환값을 Observable로 바꿀 수는 있습니다. 에러가 있다면 observer.onError()를 부르는데, 이는 구독하고 있는 상대방에게 뭔가 실패했음을 알려주는 것입니다. 실제 응답을 받으면 observable.onNext()를 부릅니다. 다음으로 뭔가 끝났다면 onComplete()를 부르죠. 이제 AnonymousDisposable를 살펴볼까요? AnonymousDisposable는 뷰 컨트롤러를 떠났거나, 서비스를 종료해서 더이상 해당 요청을 할 필요가 없을 경우처럼, 어떤 인터럽트를 받았을때 부르는 액션입니다. 비디오 업로드처럼 큰 작업을 할 때 유용하죠. request.cancel()을 하면 모든 작업을 끝냈을 때 할당됐던 자원을 제거할 수 있습니다.

observables 구독하기 (8:11)

observable 생성법을 알아봤으니 이제 observable을 구독하는 방법을 알아볼까요? 아래처럼 배열을 만든 후, 여러 다른 객체들을 부를 수 있는 확장 함수인 toObservable()를 부릅니다. 리스너 함수를 벌써 완성했네요.

[1,2,3,4,5,6]
  .toObservable()
  .subscribeNext {
    print($0)
  }

iterable 처럼 예약 구독 리스너 이벤트는 요청 실패 상황이나 다음 이벤트, 혹은 onCompleted 상황에서 여러 다른 정보를 줄 수 있습니다. 필요할 때 구독하기만 하면 됩니다.

[1,2,3,4,5,6]
  .toObservable()
  .subscribe(onNext: { (intValue) -> Void in
    // Pumped out an int
  }, onError: { (error) -> Void in
    // ERROR!
  }, onCompleted: { () -> Void in
    // There are no more signals
  }) { () -> Void in
    // We disposed this subscription
  }

Observables 결합하기 (9:14)

Rx를 사용한다는 것은 소켓 서비스를 사용하는 것과 비슷합니다. 현제 계좌 잔고를 보여주는 UI가 있고 주식 시세 표시기를 구독하는 웹소켓 서비스를 생각해보죠. 주식 시세 표시기가 여러 다른 이벤트를 보여주면 사용자가 매입할 수 있는지 표시해줍니다. 계좌 잔고가 낮으면 구매 버튼을 비활성화하고, 매입자의 가격 범위에 들어오면 활성화하는 기능이죠.

func rx_canBuy() -> Observable<Bool> {
  let stockPulse : [Observable<StockPulse>]
  let accountBalance : Observable<Double>

  return combineLatest(stockPulse, accountBalance,
    resultSelector: { (pulse, bal) -> Bool in
    return pulse.price < bal
  })
}

combineLatest는 두 개의 최근 이벤트를 결합합니다. 계좌 잔고 보다 주식 시세 표시기의 가격이 낮은지에 따라 블록이 작용합니다. 즉, 해당 주식을 살 수 있다는 뜻이죠. 이를 활용해 두 observable을 결합하고 조건을 충족하는지 판정하는 로직을 세울 수 있습니다. 결과로는 Bool 타입의 observable 값을 반환합니다.

rx_canBuy()
  .subscribeNext { (canBuy) -> Void in
    self.buyButton.enabled = canBuy
  }

rx_canBuy 메서드는 구독한 상대방에게 boolean값을 줍니다. 따라서 self.buyButton 값은 canBuy 값과 같죠.

이번엔 병합 예제를 살펴볼까요? 사용자가 선호하는 주식 시세 표시기를 보여주는 UI를 가진 앱을 생각해 보죠. Apple과 Google, Johnson & Johnson의 주식을 구독할 건데요. 이들 주식 시세 표시기는 각기 다른 소켓 엔드 포인트를 갖습니다. 주식 시세 표시기가 업데이트될 때마다 UI를 업데이트를 하도록 해보죠.

let myFavoriteStocks : [Observable<StockPulse>]

myFavoriteStocks.merge()
  .subscribeNext { (stockPulse) -> Void in
    print("\(stockPulse.symbol)/
      updated to \(stockPulse.price)/")
  }

모두 StockPulse 타입의 Observable로, 이 중 어느 하나가 업데이트돼도 알 수 있으려면 어떻게 할까요? Observable 배열을 사용하고, 여러 다른 주식 시세 표시기를 하나의 스트림으로 병합해서 구독하면 됩니다.

Rx Observables 다이어그램 (18:03)

Rx를 접한지 오래 됐지만 아직도 많은 operator를 잊곤 하고, 자주 공식 문서를 참조하곤 합니다. rxmarbles.com을 참고하면 이들 여러 이벤트의 이론적 컴퍼넌트를 확인할 수 있어 도움이 됩니다.

손쉬운 백그라운드 프로세스 (19:03)

RxSwift의 멋진 기능 하나를 소개해 드릴까 합니다. 대용량의 비디오 업로드를 백그라운드 프로세스에서 하려고 할 때 가장 효율적인 방법은 observeOn를 사용하는 것입니다.

let operationQueue = NSOperationQueue()
  operationQueue.maxConcurrentOperationCount = 3
  operationQueue.qualityOfService = NSQualityOfService.UserInitiated
  let backgroundWorkScheduler
    = OperationQueueScheduler(operationQueue: operationQueue)

videoUpload
  .observeOn(backgroundWorkScheduler)
  .map({ json in
    return json["videoUrl"].stringValue
  })
  .observeOn(MainScheduler.sharedInstance)
  .subscribeNext{ url
    self.urlLabel.text = url
  }

비디오 업로드 작업을 하면서 완료 비율을 계속 받을 수 있습니다. 이런 작업을 메인 스레드에서 할 필요는 없겠죠. 백그라운드 작업 스케줄러에게 맡기는게 좋습니다. 비디오 업로드가 모두 완료되면 UI label에 넣을 URL이 담긴 JSON이 반환됩니다. 백그라운드 작업이므로 UI에 업데이트하도록 알려야 하고, 메인 스레드로 전달돼야 합니다. observeOn(MainScheduler.SharedInstance)를 사용하면 UI를 업데이트할 수 있습니다. 안드로이드의 RxJava에서는 크래시가 발생하겠지만, 이와 달리 Swift에서 백그라운드 스레드에 뭔가 넘겨주면 경고가 발생하지 않는다는 점은 좀 아쉽습니다.

여태까지는 빙산의 일각에 불과합니다. (20:31)

이벤트를 배열처럼 활용해서 코드를 더 쉽고 멋지게 개선하는 RxSwift의 멋진 기능도 소개하겠습니다. MVVM은 뷰 컨트롤러를 보다 조직적인 세트로 바꿔주는 모델입니다. RxSwift는 동일 repository에 RxCocoa라는 자매 라이브러리를 갖고 있는데, 기본적으로 Rx-Text나 네임필드처럼 UI 뷰를 위해 모든 Cocoa 클래스를 위해 확장 메서드를 제공합니다. 구독할 대상을 줄일 수 있고 Observable을 다른 관점의 값과 결합할 수 있습니다.

멀티 플랫폼 (22:49)

세상에는 정말 많은 플랫폼이 있죠. Rx의 매력은 IO 작업을 위한 어떤 클라이언트 API도 신경쓰지 않아도 된다는 점입니다. 안드로이드나 JavaScript로 개발한다면 비동기 IO 이벤트를 만들기 위해 각각 다른 방법을 익혀야 할 겁니다. Rx는 .NET, Java, JavaScript, Swift 등 보다 대중적인 언어에 적용할 수 있는 도우미 라이브러리 그룹으로, 동일한 operator와 같은 논리를 사용해서 코드를 작성할 수 있죠. 각기 다른 언어이지만 유사하게 보입니다. Swift로 작성된 로그를 한 번 볼까요?

func rx_login(username: String, password: String) -> Observable<Any> {
  return create({ (observer) -> Disposable in
    let postBody = [
      "username": username,
      "password": password
    ]
    let request = Alamofire.request(.POST, "login", parameters: postBody)
      .responseJSON(completionHandler: { (firedResponse) -> Void in
        if let value = firedResponse.result.value {
          observer.onNext(value)
          observer.onCompleted()
        } else if let error = firedResponse.result.error {
          observer.onError(error)
        }
      })
    return AnonymousDisposable{
      request.cancel()
    }
  })
}

값에 대한 observable을 반환해주는 rx_login이 보입니다. Kotlin 버전을 볼까요?

fun rx_login(username: String, password: String): Observable<JSONObject> {
  return Observable.create ({ subscriber ->
    val body = JSONObject()
    body.put("username", username)
    body.put("password", password)
    val listener = Response.Listener<JSONObject>({ response ->
      subscriber.onNext(response);
      subscriber.onCompleted()
    })
    val errListener = Response.ErrorListener { err ->
      subscriber.onError(err)
    }
    val request = JsonObjectRequest(Request.Method.POST, "login", listener, errListener);
    this.requestQueue.add(request)
  });
}

정말 비슷해보이죠? TypeScript 버전도 한 번 보겠습니다.

rx_login = (username: string, password: string) : Observable<any> => {
  return Observable.create(observer => {
    let body = {
      username: username,
      password: password
    };
    let request = $.ajax({
      method: 'POST',
      url: url,
      data: body,
      error: (err) => {
        observer.onError(err);
      },
      success: (data) => {
        observer.onNext(data);
        observer.onCompleted();
      }
    });
    return () => {
      request.abort()
    }
  });
}

역시 정말 비슷해 보입니다. 여러 종류의 이벤트를 이해하려는 노력 없이도 테스트 케이스를 작성할 수 있습니다. Rx를 사용하면 모든 클라이언트쪽 코드나 UI 코드를 모두 작성할 필요 없이 동일한 원칙을 어느 곳에나 사용할 수 있습니다.

다음: RxSwift 예제로 감잡기 : RxSwift 시작을 위한 간단한 예제들