ReactiveX video game render loop timing

Question

I'm currently investigating reactivex in .net for use in a simulation heavy video game. My first goal is to create the basic event / rendering loop. I have the following code that does nearly exactly what I want.

var frames = Observable
    .Interval(TimeSpan.Zero, new EventLoopScheduler())
    .Timestamp()
    .Select(timestamp => new Frame
    {
        Count = timestamp.Value,
        Timestamp = timestamp.Timestamp.ToUnixTimeMilliseconds()
    });

frames.Subscribe(frame =>
{
    Console.WriteLine($"GameObject 1 - frame.Count: {frame.Count} frame.Timestamp: {frame.Timestamp}");
});

frames.Subscribe(frame =>
{
    Console.WriteLine($"GameObject 2 - frame.Count: {frame.Count} frame.Timestamp: {frame.Timestamp}");
});

frames.ToTask().Wait();

This emits frame events as fast as possible on the main thread. Perfect! However, I noticed the timestamps between two subscribers might be slightly off for the same frame. For example,

GameObject 1 - frame.Count: 772 frame.Timestamp: 1519215592309
GameObject 2 - frame.Count: 772 frame.Timestamp: 1519215592310

The timestamps are one millisecond off for the same frame. For this to work both of these gameobjects/subscribers need to get the exact same timestamp each frame otherwise the simulation may not run properly.

I'm guessing the Timestamp operator is being evaluated once per subscriber for each event emitted in the stream. Is there a way to get it so its evaluated once for all subscribers per event emitted? Thanks!

Answer 1

Whelp, I should of given it a few more minutes of documentation reading.

The solution was to make my frame event stream connectable. All I had to do was Publish the frame event stream before subscribers attached to it. Then call Connect right before the event loop started. Perfect! Damn Rx is great.

var frames = Observable
    .Interval(TimeSpan.Zero, new EventLoopScheduler())
    .Timestamp()
    .Select(timestamp => new Frame
    {
        Count = timestamp.Value,
        Timestamp = timestamp.Timestamp.ToUnixTimeMilliseconds()
    })
    .Publish();

frames.Subscribe(frame =>
{
    Console.WriteLine($"GameObject 1 - frame.Count: {frame.Count} frame.Timestamp: {frame.Timestamp}");
});

frames.Subscribe(frame =>
{
    Console.WriteLine($"GameObject 2 - frame.Count: {frame.Count} frame.Timestamp: {frame.Timestamp}");
});

frames.Connect();

frames.ToTask().Wait();