r3,cysharp

The behavior of `ThrottleFirst`/`ThrottleFirstFrame` is not consistent with UniRx

I found that the behavior of ThrottleFirst/ThrottleFirstFrame is not consistent with UniRx.

ThrottleFirst/ThrottleFirstFrame of R3 does not emit a value until timeSpan/frameCount has elapsed, even if it is the first value, while those of UniRx does not wait for timeSpan/frameCount when emitting the first value.

var startingTime = DateTime.Now;
Observable.Timer(TimeSpan.FromMilliseconds(100), TimeSpan.FromMilliseconds(100))
    .ThrottleFirst(TimeSpan.FromSeconds(2))
    .Subscribe(_ => Console.WriteLine((DateTime.Now - startingTime).TotalSeconds));

// 2.0087288
// 4.005542
// 6.0163164
// ...

The related test also assumes this behavior, and is this the expected?

This behavior makes it difficult to use for purpose that, for example, prevent double UI input...

Add SkipLatestValueOnSubscribe on ReactiveProperty

Remove conflict method between R3 and UniTask in R3.Unity

similar as #26

ReactiveProperty<T>.SetValue() is missing?

I'm currently trying out R3 by replacing UniRx in my existing Unity project.

The project has a class that inherits from ReactiveProperty<T> as follows.
In R3, I cannot implement this because there is no way to set currentValue from a child class.

/// <summary>
/// ReactiveProperty whose set value is clamped in the range of [min, max].
/// </summary>
public sealed class ClampedReactiveProperty<T> : ReactiveProperty<T> where T : IComparable<T>
{
    private readonly T min, max;

    public ClampedReactiveProperty(T initialValue, T min, T max) : base(initialValue)
    {
        this.min = min;
        this.max = max;
    }

    // XXX: I cannot override this in R3!!!
    protected override void SetValue(T value)
    {
        if (Comparer.Compare(value, min) < 0) base.SetValue(min);
        else if (Comparer.Compare(value, max) > 0) base.SetValue(max);
        else base.SetValue(value);
    }

    private static IComparer<T> Comparer { get; } = Comparer<T>.Default;
}

By the way, do you plan to implment abstract ReactiveProperty<T> to substitute UniRx's IReactiveProperty<T>?
It would allow me to implement my own ClampedReactiveProperty even if ReactiveProperty<T>.SetValue() is missing.

My project contains several classes that implement IReadOnlyReactiveProperty<T> and I have successfully replaced them all with ReadOnlyReactiveProperty<T>.
Is there any reason why there is no abstract ReactiveProperty<T>?

Bug: Unity Enter Play Mode Options

When I use Enter Play Mode Options and run following code error is throwed on play mode ends.

 Observable.Interval(TimeSpan.FromSeconds(1)).Timeout(TimeSpan.FromSeconds(5))
      .Subscribe(x =>
      {
          Debug.Log("Time:" + time);
      });

ArgumentOutOfRangeException: Specified argument was out of the range of valid values.
System.Span`1[T]..ctor (R3.IFrameRunnerWorkItem[] array, System.Int32 start, System.Int32 length) (at <17d9ce77f27a4bd2afb5ba32c9bea976>:0)
System.MemoryExtensions.AsSpan[T] (T[] array, System.Int32 start, System.Int32 length) (at <17d9ce77f27a4bd2afb5ba32c9bea976>:0)
R3.Collections.FreeListCore`1[T].Clear (System.Boolean removeArray) (at <2193946fd82a4f61aea525782e26cd9d>:0)
R3.UnityFrameProvider.Clear () (at ./Library/PackageCache/com.cysharp.r3@050f4f78fe/Runtime/UnityFrameProvider.cs:73)
R3.PlayerLoopHelper+<>c__DisplayClass8_0.<InsertRunner>b__0 (UnityEditor.PlayModeStateChange state) (at ./Library/PackageCache/com.cysharp.r3@050f4f78fe/Runtime/PlayerLoopHelper.cs:121)
UnityEditor.EditorApplication.Internal_PlayModeStateChanged (UnityEditor.PlayModeStateChange state) (at <5e6ebaa7d75c4d5399dc81ecead6b43f>:0)

Then, if I restart PlayMode again, following error is throwed every frame.

ArgumentOutOfRangeException: Specified argument was out of the range of valid values.
System.Span`1[T]..ctor (R3.IFrameRunnerWorkItem[] array, System.Int32 start, System.Int32 length) (at <17d9ce77f27a4bd2afb5ba32c9bea976>:0)
System.MemoryExtensions.AsSpan[T] (T[] array, System.Int32 start, System.Int32 length) (at <17d9ce77f27a4bd2afb5ba32c9bea976>:0)
R3.Collections.FreeListCore`1[T].AsSpan () (at <2193946fd82a4f61aea525782e26cd9d>:0)
R3.UnityFrameProvider.Run () (at ./Library/PackageCache/com.cysharp.r3@050f4f78fe/Runtime/UnityFrameProvider.cs:45)

