ReadOnlyReactivePropertySlim を CombineLatest したときに初期値を反映させる方法について (Question about Reflecting Initial Values when Combining ReadOnlyReactivePropertySlim with CombineLatest)

[kbfriston3]

ReadOnlyReactivePropertySlim を CombineLatest で結合するときの動作について質問です。

以下のように、単なる Observable を CombineLatest した場合、Value に aaa が設定されますが、 ToReadOnlyReactivePropertySlim をした後に CombineLatest をすると Value が null になってしまいます。

var ob_a = Observable.Return("aaa");
var ob_b = Observable.Return("bbb");
var ob_a_b = ob_a.CombineLatest(ob_b, (a, b) => a != null ? a : b).ToReadOnlyReactivePropertySlim();

var rp_a = Observable.Return("aaa").ToReadOnlyReactivePropertySlim();
var rp_b = Observable.Return("bbb").ToReadOnlyReactivePropertySlim();
var rp_a_b = rp_a.CombineLatest(rp_b, (a, b) => a != null ? a : b).ToReadOnlyReactivePropertySlim();

System.Diagnostics.Debug.WriteLine($"ob_a_b={ob_a_b.Value}"); // ob_a_b=aaa
System.Diagnostics.Debug.WriteLine($"rp_a_b={rp_a_b.Value}"); // rp_a_b= 

このように ReadOnlyReactivePropertySlim を CombineLatest した場合にも、 初期値（もともとの ReadOnlyReactivePropertySlim の Value ）を Value に反映する方法はないでしょうか？

ReadOnlyReactivePropertySlim にしてから CombineLatest したい理由としては Hot にして他にも使いまわしたかったからです。 （この例でいうと rp_a などを他にも使いまわしたかったから、といった感じです）

よろしくおねがいします。

English

When combining ReadOnlyReactivePropertySlim using CombineLatest, the initial value is not reflected. For example, when combining simple Observables, the value is set to “aaa”. However, after converting to ReadOnlyReactivePropertySlim and then combining, the value becomes null.

var ob_a = Observable.Return("aaa");
var ob_b = Observable.Return("bbb");
var ob_a_b = ob_a.CombineLatest(ob_b, (a, b) => a != null ? a : b).ToReadOnlyReactivePropertySlim();

var rp_a = Observable.Return("aaa").ToReadOnlyReactivePropertySlim();
var rp_b = Observable.Return("bbb").ToReadOnlyReactivePropertySlim();
var rp_a_b = rp_a.CombineLatest(rp_b, (a, b) => a != null ? a : b).ToReadOnlyReactivePropertySlim();

System.Diagnostics.Debug.WriteLine($"ob_a_b={ob_a_b.Value}"); // ob_a_b=aaa
System.Diagnostics.Debug.WriteLine($"rp_a_b={rp_a_b.Value}"); // rp_a_b= 

Is there a way to reflect the initial value (the original ReadOnlyReactivePropertySlim value) in the combined value when using CombineLatest?

The reason for wanting to use CombineLatest after converting to ReadOnlyReactivePropertySlim is to make it hot and reusable elsewhere (e.g., reusing rp_a).

Thank you.

[runceel]

質問ありがとうございます。 Observable.Return("aaa"); を Subscrive すると OnNext と OnCompleted が実行されます。

using System.Reactive.Linq;

Observable.Return("aaa").Subscribe(
    x => Console.WriteLine($"OnNext: {x}"),
    ex => Console.WriteLine($"OnError: {ex.Message}"),
    () => Console.WriteLine("OnCompleted"));

実行結果

OnNext: aaa
OnCompleted

ReadOnlyReactivePropertySlim<T> などのクラスはソースとなる IObservable<T> が完了すると ReadOnlyReactivePropertySlim<T> 自体も OnCompleted を発行して終了状態になります。 その状態になると Subscribe をしても OnNext などは呼び出されません。

