[UniRx] Reactive Extension - 2 IObservable의 형 변환

개요
IObserable<T>의 운용에 있어서 가장 기초적인 이해를 위해 T가 의미하는 바와 그 운용방법을 정리하고자 한다.

본론
IObservable<T>는, Rx에 의해 정의된 데이터 스트림의 인터페이스이다.
앞의 포스팅의 기본 정의를 보면 이벤트가 총 3개가 있는것을 알 수가 있다.

OnNext(data);
OnError(exception);
OnCompleted();

OnCompleted에 대해서는 이렇다할 데이터가 넘어오지 않는다.
이는 어디까지나 데이터의 흐름이 끊겼음을 알려주는 신호일 뿐이다.

단순하게 Coroutine의 대용으로 생각해 버리기에는 Rx의 탄생배경과 그 근간이 조금 다르다.
Coroutine은 비동기의 흐름을 하나의 thread에서 수행하기 위해 본래 1개의 callstack을 임의로 분할해(yield에 의해) 운용하는 것으로, 비동기 처리에 초점이 맞추어진것이라 한다면.

Rx의 경우 어떤 데이터가 지속적으로 변하는것을 전제로 생각되어진, 그 변화에 반응하는 반응형 프로그래밍 기법이라 할 수 있을것 같다.

이러한 반응성 프로그래밍을 하기 위해서 사용되고 있는 대표적인 기법이 Observable pattern이고, IObservable<T>는 그 인터페이스라 할 수 있다.

IObservable<T>는 T가 지속적으로 변화할 데이터이며, 이를 감시하겠다는 것을 뜻한다.

따라서 이 Observer를 subscribe한다는 말은 어떤 감시 룰을 실제로 운용하겠다는 것을 의미한다. 이 Observe라는 행위를 통해, IObservable<T>는 운용이 된다.

기본적으로 generic인 이 인터페이스를 운용하려 할때, 원하는 데이터가 변화할때가 있다. 예를들면 JSON Parsing라던가가 그 예다. 이미 generic에 의해 타입이 지정되어 있는 이 type의 observable에 대해, 다른 타입으로 어떻게 변환하는가에 대한 의문을 가질 수 있을듯 하다.

예를 들어서 ObservableWWW.Get()에 의해 웹서버에 있는 json파일을 가져와 이를 parse한다 생각해보자.

uFrame에서 공식적으로 채택되어 있는 JSON parser는 simpleJSON이므로, simpleJSON를 기준으로 예를 들어볼까 한다. 다른 implementation을 사용한다고 하더라도, 이해하는데 큰 차이는 없을거라 생각된다.

var owww = ObservableWWW.Get(url).Select(src=> JSON.Parse(src));

위와 같은 Chaining으로 변환이 가능하다.

이후에 owww를 subscribe하면 간단하게 parsing이 완료된 데이터를 얻을 수 있다.

ObservableWWW.Get(url)은, IObservable<string>이다.
여기서 Select라는 LINQ구문의 확장을 사용하면 src에 대해 가공을 실시할 수 있고, 그 결과 IObservable<JSONNode>가 반환된다.

[UniRx] Reactive Extension - 1 : Observable 직접 작성하기

개요

UniRx를 사용함에 앞서 가장 기본이 되는 Observable작성방법을 생각보다 찾기 힘들었던 관계로 글로 남겨두고자 한다.

본론

Rx는 Promise++ specs를 따르고 있다고 한다.
https://gist.github.com/staltz/868e7e9bc2a7b8c1f754

Promise/A+사양인 bluebird의 경우, 아래와 같은 형태의 Promise를 정의하는 것이 가능하다.

// Definition

var Promise = require("bluebird");
var example = new Promise(function(resolve,reject) {
    try {
    // something do
    var some_result_value = ...;
    // then
    resolve(some_result_value);
    } catch(ex) {
    reject(ex);
    }
});

// Usage

example
.then(function(some_result_value) {
// some_result_value is coming.
}).catch(function(ex) {
// ex is coming.
});

Rx의 경우 Promise의 일종이기때문에 비슷하게 정의를 할 수가있는데
위와 같은 정의를 하자면

// Definition

var example = Observable.Create<T>(observer=> {

    try {
       // something do
       T some_result_value = ...;
       observer.OnNext(some_result_value);
    } catch(ex) {
        observer.OnError(ex);
    }
    observer.OnCompleted();

    return Disposable.Create(()=> {
         // something dispose here if required
    });

});

// Usage

example.Subscribe(some_result_value=> {
      // some_result_value is coming
},ex=> {
      // ex is coming
},()=> {
      // OnCompleted calling
});

위와 같이 사용할 수 있다.

ex를 알기 쉽게 하기 위해 try catch를 넣었으나, 비동기 코드를 작성하려 할 경우 일반적인 비동기 운용에 기반한 코드를 작성하고, 필요한 시점에 위에 적힌것과 같은 이벤트 호출을 실시하면 된다.

혹은 간단하게 Coroutine용 IEnumerator function을 작성한 후, Observable.FromCoroutine()을 사용하는 방법도 있다.