Ponte com fontes assíncronas existentes

Além dos eventos do .NET, existem outras fontes de dados assíncronas no .NET Framework. Um deles é o padrão de método assíncrono. Nesse padrão de design, dois métodos são fornecidos. Um método (geralmente chamado BeginX) é usado para iniciar a computação e retorna um identificador IAsyncResult que é passado para o segundo método (geralmente chamado EndX), que, em seguida, recupera o resultado da computação. A conclusão geralmente é sinalizada implementando um delegado AsyncCallback ou um IAsyncResult.IsCompleted de sondagem. A adesão de código a esse padrão geralmente é difícil de ler e manter. Neste tópico, mostraremos como usar métodos de fábrica Rx para converter fontes de dados assíncronas em sequências observáveis.

Convertendo padrões assíncronos em sequências observáveis

Muitos métodos assíncronos no .NET são escritos com assinaturas como BeginX e EndX, em que X é o nome do método que está sendo executado de forma assíncrona. BeginX usa argumentos para executar o método, um AsyncCallback que é uma ação que usa um IAsyncResult e não retorna nada e, por fim, um estado de objeto. EndX usa o IAsyncResult que é passado do AsyncCallback para recuperar o valor da chamada assíncrona.

O operador FromAsyncPattern do tipo Observável encapsula os métodos Begin e End (que estão sendo passados como parâmetros para o operador) e retorna uma função que usa os mesmos parâmetros que Begin e retorna um observável. Esse observável representa uma sequência que publica um único valor, que é o resultado assíncrono da chamada que você acabou de especificar.

No exemplo a seguir, converteremos BeginRead e EndRead para um objeto Stream que usa o padrão IAsyncResult em uma função que retorna uma sequência observável. Para os parâmetros genéricos do operador FromAsyncPattern, especificamos os tipos dos argumentos de BeginRead até o retorno de chamada. Como o método EndRead retorna um valor, acrescentamos esse tipo como o parâmetro genérico final para FromAsyncPattern. Se você passar o mouse var para read, observará que o valor retornado de FromAsyncPattern é um delegado de função que tem a seguinte assinatura: Func<byte[], int32,int32, IObservable<int32>>, o que significa que essa função usa três parâmetros (os mesmos para BeginRead) e retorna um Int32> IObservable<. Esse IObservable contém um valor, que é o inteiro retornado por EndRead e contém o número de bytes lidos do fluxo, entre zero (0) e o número de bytes solicitados. Como agora obtemos um IObservable em vez de um IAsyncResult, podemos usar todos os operadores LINQ disponíveis para Observáveis e assinar, analisar ou compor.

Stream inputStream = Console.OpenStandardInput();
var read = Observable.FromAsyncPattern<byte[], int, int, int>(inputStream.BeginRead, inputStream.EndRead);
byte[] someBytes = new byte[10];
IObservable<int> source = read(someBytes, 0, 10);
IDisposable subscription = source.Subscribe(
                            x => Console.WriteLine("OnNext: {0}", x),
                            ex => Console.WriteLine("OnError: {0}", ex.Message),
                            () => Console.WriteLine("OnCompleted"));
Console.ReadKey();