今回の例ではすでに完了している状態になっているため CombineLatest をしても何も起きないという状態になっています。提供していただいたサンプルプログラムを以下のように書き換えると CombineLatest の結果が完了している状態になっていることが確認できると思います。

using Reactive.Bindings;
using System.Reactive.Linq;

var ob_a = Observable.Return("aaa");
var ob_b = Observable.Return("bbb");

var rp_a = Observable.Return("aaa").ToReadOnlyReactivePropertySlim();
var rp_b = Observable.Return("bbb").ToReadOnlyReactivePropertySlim();
rp_a.CombineLatest(rp_b, (a, b) => a != null ? a : b)
    .Subscribe(
        x => Console.WriteLine(x),
        ex => Console.WriteLine(ex.Message),
        () => Console.WriteLine("Completed") // この行が実行される
    );

そのためこの動作は仕様になります。目的の動作をさせたい場合は Observable.Return("aaa") などではなく OnCompleted が呼ばれない BehaviorSubject<T> などをソースとすることで実現できます。

using Reactive.Bindings;
using System.Reactive.Linq;
using System.Reactive.Subjects;

var ob_a = new BehaviorSubject<string>("aaa");
var ob_b = new BehaviorSubject<string>("bbb");
var ob_a_b = ob_a.CombineLatest(ob_b, (a, b) => a != null ? a : b).ToReadOnlyReactivePropertySlim();

var rp_a = new BehaviorSubject<string>("aaa").ToReadOnlyReactivePropertySlim();
var rp_b = new BehaviorSubject<string>("bbb").ToReadOnlyReactivePropertySlim();
var rp_a_b = rp_a.CombineLatest(rp_b, (a, b) => a != null ? a : b).ToReadOnlyReactivePropertySlim();

System.Diagnostics.Debug.WriteLine($"ob_a_b={ob_a_b.Value}"); // ob_a_b=aaa
System.Diagnostics.Debug.WriteLine($"rp_a_b={rp_a_b.Value}"); // rp_a_b=aaa

English

Thank you for your question.

When you Subscribe to Observable.Return("aaa");, both OnNext and OnCompleted are executed.

using System.Reactive.Linq;

Observable.Return("aaa").Subscribe(
    x => Console.WriteLine($"OnNext: {x}"),
    ex => Console.WriteLine($"OnError: {ex.Message}"),
    () => Console.WriteLine("OnCompleted"));

Execution result:

OnNext: aaa
OnCompleted

Classes like ReadOnlyReactivePropertySlim will also issue OnCompleted and enter a completed state when the source IObservable completes. In this state, even if you Subscribe, OnNext and other events will not be called.

In your example, since it is already in a completed state, nothing happens when you use CombineLatest. If you rewrite the provided sample program as follows, you can confirm that the result of CombineLatest is in a completed state.

using Reactive.Bindings;
using System.Reactive.Linq;

var ob_a = Observable.Return("aaa");
var ob_b = Observable.Return("bbb");

var rp_a = Observable.Return("aaa").ToReadOnlyReactivePropertySlim();
var rp_b = Observable.Return("bbb").ToReadOnlyReactivePropertySlim();
rp_a.CombineLatest(rp_b, (a, b) => a != null ? a : b)
    .Subscribe(
        x => Console.WriteLine(x),
        ex => Console.WriteLine(ex.Message),
        () => Console.WriteLine("Completed") // This line will be executed
    );

Therefore, this behavior is by design. If you want the desired behavior, you can achieve it by using a source that does not call OnCompleted, such as BehaviorSubject, instead of Observable.Return("aaa").

using Reactive.Bindings;
using System.Reactive.Linq;
using System.Reactive.Subjects;

var ob_a = new BehaviorSubject<string>("aaa");
var ob_b = new BehaviorSubject<string>("bbb");
var ob_a_b = ob_a.CombineLatest(ob_b, (a, b) => a != null ? a : b).ToReadOnlyReactivePropertySlim();

