Combining observables that depend upon other observables

Question

I'm simulating the flight of a drone using observables for the altitude. The altitude should vary according to this scheme:

1. Altitude increases from 0 to BaseAltitude, that is a fixed altitude.
2. After the BaseAltitude is reached, the drone starts cruising, describing a sine wave, starting at BaseAltitude
3. Upon a signal, the drone should start landing. This is, starting from the current altitude, the drone should go down linearly until it reaches 0.

As you might notice, when the landing starts, the altitude is unknown at design time. The takeoff sequence should take the last altitude as the start. So, one sequence depends on the last value produced by another sequence. My brain aches!

Well, I'm completely stuck with this.

The only code I have currently is below. I put it to illustrate the problem. You will get it quickly...

public class Drone
{
    public Drone()
    {
        var interval = TimeSpan.FromMilliseconds(200);

        var takeOff = Observable.Interval(interval).TakeWhile(h => h < BaseAltitude).Select(t => (double)t);

        var cruise = Observable
            .Interval(interval).Select(t => 100 * Math.Sin(t * 2 * Math.PI / 180) + BaseAltitude)
            .TakeUntil(_ => IsLanding);

        var landing = Observable
            .Interval(interval).Select(t => ??? );

        Altitude = takeOff.Concat(cruise).Concat(landing);
    }

    public bool IsLanding { get; set; }
    public double BaseAltitude { get; set; } = 100;
    public IObservable<double> Altitude { get; }
}

Answer 1

You use LastAsync to get the last value of cruise, then SelectMany into the observable you want.

You'll need to change cruise slightly to handle multiple subscriptions.

    var cruise = Observable.Interval(interval)
        .Select(t => 100 * Math.Sin(t * 2 * Math.PI / 180) + BaseAltitude)
        .TakeUntil(_ => IsLanding)
        .Replay(1)
        .RefCount();

    var landing = cruise
        .LastAsync()
        .SelectMany(maxAlt => Observable.Interval(interval).Select(i => maxAlt - i))
        .TakeWhile(alt => alt >= 0);

    Altitude = takeOff.Concat(cruise).Concat(landing);

---

Why do I need .Replay(1).Refcount()?

Everything here is a cold observable, and none of them will run concurrently. Concat actually makes sure that they are not concurrent. So the marble diagram you want will look something like this:

t        : 1-2-3-4-5-6-7-8-9-0-1-2-3-4-5-6-7-8-...
takeOff  : 1-2-3-4-5-|
cruise   :           6-7-8-7-6-|
isLanding: T-------------------F----------------
landing  :                     5-4-3-2-1-0-|

If you define landing = cruise.LastAsync()... then it will try to subscribe to cruise at time 11 and get the last value.

• If you left cruise defined as you had it, it would try to resubscribe to a new cold observable, which would result in 0 elements, because isLanding is now false.
• If you add .Publish().RefCount() to cruise definition, it would try to subscribe to the previous observable which is completed, and that also would result in 0 elements.
• .Replay(1).Refcount() caches the last value, so any subscribers that subscribe after the observable has completed will still get the last value (which is what you want).

Answer 2

You really should try to craft the observable so that it models selecting takeoff or landing at any time - just like a user of a drone might do.

That becomes quite simple if you craft your code like this:

public class Drone
{
    public Drone()
    {
        this.Altitude = ...
    }

    private bool _isLanding = true;
    private Subject<bool> _isLandings = new Subject<bool>();

    public bool IsLanding
    {
        get => _isLanding;
        set
        {
            _isLanding = value;
            _isLandings.OnNext(value);
        }
    }

    public double BaseAltitude { get; set; } = 100.0;
    public IObservable<double> Altitude { get; }
}

Each time IsLanding is changed the private _isLandings fires a value that can be used to change the mode of the drone.

Now, the definition of Altitude starts with this basic pattern:

    this.Altitude =
        _isLandings
            .Select(x => x ? landing : takeOff.Concat(cruise))
            .Switch()
            .StartWith(altitude);

The use of .Switch() here is key. Whenever _isLandings produces a value the switch chooses between landing or taking off. It becomes a single observable that responds to going up or going down.

The full code looks like this:

public class Drone
{
    public Drone()
    {
        var altitude = 0.0;
        var interval = TimeSpan.FromMilliseconds(200);

        IObservable<double> landing =
            Observable
                .Interval(interval)
                .TakeWhile(h => altitude > 0.0)
                .Select(t =>
                {
                    altitude -= 10.0;
                    altitude = altitude > 0.0 ? altitude : 0.0;
                    return altitude;
                });

        IObservable<double> takeOff =
            Observable
                .Interval(interval)
                .TakeWhile(h => altitude < BaseAltitude)
                .Select(t =>
                {
                    altitude += 10.0;
                    altitude = altitude < BaseAltitude ? altitude : BaseAltitude;
                    return altitude;
                });

        IObservable<double> cruise =
            Observable
                .Interval(interval)
                .Select(t =>
                {
                    altitude = 10.0 * Math.Sin(t * 2.0 * Math.PI / 180.0) + BaseAltitude;
                    return altitude;
                });

        this.Altitude =
            _isLandings
                .Select(x => x ? landing : takeOff.Concat(cruise))
                .Switch()
                .StartWith(altitude);
    }

    private bool _isLanding = true;
    private Subject<bool> _isLandings = new Subject<bool>();

    public bool IsLanding
    {
        get => _isLanding;
        set
        {
            _isLanding = value;
            _isLandings.OnNext(value);
        }
    }

    public double BaseAltitude { get; set; } = 100.0;
    public IObservable<double> Altitude { get; }
}

You can test it with this:

var drone = new Drone();

drone.Altitude.Subscribe(x => Console.WriteLine(x));

Thread.Sleep(2000);

drone.IsLanding = false;

Thread.Sleep(4000);

drone.IsLanding = true;