Create an observable wrapper for a non thread safe class

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> 开发者_如何学JAVACalc();
}

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?

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).

So from comments and re-reading your question I gather you want to repeatedly call Calc() on a constant thread and have the return results available as an IObservable<Int>()?

In this case, I'd use Observable.Create to wrap the Test class, and the EventLoopScheduler to ensure the invocations to Calc are on a single thread.

public class TestWrapper
{
  private Test _test;
  public IObservable<int> Calc()
  {
    return Observable.Create(obsvr =>
    {
        var fixedThreadsched = new EventLoopScheduler();
        var disp = new BooleanDisposable();
        while (!disp.IsDisposed)
        {
            fixedThreadsched.Schedule(() => obsvr.OnNext(_test.Calc()));
        }

        return disp;
    });
  }
}

use the ThreadLocal<T> class when creating your instance of Test:

var MyTEST = new ThreadLocal<Test>();

then you can use MyTEST.Value.Calc () for any calls...

Another option is to use put [ThreadStatic] on the Test member of the wrapper class... see http://msdn.microsoft.com/en-us/library/system.threadstaticattribute.aspx

Depending on whether you need more than one instance of Test you could make it a Singleton .