async/await, RX 및 LINQ를 사용한 비동기 메시지 처리

Question

소켓 프로토콜 구현을 단순화하기 위해 Reactive Extensions를 async/await과 함께 사용하고 있습니다. 특정 메시지가 도착할 때 (예 : 각 '핑'메시지에 'pong'보내기) 수행해야하는 작업이 있으며 특정 응답을 비동기 적으로 기다려야하는 방법도 있습니다.

private Subject<string> MessageReceived = new Subject<string>(); 

//this method gets called every time a message is received from socket 
internal void OnReceiveMessage(string message) 
{ 
    MessageReceived.OnNext(message); 
    ProcessMessage(message); 
} 

public async Task<string> TestMethod() 
{ 
    var expectedMessage = MessageReceived.Where(x => x.EndsWith("D") && x.EndsWith("F")).FirstOrDefaultAsync(); 
    await SendMessage("ABC"); 

    //some code... 

    //if response we are waiting for comes before next row, we miss it 
    return await expectedMessage; 
} 

TestMethod() 소켓에 "ABC"를 보내고 예 "DEF"를 수신 할 때 (그 전에 몇 가지 다른 메시지가있을 수 있습니다) 계속 다음 예는이 보여줍니다.

거의 작동하지만 경쟁 조건이 있습니다. 이 코드는 return await expectedMessage;까지 메시지를 수신하지 않는 것으로 보입니다.이 메시지는 그 전에 메시지가 도착하기 때문에 문제입니다.

Answer 1

FirstOrDefaultAsync 여기서는 잘 작동하지 않습니다. 귀하가 지적한대로 경쟁 조건을 벗어나는 await 라인까지 구독하지 않습니다. 여기 당신이 그것을 대체 할 수있는 방법은 다음과 같습니다

var expectedMessage = MessageReceived 
     .Where(x => x.EndsWith("D") && x.EndsWith("F")) 
     .Take(1) 
     .Replay(1) 
     .RefCount(); 

    using (var dummySubscription = expectedMessage.Subscribe(i => {})) 
    { 
     await SendMessage("ABC"); 

     //Some code... goes here. 

     return await expectedMessage; 
    } 

.Replay(1) 하나가있는 가정, 새 구독은 가장 최근의 항목을 얻을 수 있는지 확인합니다. 듣고있는 가입자가있는 경우에만 작동하지만, 따라서 dummySubscription입니다.