UniRx - Reactive Extensions for Unity

Yeah, Subject is native component(some Rx operator uses it internally), Rx operators is high level component.
and Subject is native 'event' expression of Rx.

If you want to write all codes under Reactive, Functional way, you should not use Subject.
But if you write under pragmatic way, Subject may helps you.

Of course primitive parts should avoid basically.
At first, chain the operators.
Second, use ReactiveProperty(wrapper of Subject) it is easy to use.
Third, IEnumerator is your friend, FromCoroutine(and ToYieldInstruction) sometimes useful.
Finally, use Subject as .NET Event for notification path.

 

Thanks for information,I am digging into ReactivyProperty and seems damn impressive...BTW Advanture Capitalist (big milestone for UniRx) game uses UniRx heavily and I found this example of generating event and subscribing from another class is quite useful too, you can have look at it here, ( 10 mins 10 secs)

 

@neuecc Is it possible to integrate UniRx with MEC (memory efficient coroutines) ? https://www.assetstore.unity3d.com/en/#!/content/54975

Since above asset saves from hell of garbage and increases performance....

 

You can create `Observable.FromMemoryEfficientCoroutine` like built-in `Observable.FromCoroutine`.

BTW, I'm planning improvement performance of ObserveEveryValueChanged, currently it make coroutine each observe properties.
I'll create central dispatcher of observe properties, it is almost same idea as MEC.

 

That would be beautiful, when can we expect it?

 

