[UniRx] Reactive Extension 4 : Array -> Observable

개요
배열을 Observable로 변환하는 방법

수순
IObservable<int> ob = new int[]{1,2,3,4}.ToObservable();

[UniRx] Reactive Extension 3 : DistinctUntilChanged, Throttle

개요

게임의 옵션메뉴에 있어서, 설정한 값들을 바로 저장, 적용하며 과도한 저장 작업이 일어나지 않도록, 메모리상의 데이터만 변경을 실시간으로 실시하며, 일정시간 변경이 없을때에 한정하여 저장작업을 실시하는 로직을 UniRx를 이용하여 표현 해 보고자 한다.

소스

   Observable  
                .EveryUpdate ()  
                .Select (_ => _hashConfigs (optionControl))  
                .DistinctUntilChanged ()  
                .Do(_=> {  
                     // update volume data on distinct.   
                     this._updateVolume(optionControl);  
                })  
                .Throttle (new TimeSpan(0,0,1))  
                .Do (_ => {  
                     // throttled value will be stored in 1 sec  
                     this._storeData(optionControl);  
                })  

updateVolume() 에서는, 내부적으로 구현되어있는 AudioManager에 대하여 볼륨을 적용하는 로직이 포함되어 있다.

optionControl은 ViewModel이다.

_hashConfigs는 ViewModel이 가지는 당시 시점에서의 데이터값의 hash값이다.
예를들어 MasterVolume,BGMVoulme,SEVolume이란 세가지 property가 있다고 가정한다면 이하와 같이 구현된다.

 int _hashConfigs(OptionControlViewModel optionControl) {  
           return      optionControl.MainVolume.GetHashCode () ^  
                     optionControl.SEVolume.GetHashCode () ^  
                     optionControl.BGMVolume.GetHashCode ()     ;  
      }  

상기 로직이 의미하는 바는

EveryUpdate() : 매 프레임마다 체크
.Select(_=> _hashConfigs(optionControl)) : 해싱값을 검사
.DistinctUntilChanged() : 변동이 있는 경우만을 추려서
.Do : 실시
.Throttle() : 일정 기간(1초) 동안 입력이 있었던 자료중의 마지막 하나.
.Do : 실시.

위와 같은 느낌으로 설명될 수 있다.

첫번째 Do의 경우, Distinct를 통해 중복값 만을 추려낸, 연속 입력에 개의치 않는 값이 주어지므로, storage에 기록하는 내용을 제외한 실질적인 볼륨 적용의 로직이 적용되는 것이 적절하다.

Throttle에서 연속입력이 추려지고 나면, 마지막의 한번에 대해서만 storage에 저장하는 로직을 적용한다면 무의미하게 많은 저장을 실시해 높은 비용을 지불하지 않고도 적절한 타이밍에 저장을 할 수 있다.

[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()을 사용하는 방법도 있다.