我正在尝试使用Reactive Extensions for .NET重写一些代码，但是我需要一些关于如何实现我的目标的指导。
我有一个类，它在一个低级库中封装了一些异步行为。考虑一下读取或写入网络的东西。当类启动时，它将尝试连接到环境，当成功时，它将通过从工作线程调用来返回信号。
我想把这个异步行为转变成一个同步调用，我已经创建了一个非常简化的例子，下面是如何实现的：

ManualResetEvent readyEvent = new ManualResetEvent(false);

public void Start(TimeSpan timeout) {
  // Simulate a background process
  ThreadPool.QueueUserWorkItem(_ => AsyncStart(TimeSpan.FromSeconds(1)));
  // Wait for startup to complete.
  if (!this.readyEvent.WaitOne(timeout))
    throw new TimeoutException();
}

void AsyncStart(TimeSpan delay) {
  Thread.Sleep(delay); // Simulate startup delay.
  this.readyEvent.Set();
}

在工作线程上运行AsyncStart只是模拟库的异步行为的一种方式，而不是我的真实的代码的一部分，在实际代码中，低级库提供线程并在回调上调用我的代码。
请注意，如果start没有在超时间隔内完成，Start方法将抛出TimeoutException。
我想重写这段代码来使用Rx。下面是我的第一次尝试：

Subject<Unit> readySubject = new Subject<Unit>();

public void Start(TimeSpan timeout) {
  ThreadPool.QueueUserWorkItem(_ => AsyncStart(TimeSpan.FromSeconds(1)));
  // Point A - see below
  this.readySubject.Timeout(timeout).First();
}

void AsyncStart(TimeSpan delay) {
  Thread.Sleep(delay);
  this.readySubject.OnNext(new Unit());
}

这是一个不错的尝试，但不幸的是它包含一个竞态条件。如果启动完成 * 快 *（例如，如果delay为0），并且如果在点A处存在附加延迟，则在First执行之前，将在readySubject上调用OnNext。m应用Timeout和First时，不会看到启动已经完成，而是会抛出一个TimeoutException。
Observable.Defer似乎是为了处理这样的问题而创建的。下面是使用Rx的稍微复杂一点的尝试：

Subject<Unit> readySubject = new Subject<Unit>();

void Start(TimeSpan timeout) {
  var ready = Observable.Defer(() => {
    ThreadPool.QueueUserWorkItem(_ => AsyncStart(TimeSpan.FromSeconds(1)));
    // Point B - see below
    return this.readySubject.AsObservable();
  });
  ready.Timeout(timeout).First();
}

void AsyncStart(TimeSpan delay) {
  Thread.Sleep(delay);
  this.readySubject.OnNext(new Unit());
}

现在，异步操作不会立即启动，而是仅在使用IObservable时启动。遗憾的是，仍然存在争用条件，但这次是在B点。如果异步操作在Defer lambda返回之前启动调用OnNext，则它仍然丢失，并且Timeout将抛出TimeoutException。
我知道我可以使用像Replay这样的操作符来缓冲事件，但是我的初始示例没有使用任何类型的缓冲。有没有办法让我使用Rx来解决我的问题而不出现争用条件？本质上，只有在IObservable连接到Timeout和First之后才开始异步操作？
根据Ana Betts的回答，以下是可行的解决方案：

void Start(TimeSpan timeout) {
  var readySubject = new AsyncSubject<Unit>();
  ThreadPool.QueueUserWorkItem(_ => AsyncStart(readySubject, TimeSpan.FromSeconds(1)));
  // Point C - see below
  readySubject.Timeout(timeout).First();
}

void AsyncStart(ISubject<Unit> readySubject, TimeSpan delay) {
  Thread.Sleep(delay);
  readySubject.OnNext(new Unit());
  readySubject.OnCompleted();
}

有趣的是，当C点的延迟比AsyncStart完成所需的时间长时，AsyncSubject会保留最后发送的通知，Timeout和First仍将按预期执行。