Hi, I have a question:
Is there any way to get or set a reactive property dynamically?
I mean, by using a string of the property`s name, to access the reactive property.
For regular properties, someComponent.getType().GetProperty("name of the property") can get a property info and then do the job, but it seems not working with reactive properties.

 

"Value" is target property.

Code (CSharp):

1. var rxprop = new ReactiveProperty<string>();
2. rxprop.GetType().GetProperty("Value").SetValue(rxprop, "foobar");
3.  
4. Debug.Log(rxprop.Value); // foobar

Code (CSharp):

1.  
2. // ----
3.     // same as rxprop.Value.Bar = "B A R";
4.     // but it does not propagate value changes.
5.  
6.     var rxprop = new ReactiveProperty<MyClass>();
7.     rxprop.Value = new MyClass();
8.  
9.     var v = rxprop.GetType().GetProperty("Value").GetValue(rxprop);
10.     typeof(MyClass).GetProperty("Bar").SetValue(v, "B A R");
11.     Debug.Log(rxprop.Value.Bar); // B A R
12.  
13. // ----
14.  
15. class MyClass
16. {
17.     public int Foo { get; set; }
18.     public string Bar { get; set; }
19. }

 

Thanks for replying, I was misled by the name "ReactiveProperty", it turned out that they are regarded as fields. And actually what I wanted to know was how to dynamically access the reactive property before considering getting and setting values. Now I can get the fieldInfo of rxprop.Take your code example, I can use var _rxprop = component.GetType().GetField("rxprop"); . Then the part _rxprop.GetValue(certainComponent) actually works fine. But I could not figure out a way to set value with the reflection _rxprop. Besides, I still want to set the value with the propagation of the change. I tried to use _rxprop.SetValue(certainComponent, new ReactiveProperty<MyClass>(newValue)) assuming we have a constructor for this reactive property. It instantiates rxprop again, it did change the value, but it loses all the subscription, indeed, the object reference is changed by doing so.

 

Subscription is managed by each ReactiveProperties.
You should use "Value" PropertyInfo.

Code (CSharp):

1. public class X
2. {
3.     ReactiveProperty<int> rxprop;
4. }
5.  
6. ---
7.  
8. var component = new X() { rxprop = new ReactiveProperty<int>(100); };
9. component.rxprop.Subscribe(x => Debug.Log(x)); // output 100 in console
10.  
11. var fieldInfo = component.GetType().GetField("rxprop");
12. var rxProp = fieldInfo.GetField(component);
13.  
14. rxProp.GetType().GetProperty("Value").SetValue(rxProp, 10000); // output 10000 in console

 

Thanks for all your help

 

i have a problem,code is below:

Code (CSharp):

1. ReactiveProperty<int>  Exp=new ReactiveProperty<int>();
2. ReactiveProperty<int>  Level=new ReactiveProperty<int>(1);
3. Exp.Where(exp=>exp>Level.Value*100).Subscibe( exp=>{Exp.Value-=Level.Value*100;Level.Value++;});
4. Exp.Subscribe(exp=>Debug.Log("Exp:"+exp));
5. Level.Subscribe(level=>Debug.Log("Level:"+level));
6. Exp.Value+=70;
7. Exp.Value+=70;
8.  

Output:
Exp:0
Level:1
Exp:70
Exp:40
Level:2
Exp:140

Exp actually is 40!! it received the older value! can you help me?
​

 

if you wrote at Line:8, Debug.Log(Exp.Value); // 40

Line:6, Exp.Value += 70 is publish 70 to all subscribers(Line:3 and Line:4), Line:3 was filtered, Line 4 write log "70".
Line:7, Exp.Value += 70 is publish 140 to all subscribers, Line:3 publish Exp.Value -= 100 is publish to all subscribers, write Log "40", after Line3 finished, Line:4 write Log "140".

 

i know the reason of this result,but i think ReactiveProperty should stop publishing the older value when a new value is arrived.how do u think?

 

I don't think so.
The problem is from recrursive change.
Sorry, you only control it manually.

 

thanks for all your replies,can you tell me the reason of control it manunlly? forgive my poor english!

 

thanks for all your replies,can you tell me the reason of control it manunlly? forgive my poor english!

 

ReactiveProperty is primitive that have to work as you wrote.
Controlling cancel value is work of operators.

 

I would make the experience the primary variable and have level derive from it. And do not reset the XP value.

Code (CSharp):

1. public class LevelProgression : MonoBehaviour
2. {
3.     void Start()
4.     {
5.  
6.         var currentExperience = new ReactiveProperty<int>(0);
7.  
8.         var currentLevel = currentExperience
9.             .Select(xp => 1 + Mathf.Floor(xp / 100))
10.             .Do(level => Debug.Log("Current level: " + level))
11.             .ToReactiveProperty();
12.  
13.         var cutOffExperience = currentExperience
14.             .Select(xp => xp % 100)
15.             .Do(xp => Debug.Log("XP for current level: " + xp))
16.             .ToReactiveProperty();
17.  
18.         currentExperience.Value += 70;
19.         currentExperience.Value += 70;
20.     }
21. }

Results in:

Code (CSharp):

1. Current level: 1
2. XP for current level: 0
3.  
4. Current level: 1
5. XP for current level: 70
6.  
7. Current level: 2
8. XP for current level: 40

Cheers,
Rob

 

Hi, I've upload UniRx 5.3 to AssetStore.

This release introduce MicroCoroutine, it is memory efficient and 10x faster than standard coroutine at run many coroutines.
Some coroutine based operators has been changed to use MicroCoroutine.

Add : ReactiveCommand
Add : MainThreadDispatcher.StartUpdateMicroCoroutine, StartFixedUpdateMicroCoroutine, StartEndOfFrameMicroCoroutine
Add : Scheduler.MainThreadFixedUpdate, MainThreadEndOfFrame
Add : ToYieldInstruction(cancellationToken)
Add : Observer.Create(onNext/onNext, onError/onNext, onCompleted) overload
Add : IReadOnlyReactiveProperty.SkipLatestValueOnSubscribe
Add : Observable.WhenAll overload (IObservable(params IObservable[] sources), this becomes breaking changes)
Add : Observable.FromMicroCoroutine
Add : Observable.AsSingleUnitObservable
Add : Observable.SubscribeWithState
Add : Observable.CreateWithState
Add : Disposable.CreateWithState
Improvement : Use MicroCoroutine on ObserveEveryValueChanged, EveryUpdate, EveryFixedUpdate,EveryEndOfFrame, NextFrame, TimerFrame, IntervalFrame, DelayFrame, SampleFrame, ThrottleFrame,ThrottleFirstFrame, TimeoutFrame
Improvement : Performance improvement for Observable.Range, Repeat when scheduler is Scheduler.Immediate
Improvement : Use Time.unscaledDeltaTime in IgnoreTimeScaleMainThreadScheduler
Fix : ReadOnlyReactiveProperty(source, initialValue) does not publish initial value on subscribe
Fix : IReadOnlyCollection has set indexer
Fix : Ambigious property of IReactiveCollection.Count, Indexer
Fix : Throw invalid error when ObservableWWW.LoadFromCacheOrDownload failed.
Breaking Changes : Added IReadOnlyReactiveProperty.HasValue
Breaking Changes : AsyncConvertsion scheduler to Scheduler.MainThread on WebGL build(WebGL doesn't support ThreadPool)
Other : Update UniRxAnalyzer 1.4.0.1 https://www.nuget.org/packages/UniRxAnalyzer

 

Hi there,

First, thanks a lot for your plugin, it seems very promising
I'm diving into this reactive functional programming thing, and the first hiccup I encounter is that I can't find an equivalent of Observable.Generate in UniRX, is it something that has been renamed or just not yet implemented ?
http://www.introtorx.com/content/v1.0.10621.0/04_CreatingObservableSequences.html#ObservableGenerate

Anyway thanks a lot for giving this stuff for free

 

Thank you for use.
Yes, Observable.Generate is not implemented yet.

 

Hi, I've made class reference.

https://github.com/neuecc/UniRx/wiki

My markdown generator is not perfect and slightly strange format, but maybe... helpful for you.

 

Thank you very much Do you have any plans for implementing the Generate method soon ?

 

I've wrote the issue for implement Generate method.
https://github.com/neuecc/UniRx/issues/105#issuecomment-221145973

 

Hi, I've released new framework PhotonWire
https://github.com/neuecc/PhotonWire
Unity Client implementation uses UniRx.

PhotonWire implements RPC Style(or MVC's Controller implementation) client-server, server-server message transportation.
Server code uses C# 5.0's async/await, fully asynchronous.
Client code uses UniRx - Reactive Extensions for Unity so everything asynchronous.
Message is encoded MsgPack(fast, small serialization format) and in Unity is used high performance pre-generated serializer.

PhotonWire supports Visual Studio 2015's analyzer which helps writing constraints.
Of course Visual Studio debugger supports, too.

 

Hi @neuecc, thanks for writing UniRx! I just started using it today and can't figure out what is probably a very simple thing. The setup is the following, I set the HerdController on a gameObject, and then I set the Hero from another gameObject in the editor.

The problem is that the subscription on the HerdController gets lost, whereas the what looks to be the same subscription
on the object holding the FloatReactiveProperty works fine. Can you see a problem in this setup?

Code (CSharp):

1. public class HerdController: MonoBehaviour {
2.  
3.     public Hero hero;
4.  
5.     void Start() {
6.         hero.observableSpeed.Subscribe (
7.             x => {
8.             Debug.Log("Value changed to = " + x);
9.         }).AddTo(this);
10.     }
11. }
12.  
13. public class Hero : MonoBehaviour {
14.  
15.     public FloatReactiveProperty observableSpeed { get; private set; }
16.  
17.     void Awake() {
18.         observableSpeed = new FloatReactiveProperty(-1);
19.     }
20.  
21.     void Start() {
22.         observableSpeed.Subscribe (
23.             x => {
24.             Debug.Log("Hero value changed to = " + x);
25.         }).AddTo(this);
26.  
27.      SetSpeed(0);
28.     }
29.  
30.     void SetSpeed(float speed) {
31.         Debug.Log("Setting speed = " + speed);
32.         observableSpeed.Value = speed;
33.     }
34. }
35.  
36. //Prints out
37. Value changed to = -1
38. Hero Value changed to = -1
39. Setting Speed = 0
40. Value changed to = 0
41. Hero Value changed to = 0
42.  
43. //Calling SetSpeed(0.75) nothing gets printed out from the HerdController
44. Setting Speed = 0.75
45. Hero Value changed to = 0.75
46.  
47.  

 

I think you should subscribe to a reactive property insead of a property that can get a reactive property

 

Hey @Yinmo thanks for your reply, I just figured out what the problem was. There's nothing wrong with that code, for some reason I had 2 of the same Hero components added to one gameObject . That's fixed for me now, cheers!

 

Try using https://docs.unity3d.com/ScriptReference/DisallowMultipleComponent.html

 

@neuecc Do you think this book is good to help me learn UniRx? https://www.manning.com/books/reactive-extensions-in-dot-net

 

I don't read it yet, but seems good.

 

Hi, today accepted UniRx 5.4 to AssetStore.
This release supports Unity 5.4.0 f3!

```
Add: Observable.FrameInterval
Add: Observable.FrameTimeInterval
Add: Observable.BatchFrame
Add: Observable.Debug(under UniRx.Diagnostics namespace)
Add: ObservableParticleTrigger and OnParticleCollisionAsObservable, OnParticleTriggerAsObservabl(after Unity 5.4) extension methods
Add: UniRx.AsyncReactiveCommand
Add: ReactiveCommand.BindToOnClick, IObservable<bool>.BindToButtonOnClick
Add: UniRx.Toolkit.ObjectPool, AsyncObjectPool
Add: UniRx.AsyncMessageBroker, asynchronous variation of MessageBroker
Add: ObserveEveryValueChanged(IEqualityComparer) overload
Add: Observable.FromCoroutine(Func<CancellationToken, IEnumerator>) overload
Add: ObservableYieldInstruction.IsDone property
Add: IPresenter.ForceInitialize(object argument)
Improvement: Where().Select(), Select().Where() peformance was optimized that combine funcs at internal
Improvement: MicroCoroutine performance was optimized that prevent refresh spike
Improvement: Observable.Return performance was optimized that reduced memory cost
Improvement: Observable.Return(bool) was optimzied perofmrance that allocate zero memory
Improvement: Observable.ReturnUnit was optimzied perofmrance that allocate zero memory
Improvement: Observable.Empty was optimzied perofmrance that allocate zero memory
Improvement: Observable.Never was optimzied perofmrance that allocate zero memory
Improvement: Observable.DelayFrame performance was optimized
Improvement: UnityEqualityComparer.GetDefault peformance was optimized
Improvement: AddTo(gameObject) dispose when ObservableTrigger is not activated
Improvement: AddTo(gameObject/component) performance was optimized by use inner CompositeDisposable of ObservableDestroyTrigger
Improvement: FromCoroutine<T>(Func<IObserver<T>, IEnumerator>) stops coroutine when subscription was disposed
Improvement: ReactiveCollection, ReactiveDictionary implements dispose pattern
Fix: ToYieldInstruction throws exception on MoveNext when reThrowOnError and has error
Fix: ObserveEveryValueChanged publish value immediately(this is degraded from UniRx 5.3)
Fix: Prevent warning on Unity 5.4 at ObservableMonoBehaviour/TypedMonoBehaviour.OnLevelWasLoaded
Fix: Remove indexer.set of IReadOnlyReactiveDictionary
Breaking Changes: Does not guaranty MicroCoroutine action on same frame
Breaking Changes: UniRx.Diagnostics.LogEntry was changed from class to struct for performance improvement
```

Some new features ReadMe: https://github.com/neuecc/UniRx/commit/823e0b553900be42ba16fe9cdeb47f741227ef2d
Reference: https://github.com/neuecc/UniRx/wiki

 

Thank you for this amazing library, it saves me a lot of time in current project!

 

Thanks for creating this awesome library! I'm wondering how would you go about creating an observable from a callback based API? For example like this (this is in RxSwift).

 

You can use Observable.Create(Safe) or use AsyncSubject.

Code (CSharp):

1. Observable.CreateSafe(observer =>
2. {
3.       runRequest(callbackResult =>
4.       {
5.             observer.OnNext(callbackResult); observer.OnCompleted();
6.      });
7.       return Disposable.Create(()=> someCancelMethod() );
8. });

 

Thanks! Do you have an example using AsyncSubject? What is the difference between Create and CreateSafe?

 

Sample of AsyncSubject.
AsyncSubject is run immediately(hot), Create(Safe) is run deferred(cold).

Code (CSharp):

1. IObservable<SomeResult> RequestAsync()
2. {
3.     var subject = new AsyncSubject<SomeResult>();
4.  
5.     runRequest(callBackResult =>
6.     {
7.         subject.OnNext(callBackResult);
8.         subject.OnCompleted();
9.     });
10.  
11.     return subject;
12. }

CreateSafe - Dispose subscription(returns IDisposable) when OnNext is failed.
Create - Does not dispose subscription.

CreateSafe recommend for cold observable(for resource safety)
Create recommend for hot observable(needs to keep event subscribe for exception durability)

 

ok got it, thanks!

 

it seems that i am doing something wrong trying to figure out ReactiveProperties

In one script i am trygin to create reactive property and update it every loop

Code (CSharp):

1.  
2.     public ReactiveProperty<float> Time;
3.     void Awake()
4.     {
5.         Time= new ReactiveProperty<float>(0);
6.         Observable.EveryUpdate().Subscribe(_ => Time= new ReactiveProperty<float>(Time.time));
7.     }
8.  

in other script i am trying to read the property but it doesnt work:

Code (CSharp):

1. Inputs.Instance.Time.Subscribe(val => Debug.Log("Time: " + val));

 

Do not change original reference, it include subscribers, if create new ReactiveProperty, lost theres.
You can change `.Value` property.

Code (CSharp):

1. public readonly ReactiveProperty<float> Time;
2.  
3. void Awake()
4. {
5.     Time= new ReactiveProperty<float>(0);
6.     Observable.EveryUpdate().Subscribe(_ => Time.Value = Time.time);
7. }

 

What is the '_' (underline character) for on the left side of the lambda operator (it isn't used anywhere on the right side)?
_ => Time.Value = Time.time

 

'_' is unused argument in lambda.
This is convention of .NET community.

 

@neuecc Thank you for authoring and maintaining this... and for keeping it open-source!

Do you have a PayPal account I can donate to?

 

Is it the same as this? () => Time.Value = Time.time

 

Thank you, but I'm glad only with your kindness.

 

Yes but Func<T> can not use "() =>" so use "_ => " instead.

 

Thanks. BTW. Do you ever attend the Tokyo Unity3D meetup, or any of the indie game events in Tokyo like Tokyo Indies?

 

Ah, I'm not very good at English, especially listening and speaking...
But sometimes I speak at events in Tokyo.

2016-09-13, Photon Seminar https://atnd.org/events/80089
- Talk about PhotonWire, multiplayer framework for Unity with Photon and UniRx - https://github.com/neuecc/PhotonWire

2016-10-01, .NET Fringe Japan 2016 http://dotnetfringe-japan.connpass.com/event/35659/
- Talk about my 30 OSS libraries(includes UniRx)

 

Hello @neuecc,

Why is the stream of ReactiveProperty only publishing distinct until changed values while a ReadOnlyReactiveProperty observable is publishing all values?

See below for an example:

Code (CSharp):

1. using UnityEngine;
2. using UniRx;
3.  
4. public class ReactivePropertyTest : MonoBehaviour
5. {
6.     private void Start()
7.     {
8.         var subject = new Subject<int>();
9.  
10.         ReactiveProperty<int> property = subject
11.             .ToReactiveProperty();
12.  
13.         property
14.             .Subscribe(item => Debug.Log("property: " + item));
15.  
16.         ReadOnlyReactiveProperty<int> readonlyProperty = subject
17.             .ToReadOnlyReactiveProperty();
18.  
19.         readonlyProperty
20.             .Subscribe(item => Debug.Log("readonlyProperty: " + item));
21.  
22.         subject.OnNext(1);
23.         subject.OnNext(1);
24.     }
25. }

Results in:

I guess the behaviour of the ReactiveProperty conforms to the standard practice of INotiftyPropertyChanged implementations. But why is then the ReadOnlyReactiveProperty behaviour different? I thought the only difference is that you cannot the set the value directly with ReadOnlyReactiveProperty?

Or are ReactiveProperty and ReadOnlyReactiveProperty each intended for completely different purposes?

I do prefer the behaviour of ReadOnlyReactiveProperty. I can always use .DistinctUntilChanged() if I only need to observe changes in the stream.

Cheers,
Rob

 

Hi @neuecc,

Do you have any suggestions as to how to implement a pattern with variable delays between subscription items?
For example, something like (pseudocode)

Code (CSharp):

1. var list = new List<MyObject>() {...};
2. list.ToObservable().RepeatUntilDestroy(this.gameObject).Subscribe(async obj =>
3. {
4.     await obj.TimeToDelay;
5.     Debug.Log("Just waited " + obj.TimeToDelay);
6. });

Obviously, async is not available, but is there any other way to accomplish the same type of behavior? Meaning, each item in the stream determines a variable delay during observation.

Thread.Sleep(...) locks up the UI (obviously), and it seems like there should be a way to do this via yield, but I cannot figure out the proper way to do that within the context of the Subscribe block.