IObservable - 替换AutoResetEvent

Question

想知道如何在下面代替AutoResetEvent？我试图考虑如何使用RX方式或使用任务，但我可以看到如何去做。

public void LogOnResponse LogOn() 
{ 

    LogOnResponse logOnResponse = null; 
    var waitEvent = new AutoResetEvent(false); 

    var listener = _connection.LoginStatusEvent 
         .Where(x => x.LoginState == LoginState.LoggedOn 
            || x.LoginState == LoginState.LoggedRejected); 

    listener.Subscribe(x => { 
          logOnResponse = new LogOnResponse(); 
          logOnResponse.InformationMessage = x.Message; 
          logOnResponse.IsAuthenticated = x.LoginState == LoginState.LoggedOn; 
          waitEvent.Set(); 
         }); 

    connection.Login(connectionInfo); 

    waitEvent.WaitOne(2000); 

    return logOnResponse; 
} 

Answer 1

考虑结构化，像这样的代码：

var listener = _velocityConnection.LoginStatusEvent 
    .Where(x => x.LoginState == LoginState.LoggedOn || x.LoginState == LoginState.LoggedRejected); 

var logOnResponse = listener.Select(x => 
    new LogOnResponse() { 
     logOnResponse.InformationMessage = x.Message; 
     logOnResponse.IsAuthenticated = x.LoginState == LoginState.LoggedOn; 
    } 
).Timeout(TimeSpan.FromSeconds(2), Observable.Return(new LogOnResponse() {...})) 
    .Publish() 
    .RefCount(); 

connection.Login(connectionInfo); 
return logOnResponse.First(); 

我们使用Publish这么执行的顺序并不重要，并引用次数（）隐藏ConnectableObservable

Answer 2

我会该方法返回一个IObservable<LogOnResponse>，所以你不需要等到你准备好返回。您还必须留意比赛条件，因为您的登录电话可能会在您有机会订阅之前完成。像这样的东西应该处理竞争条件，并通过返回的IObservable返回响应。你可以明显地订阅这个来获得LogOnResponse，一旦它出现。

public IObservable<LogOnResponse> LogOn() 
{ 
    return Observable.CreateWithDisposable<LogOnResponse>(observer => 
    { 
     var loginDisposable = _velocityConnection.LoginStatusEvent 
         .Where(x => x.LoginState == LoginState.LoggedOn 
            || x.LoginState == LoginState.LoggedRejected) 
         .Select(x => { 
          logOnResponse = new LogOnResponse(); 
          logOnResponse.InformationMessage = x.Message; 
          logOnResponse.IsAuthenticated = x.LoginState == LoginState.LoggedOn; 
          return logOnResponse; 
         }).Take(1).Subscribe(observer); 

     connection.Login(connectionInfo); 

     return loginDisposable; 
    }); 
} 

在这里需要注意的一件事是调用此方法只会返回IObservable。直到您订阅，才会调用实际登录。

// Gets the IObservable<LogOnResult> 
var logonResponseAsObservable = LogOn(); 
//Execute the logon and wait for a response asynchronously 
logonResponseAsObservable.Subscribe(response => HandleLogOnResponse(response));