ReactiveCollection

UniRx had a class called ReactiveCollection. Does or will R3 have this as well?

SelectAwait, WhereAwait, DoAwait

integration with async/await

Missing operator: Observable<Observable<T>> Merge, Concat

[Questions] Interaction with DynamicData ?

Hi,

As always thank for sharing your incredible work.

I often use dotnet/reactive with DynamicData to build and bind collections in WPF.
As you have published ObservableCollections, I suppose you have also this kind of needs.

Do you think we can imagine building (and binding) collections with R3 ?

regards.

Delta time in R3.Godot

In Godot, there is no direct equivalent to Unity's Time.deltaTime, and we use the delta parameter of the _Process or _PhysicsProcess method.

Currently, in R3, there is no straightforward way to access delta.

GodotFrameProvider and FrameProviderDispatcher store the value as StrongBox<double>, and I would appreciate it being exposed through some means.

ReactiveProperty PropertyDrawer for Unity

needs SerializableReactiveProperty?

Feedback: ReadOnlyReactiveProperty<T> is a fairly long name

I was toying with adopting a reactive style using R3, and I found myself using reactive properties everywhere. To ensure encapsulation we need to use the read only variant. For something used so often, I found that "ReadOnlyReactiveProperty" is an uncomfortably long name.

Since R3 uses abstract base classes, it becomes tricky to wrap it in something like "ICell" or "IReactive" without losing all the extension methods.

This is just a personal observation from someone trying to adopt R3. I don't know the correct approach to this problem. Keeping the descriptive name makes sense for many reasons, but as as user I wish we could have something shorter.

[Request] Alternative implementation of AsyncReactiveCommand

This may overlap with the story #86 , but I would like to do the simultaneous control of multiple UIs in R3, which was possible with the use of AsyncReactiveCommand in UniRx.

Drop for SelectAwait/WhereAwait/SubscribeAwait is basically a recognition that controls a single Observable, which is also useful, but falls one step short of the functionality we want.

AsyncReactiveCommand which exists in UniRx was implemented in the form of returning IObservable, but I personally hope that R3, which is highly compatible with async/await, will be implemented with a structure using async/await.

Wrapping ReactiveProperty throws "InvalidOperationException: Disposable is already assigned" when subscribing

I was prototyping some ideas and wanted write a ReadOnlyReactiveProperty variant that can be overridden (a sort of public setter). Since in R3 we have no interfaces, I needed to inherit from ReadOnlyReactiveProperty while wrapping another ReactiveProperty as shown in the code beneath.

When trying to subscribe to Overridable, it throws "InvalidOperationException: Disposable is already assigned". Are we not meant to wrap the classes, or is this a bug?

public sealed class Overridable<T> : ReadOnlyReactiveProperty<T>, IOverridable<T>
    {
        public override T CurrentValue => _property.CurrentValue;

        private readonly ReactiveProperty<T> _property;
        private readonly T _baseValue;
        private readonly List<OverrideEntry> _overrides;

        private class OverrideEntry
        {
            public int Priority { get; }
            public Func<T, T> Func { get; }
            public int InsertionOrder { get; }

            public OverrideEntry(int priority, Func<T, T> func, int insertionOrder)
            {
                Priority = priority;
                Func = func;
                InsertionOrder = insertionOrder;
            }
        }

        public Overridable(T value)
        {
            _baseValue = value;
            _property = new ReactiveProperty<T>(value);
            _overrides = new List<OverrideEntry>();
        }

        protected override IDisposable SubscribeCore(Observer<T> observer)
        {
            return _property.Subscribe(observer);
        }

        public override void Dispose()
        {
            _property.Dispose();
        }

        public IDisposable Override(Func<T, T> func, int priority = 0)
        {
            var insertionOrder = _overrides.Any() ? _overrides.Max(o => o.InsertionOrder) + 1 : 0;
            var entry = new OverrideEntry(priority, func, insertionOrder);
            _overrides.Add(entry);
            _overrides.Sort((x, y) => x.Priority != y.Priority ? x.Priority.CompareTo(y.Priority) : x.InsertionOrder.CompareTo(y.InsertionOrder));
            UpdateCurrentValue();

            return Disposable.Create(() =>
            {
                _overrides.Remove(entry);
                UpdateCurrentValue();
            });
        }

        private T CalculateCurrentValue()
        {
            return _overrides.Aggregate(_baseValue, (current, entry) => entry.Func(current));
        }

        private void UpdateCurrentValue()
        {
            _property.Value = CalculateCurrentValue();
        }
    }

