Created by Yoshifumi Kawai(neuecc)
UniRx(Reactive Extensions for Unity) is re-implementation of .NET Reactive Extensions. Official Rx is great but can't work on Unity and has some issue of iOS AOT. This library remove there issues and add some specified utility for Unity.
UniRx is available in Unity Asset Store(FREE) - http://u3d.as/content/neuecc/uni-rx-reactive-extensions-for-unity/7tT
Presentation - http://www.slideshare.net/neuecc/unirx-reactive-extensions-for-unityen
Unity Forums support thread, ask me any questions - http://forum.unity3d.com/threads/248535-UniRx-Reactive-Extensions-for-Unity
Ordinary, Unity Network operation use WWW and Coroutine but Coroutine is not good practice for asynchronous operation.
Let me show some example.
- Coroutine can't return result value. (because return type should be IEnumerator)
- It also won't handle exception. (because yield return can't surrond with try-catch) These lack of composability cause Operation close-coupled, and we have to write huge monolithic IEnumerator.
Rx curing asynchronous blues like that. Rx is a library to compose asynchronous and event-based programs using observable collections and LINQ-style query operators.
GameLoop(every Update, OnCollisionEnter, etc), Sensor(like Kinect, Leap Motion, etc) is all of event. Rx considere event as reactive sequence which is possible to compose and perform time-based operations easily by using many LINQ query operators.
Unity is single thread but UniRx helps multithreading for join, cancel, access GameObject etc.
async operation, use ObservableWWW, it's Get/Post returns IObservable.
ObservableWWW.Get("http://google.co.jp/")
.Subscribe(
x => Debug.Log(x), // onSuccess
ex => Debug.LogException(ex)); // onErrorRx is composable, cancelable and can query with LINQ query expressions.
// composing asynchronous sequence with LINQ query expressions
var query = from google in ObservableWWW.Get("http://google.com/")
from bing in ObservableWWW.Get("http://bing.com/")
from unknown in ObservableWWW.Get(google + bing)
select new { google, bing, unknown };
var cancel = query.Subscribe(x => Debug.Log(x));
// Call Dispose is cancel.
cancel.Dispose();Parallel request use Observable.WhenAll
// Observable.WhenAll is for parallel asynchronous operation
// (It's like Observable.Zip but specialized for single async operations like Task.WhenAll)
var parallel = Observable.WhenAll(
ObservableWWW.Get("http://google.com/"),
ObservableWWW.Get("http://bing.com/"),
ObservableWWW.Get("http://yahoo.com/"));
parallel.Subscribe(xs =>
{
Debug.Log(xs[0]); // google
Debug.Log(xs[1]); // bing
Debug.Log(xs[2]); // yahoo
});with progress
// notifier for progress
var progressNotifier = new ScheduledNotifier<float>();
progressNotifier.Subscribe(x => Debug.Log(x)); // write www.progress
// pass notifier to WWW.Get/Post
ObservableWWW.Get("http://google.com/", progress: progressNotifier).Subscribe();Error Handling
// If WWW has .error, ObservableWWW throws WWWErrorException to onError pipeline.
// WWWErrorException has RawErrorMessage, HasResponse, StatusCode, ResponseHeaders
ObservableWWW.Get("http://www.google.com/404")
.CatchIgnore((WWWErrorException ex) =>
{
Debug.Log(ex.RawErrorMessage);
if (ex.HasResponse)
{
Debug.Log(ex.StatusCode);
}
foreach (var item in ex.ResponseHeaders)
{
Debug.Log(item.Key + ":" + item.Value);
}
})
.Subscribe();IEnumerator(Coroutine) is Unity's primitive asynchronous tool. UniRx integrates coroutine and IObservable. Write async in coroutine, Orchestrate many coroutines by UniRx is best way for control asynchronous flow.
// two coroutines
IEnumerator AsyncA()
{
Debug.Log("a start");
yield return new WaitForSeconds(1);
Debug.Log("a end");
}
IEnumerator AsyncB()
{
Debug.Log("b start");
yield return new WaitForEndOfFrame();
Debug.Log("b end");
}
// main code
// Observable.FromCoroutine convert IEnumerator to Observable<Unit>.
// other shorthand, AsyncA().ToObservable()
// after completed AsyncA, run AsyncB as continuous routine.
// UniRx expands SelectMany(IEnumerator) as SelectMany(IEnumerator.ToObservable())
var cancel = Observable.FromCoroutine(AsyncA)
.SelectMany(_ => AsyncB())
.Subscribe();
// you can stop coroutine use subscription's Dispose.
cancel.Dispose();If you needs coroutine with return value, normally we use callback. Observable.FromCoroutine supports convert coroutine to IObservable[T] with cancellation.
// public method
public static IObservable<string> GetWWW(string url)
{
// convert coroutine to IObservable
return Observable.FromCoroutine<string>((observer, cancellationToken) => GetWWWCore(url, observer, cancellationToken));
}
// IObserver is callback publisher
// note: Principles of IObserver is "OnNext* (OnError | Oncompleted)?"
static IEnumerator GetWWWCore(string url, IObserver<string> observer, CancellationToken cancellationToken)
{
var www = new UnityEngine.WWW(url);
while (!www.isDone && !cancellationToken.IsCancellationRequested)
{
yield return null;
}
if (cancellationToken.IsCancellationRequested) yield break;
if (www.error != null)
{
observer.OnError(new Exception(www.error));
}
else
{
observer.OnNext(www.text);
observer.OnCompleted(); // IObserver needs OnCompleted after OnNext!
}
}more application examples, following is multiple OnNext pattern.
public static IObservable<float> ToObservable(this UnityEngine.AsyncOperation asyncOperation)
{
if (asyncOperation == null) throw new ArgumentNullException("asyncOperation");
return Observable.FromCoroutine<float>((observer, cancellationToken) => RunAsyncOperation(asyncOperation, observer, cancellationToken));
}
static IEnumerator RunAsyncOperation(UnityEngine.AsyncOperation asyncOperation, IObserver<float> observer, CancellationToken cancellationToken)
{
while (!asyncOperation.isDone && !cancellationToken.IsCancellationRequested)
{
observer.OnNext(asyncOperation.progress);
yield return null;
}
if (!cancellationToken.IsCancellationRequested)
{
observer.OnNext(asyncOperation.progress); // push 100%
observer.OnCompleted();
}
}
// usecase
Application.LoadLevelAsync("testscene")
.ToObservable()
.Do(x => Debug.Log(x)) // output progress
.Last() // last sequence is load completed
.Subscribe();// Observable.Start is start factory methods on specified scheduler
// default is on ThreadPool
var heavyMethod = Observable.Start(() =>
{
// heavy method...
System.Threading.Thread.Sleep(TimeSpan.FromSeconds(1));
return 10;
});
var heavyMethod2 = Observable.Start(() =>
{
// heavy method...
System.Threading.Thread.Sleep(TimeSpan.FromSeconds(3));
return 10;
});
// Join and await two other thread values
Observable.WhenAll(heavyMethod, heavyMethod2)
.ObserveOnMainThread() // return to main thread
.Subscribe(xs =>
{
// Unity can't touch GameObject from other thread
// but use ObserveOnMainThread, you can touch GameObject naturally.
(GameObject.Find("myGuiText")).guiText.text = xs[0] + ":" + xs[1];
}); UniRx has two extended MonoBehaviour. TypedMonoBehaviour is typesafe MonoBehaviour.
public class Test : TypedMonoBehaviour
{
// all message is overridable, it's typesafe
public override void Update()
{
base.Update();
}
// use Coroutine, use "new" keyword
new public IEnumerator Start()
{
while (true)
{
yield return null;
}
}
}ObservableMonoBehaviour is extended TypedMonoBehaviour. All message is Observable.
public class Test : ObservableMonoBehaviour
{
public override void Awake()
{
// All events can subscribe by ***AsObservable
this.OnMouseDownAsObservable()
.SelectMany(_ => this.UpdateAsObservable())
.TakeUntil(this.OnMouseUpAsObservable())
.Select(_ => Input.mousePosition)
.Repeat()
.Subscribe(x => Debug.Log(x));
// If you use ObservableMonoBehaviour, must call base method
base.Awake();
}
}Use Subject(or AsyncSubject for async operation). For example...
public class LogCallback
{
public string Condition;
public string StackTrace;
public UnityEngine.LogType LogType;
}
public static class LogHelper
{
static Subject<LogCallback> subject;
public static IObservable<LogCallback> LogCallbackAsObservable()
{
if (subject == null)
{
subject = new Subject<LogCallback>();
// Publish to Subject in callback
UnityEngine.Application.RegisterLogCallback((condition, stackTrace, type) =>
{
subject.OnNext(new LogCallback { Condition = condition, StackTrace = stackTrace, LogType = type });
});
}
return subject.AsObservable();
}
}
// method is separatable and composable
LogHelper.LogCallbackAsObservable()
.Where(x => x.LogType == LogType.Warning)
.Subscribe();
LogHelper.LogCallbackAsObservable()
.Where(x => x.LogType == LogType.Error)
.Subscribe();// Unity's singleton UiThread Queue Scheduler
Scheduler.MainThreadScheduler
ObserveOnMainThread()/SubscribeOnMainThread()
// Global StartCoroutine runner
MainThreadDispatcher.StartCoroutine(enumerator)
// push value on every update time
Observable.EveryUpdate().Subscribe();
// push value on every fixedUpdate time
Observable.EveryFixedUpdate().Subscribe();
// delay on frame time
Observable.Return(42).DelayFrame(10);
// convert Coroutine to IObservable
Observable.FromCoroutine((observer, token) => enumerator(observer, token));
// convert IObservable to Coroutine
Observable.Range(1, 10).ToCoroutine();UniRx has AotSafe Utilities.
// create safety iterator
Enumerable.Range(1, 10).AsSafeEnumerable().ToArray();
// elements wrap to class
Enumerable.Range(1, 10).WrapValueToClass(); // IEnumerable<Tuple<int>>
Observable.Range(1, 10).WrapValueToClass(); // IObservable<Tuple<int>>Please see AOT Exception Patterns and Hacks.
This wiki is recommended way for learn Rx. You can understand behavior of all operators by graphical marble diagram.
Introduction to how to use Rx for Game.
Great online tutorial and eBook.
Original project home.
Many Videos and slides and documents.
Unity Forums support thread, ask me any questions - http://forum.unity3d.com/threads/248535-UniRx-Reactive-Extensions-for-Unity
We welcome to your contribute such as bug report, request, and pull request.
At first, see and please write GitHub issues.
Source code is available in Assets/UniRx/Scripts.
This project is using Visual Studio with UnityVS.
Yoshifumi Kawai(a.k.a. neuecc) is software developer in Japan. He is Director/CTO at Grani, Inc. Grani is top social game developer in Japan. He awarded Microsoft MVP for Visual C# since 2011. He is known by creator of linq.js(LINQ to Objects for JavaScript)
Blog: http://neue.cc/ (JPN)
Twitter: https://twitter.com/neuecc (JPN)
This library is under MIT License.