F＃RX将异步模式转换为可观察序列

Question

我想将RX的一些C＃转换为F＃代码。 以下是C＃代码：

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(); 

以下是我在F＃代码：

#light 
open System 
open System.Collections.Generic 
open System.Linq 
open System.Reactive 
open System.Reactive.Linq 
open System.Reactive.Subjects 
open System.Threading 
open System.IO 

let inputStream: Stream = Console.OpenStandardInput() 
let read = Observable.FromAsyncPattern<byte[], int, int, int>(inputStream.BeginRead, inputStream.EndRead) 
let someBytes: byte[] = Array.create 10 (byte(0)) 
let source: IObservable<int> = read(someBytes, 0, 10) 

let subscription:IDisposable = 
       source.Subscribe((fun x -> printfn "OnNext %A" x), 
           (fun ex -> printfn "OnError %A" ex.Message), 
           (fun _ -> printfn "OnComplete")) 
Console.ReadKey() |> ignore 

，但我得到了以下编译器错误：

Error 1 No overloads match for method 'FromAsyncPattern'. The available overloads are shown below (or in the Error List window).

Error 2 Possible overload: 'Observable.FromAsyncPattern<'TResult>(begin: Func<AsyncCallback,obj,IAsyncResult>, end: Func<IAsyncResult,'TResult>) : Func<IObservable<'TResult>>'.

Error 3 Possible overload: 'Observable.FromAsyncPattern<'T1,'TResult>(begin: Func<'T1,AsyncCallback,obj,IAsyncResult>, end: Func<IAsyncResult,'TResult>) : Func<'T1,IObservable<'TResult>>'.

看来编译器F＃没有认识到重载，但我不知道如何让F＃编译器更聪明地完成它的工作。 请给我看你的代码。 谢谢，祝你周末愉快。 John

Answer 1

问题是FromAsyncPattern的参数是代表（类型Func<...>）的代表而不是F＃函数（类型_->_）。因此，您不能只传递一个方法（如inputStream.BeginRead），但需要传递一个显式函数（例如fun a b c d e -> inputStream.BeginRead(a,b,c,d,e)）或显式构造一个委托（例如System.Func<_,_,_,_,_,_>(fun a b c d e -> inputStream.BeginRead(a,b,c,d,e))）。

在你的情况，我认为

let read = Observable.FromAsyncPattern(
      (fun a b c d e -> inputStream.BeginRead(a,b,c,d,e)), 
      fun r -> inputStream.EndRead r) 

应该工作。