Subject.AsObservable with ObserveOnMainThread: OnCompleted Not Triggered

[yahagi-day]

I m using Unity 2022.3.31f1 and R3 1.3.0., subscribing to an Observable created with Subject.AsObservable using ObserveOnMainThread doesn't trigger OnCompleted. Below is the sample code.

using R3;

public class Test
{
    public Observable<Unit> TestObservable;
    private Subject<Unit> _testSubject = new Subject<Unit>();

    public Test()
    {
        TestObservable = _testSubject.AsObservable();
    }
    public void OnNext()
    {
        _testSubject.OnNext(Unit.Default);
    }
    
    public void OnCompleted()
    {
        _testSubject.OnCompleted();
    }
}

using System;
using System.Threading.Tasks;
using UnityEngine;
using R3;

public class NewBehaviourScript : MonoBehaviour
{
    private Test test = new Test();
    
    async void Start()
    {
        test.TestObservable.ObserveOnMainThread().Subscribe(
            onNext: _ => { Debug.Log("Next"); },
            onCompleted: _ => Debug.Log("Completed"));

        await Task.Delay(TimeSpan.FromSeconds(10));
        test.OnNext();
        await Task.Delay(TimeSpan.FromSeconds(10));
        test.OnCompleted();
        Debug.Log("Done");
    }
}

When executing this, OnNext is called, but OnCompleted is skipped and "Done" is output.

[TORISOUP]

I looked into it and found that AsObservable() uses WrappedObserver, which seems to be the cause of the issue. Since WrappedObserver has AutoDisposeOnCompleted = true, it immediately calls observer.Dispose() after Subject.OnCompleted() is invoked. As a result, subsequent operations—including ObserveOn—are stopped.

[TORISOUP]

Just as my personal opinion, one possible approach might be to make AutoDisposeOnCompleted configurable via the WrappedObserver constructor, and allow it to be specified when calling Wrap(). If it were set to false only in the case of AsObservable(), the impact might be minimized.

Alternatively, defining a dedicated WrappedObserver specifically for AsObservable() might be a simpler solution.

[TORISOUP]

Alternatively, defining a dedicated WrappedObserver specifically for AsObservable() might be a simpler solution.

I've submitted a pull request with a fix based on this approach.

[github-actions[bot]]

This issue is stale because it has been open 180 days with no activity. Remove stale label or comment or this will be closed in 30 days.