Workarounds for RX glitches?

Question

I'm experimenting with Reactive Extensions on various platforms, and one thing that annoys me a bit are the glitches.

Even though for UI code these glitches might not be that problematic, and usually one can find an operator that works around them, I still find debugging code more difficult in the presence of glitches: the intermediate results are not important to debug, but my mind does not know when a result is intermediate or "final".

Having worked a bit with pure functional FRP in Haskell and synchronous data-flow systems, it also 'feels' wrong, but that is of course subjective.

But when hooking RX to non-UI actuators (like motors or switches), I think glitches are more problematic. How would one make sure that only the correct value is send to the external actuators?

Maybe this can be solved by some 'dispatcher' that knows when some 'external sensor' fired the initiating event, so that all internal events are handled before forwarding the final result(s) to the actuators. Something like described in the flapjax paper.

The question(s) I hope to get answers for are:

1. Is there something in RX that makes fixing glitches for synchronous notifications impossible?
2. If not, does a (preferably production quality) library or approach exists for RX that fixes synchronous glitches? Especially for the single-threaded Javascript this might make sense?
3. If no general solution exists, how would RX be used to control external sensors/actuators without glitches at the actuators?

Let me give an example

Suppose I want to print a sequence of tuples (a,b) where the contract is

a=n    b=10 * floor(n/10)

n is a natural number stream = 0,1,2....

So I expect the following sequence

(a=0, b=0)
(a=1, b=0)
(a=2, b=0)
...
(a=9, b=0)
(a=10, b=10)
(a=11, b=10)
...

In RX, to make things more interesting, I will use filter for computing the b stream

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

var a = n.Select(t => "a=" + t);

var b = n.Where(t => t % 10 == 0)
        .Select(t => "b=" + t);

var ab = a.CombineLatest(b, Tuple.Create);

ab.Subscribe(Console.WriteLine);

This gives what I believed to be a glitch (temporary violation of the invariant/contract):

(a=0, b=0)
(a=1, b=0)
(a=2, b=0)
...
(a=10, b=0) <-- glitch?
(a=10, b=10)
(a=11, b=10)

I realize that this is the correct behavior of CombineLatest, but I also thought this was called a glitch because in a real pure FRP system, you do not get these intermediate-invariant-violating results.

Note that in this example, I would not be able to use Zip, and also WithLatestFrom would give an incorrect result.

Of course I could just simplify this example into one monadic computation, never multi-casting the n stream occurrences (this would mean not being able to filter but just map), but that's not the point: IMO in RX you always get a 'glitch' whenever you split and rejoin an observable stream:

    s
   / \
  a   b
   \ /
    t

For example, in FlapJAX you don't get these problems.

Does any of this make sense?

Thanks a lot, Peter

Answer 1

Update: Let me try to answer my own question in an RX context.

First of all, it seems my understanding of what a "glitch" is, was wrong. From a pure FRP standpoint, what looked like glitches in RX to me, seems actually correct behavior in RX.

So I guess that in RX we need to be explicit about the "time" at which we expect to actuate values combined from sensors.

In my own example, the actuator is the console, and the sensor the interval n.

So if I change my code

ab.Subscribe(Console.WriteLine);

into

ab.Sample(n).Subscribe(Console.WriteLine);

then only the "correct" values are printed.

This does mean that when we get an observable sequence that combines values from sensors, that we must know all the original sensors, merge them all, and sample the values with that merged signal before sending any values to actuators...

So an alternative approach would be to "lift" IObservable into a "Sensed" structure that remembers and merges the originating sensors, for example like this:

public struct Sensed<T>
{
    public IObservable<T> Values;
    public IObservable<Unit> Sensors;

    public Sensed(IObservable<T> values, IObservable<Unit> sensors)
    {
        Values = values;
        Sensors = sensors;
    }

    public IObservable<Unit> MergeSensors(IObservable<Unit> sensors)
    {
        return sensors == Sensors ? Sensors : Sensors.Merge(sensors);
    }

    public IObservable<T> MergeValues(IObservable<T> values)
    {
        return values == Values ? Values : Values.Merge(values);
    }
}

And then we must transfer all RX method to this "Sensed" structure:

public static class Sensed
{
    public static Sensed<T> Sensor<T>(this IObservable<T> source)
    {
        var hotSource = source.Publish().RefCount();
        return new Sensed<T>(hotSource, hotSource.Select(_ => Unit.Default));
    }

    public static Sensed<long> Interval(TimeSpan period)
    {
        return Observable.Interval(period).Sensor();
    }

    public static Sensed<TOut> Lift<TIn, TOut>(this Sensed<TIn> source, Func<IObservable<TIn>, IObservable<TOut>> lifter)
    {
        return new Sensed<TOut>(lifter(source.Values), source.Sensors);
    }

    public static Sensed<TOut> Select<TIn, TOut>(this Sensed<TIn> source, Func<TIn, TOut> func)
    {
        return source.Lift(values => values.Select(func));
    }

    public static Sensed<T> Where<T>(this Sensed<T> source, Func<T, bool> func)
    {
        return source.Lift(values => values.Where(func));
    }

    public static Sensed<T> Merge<T>(this Sensed<T> source1, Sensed<T> source2)
    {
        return new Sensed<T>(source1.MergeValues(source2.Values), source1.MergeSensors(source2.Sensors));
    }

    public static Sensed<TOut> CombineLatest<TIn1, TIn2, TOut>(this Sensed<TIn1> source1, Sensed<TIn2> source2, Func<TIn1, TIn2, TOut> func)
    {
        return new Sensed<TOut>(source1.Values.CombineLatest(source2.Values, func), source1.MergeSensors(source2.Sensors));
    }

    public static IDisposable Actuate<T>(this Sensed<T> source, Action<T> next) 
    {
        return source.Values.Sample(source.Sensors).Subscribe(next);
    }
}

My example then becomes:

var n = Sensed.Interval(TimeSpan.FromMilliseconds(100));
var a = n.Select(t => "a=" + t);
var b = n.Where(t => t % 10 == 0).Select(t => "b=" + t);
var ab = a.CombineLatest(b, Tuple.Create);
ab.Actuate(Console.WriteLine);

And again only the "desired" values are passed to the actuator, but with this design, the originating sensors are remember in the Sensed structure.

I'm not sure if any of this makes "sense" (pun intended), maybe I should just let go of my desire for pure FRP, and live with it. After all, time is relative ;-)

Peter