var rp_a = new BehaviorSubject<string>("aaa").ToReadOnlyReactivePropertySlim();
var rp_b = new BehaviorSubject<string>("bbb").ToReadOnlyReactivePropertySlim();
var rp_a_b = rp_a.CombineLatest(rp_b, (a, b) => a != null ? a : b).ToReadOnlyReactivePropertySlim();

System.Diagnostics.Debug.WriteLine($"ob_a_b={ob_a_b.Value}"); // ob_a_b=aaa
System.Diagnostics.Debug.WriteLine($"rp_a_b={rp_a_b.Value}"); // rp_a_b=aaa

[kbfriston3]

ご回答ありがとうございます。

完了している状態になっているため CombineLatest をしても何も起きないということなのですね。 承知いたしました。

BehaviorSubjectを使った修正案もありがとうございます。 ここについて追加で質問させてください。

今回の例では、サンプルを簡単にするために Observable.Return を使っていたのですが、 実際のIObservableの発生源は ObserveProptery メソッドであったり、 CollectionChangedAsObservable メソッドであったり、様々です。

こういった場合、単純に new BehaviorSubject で置き換えるといったことがかなわないのですが、このような場合はどうすればいいでしょうか？

試してみたところ、以下のようにReadOnlyReactivePropertySlimからBehaviorSubjectに変換できることはわかりました。そして、これらをCombineLatestすることで当初の目的は達成できたのですが、どうにも冗長な気もしたので質問させていただきました。

var bs_a = new BehaviorSubject<string>(rp_a.Value);
rp_a.Subscribe(x => bs_a.OnNext(x), ex => bs_a.OnError(ex), () => { });

よろしくお願いします。

English

Thank you for your response.

Thank you also for the suggested fix using BehaviorSubject. I have an additional question about this.

In this example, I was using Observable.Return to simplify the sample. But the actual source of IObservable can be the ObserveProptery method, the CollectionChangedAsObservable method, and so on. In these cases, I can't simply replace it with a new BehaviorSubject.

What should I do in these cases?

I have tried and found that I can convert from a RedOnlyReactivePropertySlim to a BehaviorSubject as follows. And by CombineLatest, I was able to achieve my initial goal, but I felt it was redundant, so I asked the question.

var bs_a = new BehaviorSubject<string>(rp_a.Value);
rp_a.Subscribe(x => bs_a.OnNext(x), ex => bs_a.OnError(ex), () => { });

Thanks in advance.

[runceel]

@kbfriston3 今回の例が簡単にするために Observable.Return("aaa") のようにしているので OnCompleted が発行されて CombineLatest が期待通りに動いていないだけのような気がします。

実際に多くのケースで使われる ObserveProperty や CollectionChangedAsObservable などは Dispose を明示的に呼び出すまでは。OnCompleted は発行されません。そのため以下のように ObserveProperty を使ったコードでは CombineLatest がきちんと動作すると思います。

using Reactive.Bindings;
using Reactive.Bindings.Extensions;
using System.ComponentModel;
using System.Reactive.Linq;

var person = new Person { FirstName = "Jiro", LastName = "Inoue" };
var firstName = person.ObserveProperty(x => x.FirstName).ToReadOnlyReactivePropertySlim("");
var lastName = person.ObserveProperty(x => x.LastName).ToReadOnlyReactivePropertySlim("");

var fullName = firstName.CombineLatest(lastName, (f, l) => $"{f} {l}")
    .ToReadOnlyReactivePropertySlim("");

fullName.Subscribe(Console.WriteLine);

person.FirstName = "Taro";
person.LastName = "Yamada";
person.FirstName = "Hanako";
person.LastName = "Suzuki";

Console.WriteLine("## Unsbscribe");
firstName.Dispose();
lastName.Dispose();

// No output because of source properties are disposed.
person.FirstName = "Ichiro";
person.LastName = "Sato";

class Person : INotifyPropertyChanged
{
    public event PropertyChangedEventHandler? PropertyChanged;

