最後にストリームを読み込むObservableを作成する適切な方法は何ですか

Question

私はここで苦労しています。通常、私は本を読んだが、まだ何もない。私は、RXを使ってストリームを読むこととはさまざまなことの無数の例を見出しましたが、私は頭を悩ますのが非常に難しいと感じています。

Observable.FromAsyncPatternを使用して、ストリームのBeginRead/EndReadメソッドまたはBeginReadLine/EndReadLineメソッドのラッパーを作成することができます。

しかし、これは最初の観察者が購読したときに一度だけ読み込みます。

ストリームのエラーまたは終了までOnNextを読み込み、ポンピングするObservableが必要です。

これに加えて、私はどのようにして複数のサブスクライバと観測値を共有してアイテムを取得できるのかを知りたいと思います。

Answer 1

ソリューションは、ここで

をObservable.Create使用することです（接続を呼び出すことを忘れないでくださいストリームの任意の種類

public static IConnectableObservable<Command> GetReadObservable(this CommandReader reader) 
    { 

     return Observable.Create<Command>(async (subject, token) => 
     { 


      try 
      { 

       while (true) 
       { 

        if (token.IsCancellationRequested) 
        { 
         subject.OnCompleted(); 
         return; 
        } 

        //this part here can be changed to something like this 
        //int received = await Task.Factory.FromAsync<int>(innerSocket.BeginReceive(data, offset, size, SocketFlags.None, null, null), innerSocket.EndReceive); 

        Command cmd = await reader.ReadCommandAsync(); 

        subject.OnNext(cmd); 

       } 

      } 

      catch (Exception ex) 
      { 
       try 
       { 
        subject.OnError(ex); 
       } 
       catch (Exception) 
       { 
        Debug.WriteLine("An exception was thrown while trying to call OnError on the observable subject -- means you're not catching exceptions everywhere"); 
        throw; 
       } 
      } 

     }).Publish(); 

    } 

を読み取るためadapatedことができる一例です）返さIConnectableObservable

Answer 2

rxxを使用して、李氏の答えに追加：

using (new FileStream(@"filename.txt", FileMode.Open) 
     .ReadToEndObservable() 
     .Subscribe(x => Console.WriteLine(x.Length))) 
{ 
    Console.ReadKey(); 
} 

リードバッファの長さが出力されます。

Answer 3

へぇ - つもりリユースここに私の他の回答（それのほか、一部、とにかく）の1：

参考：

：それは Reading from NetworkStream corrupts the buffer

が、私はこのような拡張メソッドを持っています

public static class Ext 
{   
    public static IObservable<byte[]> ReadObservable(this Stream stream, int bufferSize) 
    {   
     // to hold read data 
     var buffer = new byte[bufferSize]; 
     // Step 1: async signature => observable factory 
     var asyncRead = Observable.FromAsyncPattern<byte[], int, int, int>(
      stream.BeginRead, 
      stream.EndRead); 
     return Observable.While(
      // while there is data to be read 
      () => stream.CanRead, 
      // iteratively invoke the observable factory, which will 
      // "recreate" it such that it will start from the current 
      // stream position - hence "0" for offset 
      Observable.Defer(() => asyncRead(buffer, 0, bufferSize)) 
       .Select(readBytes => buffer.Take(readBytes).ToArray())); 
    } 
} 

おそらく、そうのような形式で記述されたとして、これを使用することができます。

// Note: ToEnumerable works here because your filestream 
// has a finite length - don't do this with infinite streams! 
var blobboData = stream 
    .ReadObservable(bufferSize) 
    // take while we're still reading data 
    .TakeWhile(returnBuffer => returnBuffer.Length > 0) 
    .ToEnumerable() 
    // mash them all together 
    .SelectMany(buffer => buffer) 
    .ToArray(); 

Answer 4

上のあなた複数のリーダー間の共有を制御するために、ストリームの終わりまでラインを読み取るためにRepeatを使用し、PublishまたはReplayを使用することができます。

端までの任意のストリームから行を読み取るための簡単な、完全な受信溶液の例は、次のようになります

public static IObservable<string> ReadLines(Stream stream) 
{ 
    return Observable.Using(
     () => new StreamReader(stream), 
     reader => Observable.FromAsync(reader.ReadLineAsync) 
          .Repeat() 
          .TakeWhile(line => line != null)); 
} 

この解決策はまた、ReadLine戻りnullの場合、ストリームの終了事実を利用しますが達成された。