[Unity] UniRx

UniRx란?

- .NET Reactive Extension을 유니티에 맞게 개량

- Linq 스타일 쿼리 연산자를 사용하여 비동기 및 이벤트 기반 프로그램을 위한 라이브러리

- 스트림 생성, 오퍼레이터로 가공, 구독 사용

 

스트림 (Stream)

- 연속적인 이벤트(메시지)의 흐름

- 분기, 병합 가능 / IObservable<T>로 구독 가능

namespace System
{
    public interface IObserver<in T>
    {
        void OnCompleted();
        void OnError(Exception error);
        void OnNext(T value);
    }
    
    public interface IObservable<out T>
    {
        IDisposable Subscribe(IObserver<T> observer);
    }
    
    public interface IDisposable
    {
        void Dispose();
    }
}

IObserver : 메시지 발행을 위한 인터페이스

- OnCompleted() : 스트림이 완료됨을 통지

- OnError(Exception error) : 에러 발생 시, 예외 통지 후 구독 종료

- OnNext() : 일반적인 메시지 

 

IObservable : 메시지 구독을 위한 인터페이스

- Subscribe(IObserver<T> observer) : 메시지 구독을 위한 메서드

 

IDisposable : 메시지 구독 취소를 위한 인터페이스

- Dispose() : 구독 종료를 위한 메서드

 

Subject<T>

- 구독, 취소, 메시지 발행

namespace UniRx
{
    // Event를 대체하는 Subject
    public interface ISubject<TSource, TResult> : IObserver<TSource>, IObservable<TResult> {}
    public interface ISubject<T> : ISubject<T, T>, IObserver<T>, IObservable<T> {}
    
    public sealed class Subject<T> : ISubject<T>, IDisposable, IOptimizedObservable<T> { ... }
}

Unity Event -> Observable

testBtn.OnClickAsObservable().Subscribe(
    _ => { Debug.Log("OnNext()"); },
    _ => { Debug.LogError("OnError()"); },
    () => { Debug.Log("OnCompleted()"); }
    );

Update -> Observable

// 정적 Update 스트림 ( 수동으로 해제해야 함 )
var observable = Observable.EveryUpdate();

// gameObject의 Destroy시점에 OnCompleted
var obervable1 = this.UpdateAsObservable();
var obervable2 = gameObject.UpdateAsObservable();

// Dispose될 때 OnCompleted ( 프레임 사이 변화만 감지 )
testBtn.ObserveEveryValueChanged(x => x.colors)
    .Subscribe(_ => Test());

// 위 코드와 동일
this.UpdateAsObservable()
    .Select(_ => testBtn.colors)
    .DistinctUntilChanged()
    .Subscribe(_ => Test());

ReactiveProperty

- 값이 변할때마다 스트림을 통해 OnNext 메시지 발생

private ReactiveProperty<int> _reactiveProperty = new ReactiveProperty<int>();
public ReadOnlyReactiveProperty<int> CurrTime => _reactiveProperty.ToReadOnlyReactiveProperty();

public void Test()
{
    CurrTime.Subscribe(
        time => { Debug.Log($"시간 : {time}"); },
        () => { Debug.Log("구독 종료"); });
}

스트림의 구독 (Subscribe) 

- 스트림은 Subscribe된 순간에 생성됨

- OnError, OnComplete 발생 시 구독 종료

 

스트림의 종료시점 (OnComplete)

- 구독은 OnCompleated 호출되는 시점까지 지속됨

- 직접 IDisposable 인터페이스의 Dispose()를 호출한 경우 OnCompleted()가 호출되지 않음

 

1) 게임 오브젝트 연결 시

- Destroy 시점 ( ObservableTriggers )

 

2) Mono가 아닌 C# 객체일 경우

- GC에 의해 Dispose됨

 

3) 오퍼레이터를 통해 조건을 설정한 경우

- 조건 만족 시 호출됨 ( TakeWhile, TakeUntil, etc )

 + 주의) Repeat를 같이 사용한 경우 무한 루프가 될 수 있음

 

4) 정적 스트림 구독 시

- 수동으로 구독 취소를 해주어야 함

+ AddTo로 오브젝트에 연결시킬 경우 연결된 게임 오브젝트가 파괴될 때 Dispose됨

Observable.EveryUpdate().Subscribe(x => Debug.Log(x)).AddTo(this);
Observable.IntervalFrame(30).Subscribe(x => Debug.Log(x)).AddTo(gameObject);

Observable의 타입

1) Cold Observable

- 구독(Subscribe)될때마다 새롭게 생성되어, 별도의 스트림이 됨

- 구독하기 전까지 메시지를 발행하지 않고, Observer가 없으면 동작하지 않음

+ 대부분의 스트림은 Cold

var stream = Observable.Interval(TimeSpan.FromSeconds(1));
stream.Subscribe(x => Debug.Log($"1 : {Time.time.ToString()}"));
yield return new WaitForSeconds(0.1f);
stream.Subscribe(x => Debug.Log($"2 : {Time.time.ToString()}"));
yield return new WaitForSeconds(0.1f);
stream.Subscribe(x => Debug.Log($"3 : {Time.time.ToString()}"));

2) Hot Observable

- 구독 여부에 관계없이 메시지를 발행함

- 스트림을 분기하거나 메시지를 분배하는 것이 가능
+ 구독 여부에 관계 없이 메시지 발행하여 구독 시기에 따라 메시지를 발행받지 못할 수 있음

+ Publish()를 통해 Hot으로 변경 가능

var stream = Observable.Interval(TimeSpan.FromSeconds(1)).Publish().RefCount();
stream.Subscribe(x => Debug.Log($"1 : {Time.time.ToString()}"));
yield return new WaitForSeconds(0.1f);
stream.Subscribe(x => Debug.Log($"2 : {Time.time.ToString()}"));
yield return new WaitForSeconds(0.1f);
stream.Subscribe(x => Debug.Log($"3 : {Time.time.ToString()}"));

 

+ 코루틴 <-> 스트림 변환 가능

- Observable.FromCoroutine

- Observable.FromCoroutineValue<T>

- FromCoroutine<T>

- StartAsCoroutine<T>

 

참고자료 ( 데브쿠키 박민근 - UniRx 소개 )

https://www.slideshare.net/slideshow/160402-unirx/60399953