    // FirstName and LastName properties
    private string _firstName = "";
    public string FirstName
    {
        get { return _firstName; }
        set
        {
            _firstName = value;
            PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(FirstName)));
        }
    }
    private string _lastName = "";
    public string LastName
    {
        get { return _lastName; }
        set
        {
            _lastName = value;
            PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(LastName)));
        }
    }
}

出力結果は以下のようになります。

Jiro Inoue
Taro Inoue
Taro Yamada
Hanako Yamada
Hanako Suzuki
## Unsbscribe

もし、うまく動かない実際の例を出してもらえれば動かない原因を確認することは出来ると思いますが現在のサンプルが動かない理由は、既に完了状態になっている IObservable に CombineLatest をしても動かないという回答になってしまいます。

English

In this example, Observable.Return("aaa") is used to simplify the case, so OnCompleted is issued, and CombineLatest does not work as expected.

In many cases, ObserveProperty and CollectionChangedAsObservable are used, and OnCompleted is not issued until Dispose is explicitly called. Therefore, in the code using ObserveProperty as shown below, CombineLatest should work properly.

using Reactive.Bindings;
using Reactive.Bindings.Extensions;
using System.ComponentModel;
using System.Reactive.Linq;

var person = new Person { FirstName = "Jiro", LastName = "Inoue" };
var firstName = person.ObserveProperty(x => x.FirstName).ToReadOnlyReactivePropertySlim("");
var lastName = person.ObserveProperty(x => x.LastName).ToReadOnlyReactivePropertySlim("");

var fullName = firstName.CombineLatest(lastName, (f, l) => $"{f} {l}")
    .ToReadOnlyReactivePropertySlim("");

fullName.Subscribe(Console.WriteLine);

person.FirstName = "Taro";
person.LastName = "Yamada";
person.FirstName = "Hanako";
person.LastName = "Suzuki";

Console.WriteLine("## Unsbscribe");
firstName.Dispose();
lastName.Dispose();

// No output because of source properties are disposed.
person.FirstName = "Ichiro";
person.LastName = "Sato";

class Person : INotifyPropertyChanged
{
    public event PropertyChangedEventHandler? PropertyChanged;

    // FirstName and LastName properties
    private string _firstName = "";
    public string FirstName
    {
        get { return _firstName; }
        set
        {
            _firstName = value;
            PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(FirstName)));
        }
    }
    private string _lastName = "";
    public LastName
    {
        get { return _lastName; }
        set
        {
            _lastName = value;
            PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(nameof(LastName)));
        }
    }
}

The output will be as follows:

Jiro Inoue
Taro Inoue
Taro Yamada
Hanako Yamada
Hanako Suzuki
## Unsbscribe

If you can provide an example that doesn’t work well, I can check the cause of the issue. However, the reason the current sample doesn’t work is that CombineLatest doesn’t work with an IObservable that is already in a completed state.

[kbfriston3]

@runceel ご回答ありがとうございます。

おっしゃる通りでした。 ObserveProperty だけを使って CombineLatest すると意図した通りに動作しました。

もともとのコードでは、 Observable.Return と ObserveProperty を混ぜて CombineLatest していたのが問題だったようです。

Text_A = "aaa";

var rp_a = this.ObserveProperty(a => Text_A).ToReadOnlyReactivePropertySlim();
var rp_b = Observable.Return("bbb").ToReadOnlyReactivePropertySlim();
var rp_a_b = rp_a.CombineLatest(rp_b, (a, b) => a != null ? a : b).ToReadOnlyReactivePropertySlim();
Console.WriteLine($"rp_a_b={rp_a_b.Value}"); // rp_a_b=

丁寧に質問にお答えいただきありがとうございました。

English

Thank you for your response.

You were correct. CombineLatest using ObserveProperty worked as intended.

In my code, I was doing CombineLatest with a mixture of Observable.Return and ObserveProperty. This seems to have been the problem.

Thank you for taking the time to answer my questions.