Create an observable wrapper for a non thread safe class

Question

I have a class,

public class Test
{
  public int Calc();
}

which requires all calls to Calc to be performed on the same thread as the one on which Test was created. I need to create Test once (expensive operation) and call Calc multiple times.

I'd like to have a wrapper that will let me call Calc asynchronousely:

public class TestWrapper
{
  private Test _test;
  public IObservable<int> Calc();
}

One way to do it would be to create a BackgroundWorker or a Thread and use it as a guarantee that all operations on Test are on the same thread. For simplicity, we can assume that all calls to Calc() will be executed sequentially, so no need to worry about queueing.

Is there a more elegant RX way to do it?

Answer 1

If it's possible for Test to be created when TestWrapper is created then this class seems to meet your requirements:

public class TestWrapper
{
    public TestWrapper(Func<Test> factory)
    {
        _scheduler = new EventLoopScheduler();
        _test = Observable.Start(factory, _scheduler).First();
    }

    private readonly EventLoopScheduler _scheduler;
    private readonly Test _test;

    public IObservable<int> Calc()
    {
        return Observable.Start(() => _test.Calc(), _scheduler);
    }
}

It is used like so:

var testWrapper = new TestWrapper(() => new Test());
testWrapper.Calc().Subscribe(x => { });

I've tested it and it creates Test on the same thread that Calc is executed on. The subscription, on the other hand, is handled on the same thread as the testWrapper itself was created on (i.e. the calling thread).