Take the last item pushed to an Observable (Sequence)

Question

I have an IObservable<Item> inside a class and I want to expose a read-only property that provides the last item pushed to the observable at a given time. So it will provide a single value of Item.

If no value has been pushed, then it will have to return a default value.

How can I do this without having to subscribe to the observable and having a "backing field"?

Answer 1

Just to supplement @Asti's answer a bit here, and perhaps help you with your frustration:

An observable isn't a physical 'thing', it's more a logical concept. Rx is often compared to LINQ, and it's a fair comparison much of the time. It breaks down though when you start talking data structures: LINQ's enumerables are similar enough to Lists for learning purposes.

However, on the Rx side, there's simply no good equivalent to List. An observable is a transient data structure, all operators deal with this transient state. If you're looking for a permanent state, you're leaving Rx.

Having said that, converting an observable to some sort of state is a common problem, and there are some packages that may help you: ReactiveUI is perhaps the most known. ReactiveProperty is another. Both of these packages are flawed, but may help you.

If you're simply looking for an easier way to get a backing field, without boiler-plating out a backing field, this will work:

public static class ReactivePropertyExtensions
{
    public static ReactiveProperty<T> ToReactiveProperty<T>(this IObservable<T> source)
    {
        return new ReactiveProperty<T>(source);
    }

    public static ReactiveProperty<T> ToReactiveProperty<T>(this IObservable<T> source, T defaultValue)
    {
        return new ReactiveProperty<T>(source, defaultValue);
    }
}

public class ReactiveProperty<T> : IDisposable
{
    private IObservable<T> Source { get; }
    private IDisposable Subscription { get; }
    public T Value { get; private set; }

    public ReactiveProperty(IObservable<T> source)
        : this(source, default(T)) { }

    public ReactiveProperty(IObservable<T> source, T defaultValue)
    {
        Value = defaultValue;
        Source = source;
        Subscription = source.Subscribe(t => Value = t);
    }

    public void Dispose()
    {
        Subscription.Dispose();
    }
}

Example use:

var ticker = Observable.Interval(TimeSpan.FromSeconds(1))
    .Publish().RefCount();

var latestTickerValue = ticker.ToReactiveProperty();
Console.WriteLine(latestTickerValue.Value);
await Task.Delay(TimeSpan.FromSeconds(1));
Console.WriteLine(latestTickerValue.Value);
await Task.Delay(TimeSpan.FromSeconds(3));
Console.WriteLine(latestTickerValue.Value);

Answer 2

Assuming a hot observable.

For observable = source.Replay(1); observable.Connect();

Provide the value with:

public int Value => observable.Take(1).Amb(Observable.Return(defaultValue)).Wait();

This will return a default value in case no values have been pushed.

You want a transition from Reactive to state, so a backing field isn't a terrible option. You mentioned that you don't want to subscribe, but to observe anything: something, somewhere has to subscribe.