Rx Mock Accelerometer using Observable.GenerateWithTime
When people ask about resources to learn how to use Rx on the Windows Phone, I often point them to the How to guides, including the How to: Use Reactive Extensions to Emulate and Filter Accelerometer Data for Windows Phone. Unfortunately, I have a problem with the underlying implementation of the emulation method which returns values through an iterator (yield) and IEnumerable<T> and then cast the IEnumerable to IObservable. As I pointed out in a MSDN forum post, It would be better to recommend returning just an IObservable natively and replace the yield with OnNext(T).
In reviewing this for my MIX 2011 presentation, I decided to take another look at the implementation and found a much better solution. When mocking the accelerometer, we essentially want to issue new values at a regular interval. To start, let’s take a look at the mocking method from the original How to guide:
static IEnumerable<Vector3> EmulateAccelerometerReading()
{
// Create a random number generator
Random random = new Random();
// Loop indefinitely
for (double theta = 0; ; theta+=.1 )
{
// Generate a Vector3 in which the values of each axes slowly drift between -1 and 1 and
// then normalize the vector
Vector3 reading = new Vector3((float)Math.Sin(theta), (float)Math.Cos(theta * 1.1), (float)Math.Sin(theta * .7));
reading.Normalize();
// At random intervals, generate a random spike in the data
if (random.NextDouble() > .95)
{
reading = new Vector3((float)(random.NextDouble() * 3.0 - 1.5),
(float)(random.NextDouble() * 3.0 - 1.5),
(float)(random.NextDouble() * 3.0 - 1.5));
}
// return the vector and then sleep
yield return reading;
Thread.Sleep(100);
}
}
Essentially, this implementation sets up an infinite loop and issues new values using the yield return statement and then sleeps 100 milliseconds. As I wrote last year in the MSDN forum, a better option is to return an IObservable and use OnNext rather than yield return as follows:
public static IObservable<Vector3> GetEmulator()
{
var obs = Observable.Create<Vector3>(subscriber =>
{
Random rnd = new Random();
for (double theta = 0; ; theta += .1)
{
Vector3 reading = new Vector3(
(float)Math.Sin(theta),
(float)Math.Cos(theta * 1.1),
(float)Math.Sin(theta * .7));
reading.Normalize();
if (rnd.NextDouble() > .95)
{
reading = new Vector3(
(float)(rnd.NextDouble() * 3.0 - 1.5),
(float)(rnd.NextDouble() * 3.0 - 1.5),
(float)(rnd.NextDouble() * 3.0 - 1.5));
}
// return the vector and sleep before repeating
Thread.Sleep(100);
subscriber.OnNext(reading);
}
});
return obs;
}
In preparation for Mix, I decided to revisit this and replace the looping implementation with the native Observable.GenerateWithTime method as follows:
public static IObservable<Vector3> GetAccelerometer()
{
var obs = Observable.GenerateWithTime<double, Vector3>(
initialState: 0,
condition: _ => true,
resultSelector: theta => new Vector3((float)Math.Sin(theta),
(float)Math.Cos(theta * 1.1),
(float)Math.Sin(theta * .7)),
timeSelector: _ => TimeSpan.FromMilliseconds(100),
iterate: theta => theta + .1)
.ObserveOnDispatcher();
return obs;
}Notice now we no longer have a loop in our code or a timer sleep. Those pieces are essentially handled for us inside the implementation of the GenerateWithTime. Let’s break this object instantiation down a bit to know what’s going on.
Observable.GenerateWithTime takes two generic type arguments, the double represents the type which we will increment as new values are generated. This allows us to set range values and incrementors similar to how we setup the “for” loop and potentially escape from the loop as necessary. The second generic type indicates the type of values that the observable will create. In this case, we return Vector3 types from the phone’s XNA libraries.
The first input parameter (initialState) sets the starting value of the for loop.
The condition parameter takes a lambda expression allowing us to evaluate if we should continue issuing values. In this case, we’ll keep issuing values as long as something is subscribed to us, thus we return “true”.
The resultSelector parameter takes a input value which is the current value of our iteration and returns some new value based on that input value. Here is where we generate our new vector value based on the current value.
The timeSelector parameter allows us to specify how often we want to issue new values. This would be the same as our sleep value from the original examples.
The iterate parameter allows us to increment our looping value. This is the same as the final portion of the original “for” loop declaration.
Before we return the observable, I add a method to ObserveOnDispatcher to make sure that we delegate back to the dispatcher thread. If we didn’t do this, the GenerateWithTime moves our execution context to the timer’s thread that it generates rather than leaving us on the calling thread.
Using the GenerateWithTime method allows us to abstract the looping implementation details and provides a more declarative/functional mechanism for generating the observable values. The nice thing is that we can consume this just as we would any other observable.
If you want to see this example in action, download my Rx for WP7 samples from MIX and try out the WP7 accelerometer sample. You will see that the mock value is used when you are inside the emulator and the actual accelerometer values are used when connected to a device. Regardless of how the values are generated, we process them with the same Rx subscription pipeline:
subscribed = accelReadings.Subscribe(args =>
{
Single multiplier;
if (Single.TryParse(this.MoveMultiplier.Text, out multiplier))
{
ButtonTransform.X = args.X * multiplier;
ButtonTransform.Y = args.Y * multiplier;
}
});