How do I turn a polling system into an Rx.Net IObservable?

Question

I have a game (based on MonoGame / XNA) with an update method like so:

public void Update(GameTime gameTime)
{
    component.Update(gameTime);
}

I would like to convert this to the Reactive pattern. My current solution is:

public void Initialize()
{
    updateSubject = new Subject<GameTime>();

    component = new Component();

    updateSubject.Subscribe((gameTime) => component.Update(gameTime));
}

public void Update(GameTime gameTime)
{
    updateSubject.OnNext(gameTime);
}

I am new to Rx so I am still learning the best way of doing things. I read that Subject should be avoided and Observable.Create should be used instead.

Is Subject appropriate here?

How could I use Observable.Create in this case?

Answer 1

The key issue you're facing here is that you need a source for your observable. In general you can create observables from a variety of sources from events, delegates, tasks, many of the observable extensions (like .Interval or .Generate), and subjects.

In your case you must have a source that you can have code, external to your observable, push values to. In this case a subject is perfectly fine, but you could just use a delegate also.

If you use a subject then your code is fine. The only downside is that you could call updateSubject.OnCompleted and finish the observable.

If you want to use a delegate then your code could look like this:

private Action<GameTime> updateGameTime = null;

public void Initialize()
{
    component = new Component();

    Observable
        .FromEvent<GameTime>(a => updateGameTime += a, a => updateGameTime -= a)
        .Subscribe((gameTime) => component.Update(gameTime));
}

public void Update(GameTime gameTime)
{
    updateGameTime(gameTime);
}

In this way the only thing you can do with updateGameTime is pass in a new GameTime - you can't "accidentally" end the sequence.

Now the whole issue with using subjects versus Observable.Create is one of state. In your code you need state, so a subject is OK. In general though, and whenever possible, it is advisable to encapsulate state - and that's what Observable.Create does for you.

Take this example:

var i = -1;
var query =
    Observable
        .Range(0, 10).Select(x =>
        {
            i = -i * 2;
            return x * i;
        });

If I subscribe to this observable twice I get these two sequences:

(1)

0 
-4 
16 
-48 
128 
-320 
768 
-1792 
4096 
-9216 

(2)

0 
-4096 
16384 
-49152 
131072 
-327680 
786432 
-1835008 
4194304 
-9437184 

The sequence changes because I used state (i.e. var i = -1;).

Had I written the code with Observable.Create I could avoid this state:

var query =
    Observable
        .Create<int>(o =>
        {
            var i = -1;
            return
                Observable
                .Range(0, 10).Select(x =>
                {
                    i = -i * 2;
                    return x * i;
                })
                .Subscribe(o);
        });

It is still the same query, but the state is encapsulated, so if I subscribe twice now I get:

(1)

0 
-4 
16 
-48 
128 
-320 
768 
-1792 
4096 
-9216 

(2)

0 
-4 
16 
-48 
128 
-320 
768 
-1792 
4096 
-9216 

There are times when writing complex queries that you might think that using a subject would make it much easier, and in general, that's where mistakes occur. You should always try to find a pure operator approach before using subjects in this case. If you can't then encapsulate the use of a subject in Observable.Create.

It's times like yours that using a subject is fine because you need that external state.

Answer 2

Just pointing out that you code unnessecarily uses Rx.

public void Initialize()
{
    //updateSubject = new Subject<GameTime>();

    component = new Component();

    //updateSubject.Subscribe((gameTime) => component.Update(gameTime));
}

public void Update(GameTime gameTime)
{
    //updateSubject.OnNext(gameTime);
    component.Update(gameTime)
}

Here I have removed the Subject and just call directly through to the components Update method, to illustrate the point.

Perhaps you are looking for a private method of polling? In which case Observable.Interval could be a good place to start.