Error Log:

InvalidOperationException: Disposable is already assigned.

R3.SingleAssignmentDisposableCore.ThrowAlreadyAssignment () (at <9e98c8d55c734e6b90c287e63a55ae85>:0)

R3.SingleAssignmentDisposableCore.set_Disposable (System.IDisposable value) (at <9e98c8d55c734e6b90c287e63a55ae85>:0)

R3.Observable`1[T].Subscribe (R3.Observer`1[T] observer) (at <9e98c8d55c734e6b90c287e63a55ae85>:0)

R3.ObservableSubscribeExtensions.Subscribe[T] (R3.Observable`1[T] source, System.Action`1[T] onNext) (at <9e98c8d55c734e6b90c287e63a55ae85>:0)

Not working releasing event subscription of Observable.Amb()

We noticed this behavior while working with Unity, but since the problem was not specifically dependent on a package for Unity, we prepared the sample code in pure C#.

(Forgive me if I've written this in a messy way.)

using R3;

public static class Test
{
    public static event Action OnTest1ed;

    public static Observable<Unit> Test1Observable()
        => Observable.FromEvent(h => OnTest1ed += h, h => OnTest1ed -= h);

    public static event Action OnTest2ed;

    public static Observable<Unit> Test2Observable()
        => Observable.FromEvent(h => OnTest2ed += h, h => OnTest2ed -= h);

    public static void Main()
    {
        Observable.Amb(
                Test1Observable().Do(_ => Console.WriteLine("Test1ed")).Select(_ => 1),
                Test2Observable().Do(_ => Console.WriteLine("Test2ed")).Select(_ => 2))
            .Take(1)
            .Subscribe(value => Console.WriteLine($"Subscription:{value}"));

        OnTest2ed?.Invoke();
        OnTest2ed?.Invoke();
    }
}

The results of this program are as follows:

Test2ed       
Subscription:2
Test2ed

Indeed, the subscription process itself is executed only once, as specified, but the process specified in Do remains in the event.

SelectMany does not notify OnCompleted

Tested Ver.0.1.14 with .NET 8.0:

using R3;

Console.WriteLine("start");

Observable.Range(1, 2)
    .SelectMany(i => Observable.Return(i * 2))
    .Subscribe(onNext: Console.WriteLine,
               onCompleted: _ => Console.WriteLine("onCompleted"));

Console.WriteLine("end");

Output:

start
2
4
end

AggregateAPIs overload

refine Aggregate Operators

Selector APIs, etc...

does EveryValueChanged support get oldvalue ?

At present, it seems that only the value after Change can be obtained, which is the new value.

`SerializableReactiveProperty<T>` OnNext call in `OnAfterDeserialize` is problematic

Calling OnNext in OnAfterDeserialize causes issues because many Unity operations cannot be called from that execution path. For example:

UnityException: SetFloatImpl is not allowed to be called during serialization, call it from OnEnable instead.

or

UnityException: set_localRotation is not allowed to be called during serialization, call it from OnEnable instead.

Implement Operators

dotnet/reactive

UniRx

~~- [ ] AsSingleUnitObservable~~

Time/FrameProviders for platforms

Godot FrameProvider

Add EmbedingResource

Maybe GC optimization in Unity?

Happy to see R3 created.I have used UniRx years.UniRx use Gc a lot.How about reduce the memory usage or make some thing pooled?
I have a simple solution but not been test and validated,it still a toy.see BoysheO.Extensions/src/UnityReactive.Core

What's it design?
"event object whitch is oop"

The IObserver.OnError is not very useful in practice. In situations where handling errors is a common requirement, such as dealing with network-related issues, using UniTask is more beneficial and easier to debug. IObserver appears more as an alternative to C#'s events, providing abstract capabilities that events lack. Personally, my implementation leans towards removing the OnError interface. If an event source wants to push an error notification, it is advisable to consider wrapping the error within an element, for instance, returning something like OneOf{Exception, Element}.

2.The IObservable.Subscribe method currently returns an IDisposable object, which can be prone to consuming garbage collection resources. I believe designing the interface as IObservable.Subscribe + IObservable.Unsubscribe is appropriate. However, a challenge arises if Subscribe is called multiple times, as IObservable.Unsubscribe might struggle to distinguish which listener to operate on. Perhaps changing the IDisposable returned by IObservable.Subscribe to return a value object could be a suitable solution. This way, it would only require maintaining an ID table internally within the Observable object.Certainly, my approach is more aggressive. I use a Manager class to maintain all the ID tables, instead of maintaining them within the Observable.

3.An event dispatched to be published elsewhere should have tracking information. My approach is to simply add a trackingMessage to the Observable. I believe this would be immensely helpful for debugging. Otherwise, when an IObserver throws an error, we can only see the stack trace printed from the scheduler.

4.After developing my UnityReactive library to address the aforementioned issues, I've encountered a dilemma. Using Linq inevitably creates a class of objects like "middlewares," leading to increased garbage collection (GC) overhead. This poses a challenge for UnityReactive managed by UnityReactiveManager. Perhaps I should implement a parsing language similar to SQL or create a separate Manager for each operator to manage their respective states. I haven't decided which approach to take yet; it seems both have their drawbacks and may not be ideal.

Although it might sound repetitive, I want to emphasize the GC issue. In Unity, this is crucial. Striking the optimal balance between GC and usability is a must.

Package Manager support for Unity

Currently the installation mechanism requires going through NuGet and manually changing the csproj for language settings. This is is somewhat heavy compared with UniRx which had only needed a scoped registry. Do you expect simpler package manager dependency to be possible at all ? Only in future versions of Unity with the required C# features ?

R3 and MessagePipe are not compatible?

After installing MessagePipe and R3 I got the following errors:

Library/PackageCache/com.cysharp.r3@ec1e91b2c0/Editor/ObservableTrackerWindow.cs(44,29): error CS0122: 'SplitterGUILayout' is inaccessible due to its protection level

Library/PackageCache/com.cysharp.r3@ec1e91b2c0/Editor/ObservableTrackerWindow.cs(59,13): error CS0122: 'SplitterGUILayout' is inaccessible due to its protection level

Library/PackageCache/com.cysharp.r3@ec1e91b2c0/Editor/ObservableTrackerWindow.cs(67,13): error CS0122: 'SplitterGUILayout' is inaccessible due to its protection level

GUID [14cc21ce00a64784eab22336b747c4f8] for asset 'Packages/com.cysharp.r3/Editor/SplitterGUILayout.cs' conflicts with:
  'Packages/com.cysharp.messagepipe/Editor/SplitterGUILayout.cs' (current owner)
We can't assign a new GUID because the asset is in an immutable folder. The asset will be ignored.

Missing operators

There is a link to Reactive X Operator documentation in readme, but not all of them are implemented in this version (ex. Join, GroupBy). Are there plans to implement other operators?

Missing AsyncReactiveCommand<T>

Hi,

Thanks for the great library, been using it (in Unity) for a couple of weeks.

I'm really missing an equivalent for AsyncReactiveCommand which exists in UniRx, so I'm curious to hear whether there are plans to add it to R3, and if so if there's a timeline for it?

Thanks in advance.

use versionDefines to minimize package reference for Unity

https://github.com/Cysharp/UniTask/blob/master/src/UniTask/Assets/Plugins/UniTask/Runtime/UniTask.asmdef

Debounce vs Throttle

dotnet/reactive#1471

Preview1 Release Task

non generics interface for BindableReactiveProperty

Library Naming(R3 or UniRx or others)

should keep going UniRx(means Unified Rx), or use current R3, or others?
R3 means third-generation of Rx.

SerializableReactiveProperty serialized multiple times error

This started happening when I updated to the latest version today, my previous version was pulled on Jan 24, 2024, so the change that introduced the issue must have happened after that date

I have a SerializableReactiveProperty defined as

[SerializeField] private SerializableReactiveProperty<int> currentEnergy

And every time it compiles it shows the following error in the console

My class inherits from MonoBehaviour and has no child classes

SubscriptionTracker -> ObservableTracker

ReactiveProperty JsonConverter

for System.Text.Json

Avalonia FrameProvider

Throttle sampling window accepts Observable, Task

https://gist.github.com/yKimisaki/46703e756d7712800c11ea4e0b5d9ec4

AsIObservable -> AsSystemObservable

More readable.
suggested at #10 (comment) (thanks)

BindableReactiveProperty validation option

bool validateOnEnable?

`BindableReactiveProperty<T>` for XAML platforms

class BindableReactiveProperty<T> : ReactiveProperty<T>, INotifyPropertyChanged, INotifyDataErrorInfo
{
}

vs `IObservable<T>`

The current API uses a custom definition that differs from IObservable<T> and is therefore incompatible as is (conversion is possible with AsIObservable).
Whether we should go with a custom definition or conform to IObservable<T>.

Unity Trigger Source Generator

#7
It seems there is a significant demand for Triggers. In the current state of C#, it might be better to have users create only the necessary triggers on-demand, rather than preparing a large number of triggers in advance. Source Generators seem like a suitable candidate for this task.