async Zadanie <IEnumerable> z zwrotem plonów?

Question

Poniższa metoda nie jest kompilowana. Alternatywy? Komunikat

public static async Task<IEnumerable<object[]>> GetRecordsAsync(
    this Transaction transaction, 
    string commandText, 
    params SqlParameter[] parameters) 
{ 
    // Get a SqlDataReader 
    var reader = await transaction.GetReaderAsync(commandText, parameters); 
    var fieldCount = -1; 
    // Begin iterating through records asynchronously 
    while (await reader.ReadAsync()) // Note we don't loop until .ReadAsync returns a boolean 
    { 
     // Grab all the field values out 
     if (fieldCount < 0) 
      fieldCount = reader.FieldCount; 
     var fields = new object[fieldCount]; 
     reader.GetValues(fields); 
     // Yield return the field values from this record 
     yield return fields; 
    } 
} 

błędu:

Ciało 'TransactionExtensions.GetRecordsAsync (transakcji, łańcuch, params SqlParameter [])' nie może być blok iteracyjnej, ponieważ 'zadania>' jest typu interfejsu iteracyjnej

nie widzę sposobu na dostosowanie this answer do podobnego brzmienia (ale inny) pytanie, bo nie wiem a priori ile razy pętla pójdzie.

Edit: stałe formatowania

Answer 1

podstawie komentarzu @ SLaks koszulka na pytanie, oto ogólna alternatywa korzystania Reactive Extensions:

/// <summary> 
/// Turns the given asynchronous functions into an IObservable 
/// </summary> 
static IObservable<T> ToObservable<T>(
    Func<Task<bool>> shouldLoopAsync, 
    Func<Task<T>> getAsync) 
{ 
    return Observable.Create<T>(
     observer => Task.Run(async() => 
      { 
       while (await shouldLoopAsync()) 
       { 
        var value = await getAsync(); 
        observer.OnNext(value); 
       } 
       observer.OnCompleted(); 
      } 
     ) 
    ); 
} 

Przykład użycia, dostosowane do rozwiązywania konkretnego przypadku kwestia to:

/// <summary> 
/// Asynchronously processes each record of the given reader using the given handler 
/// </summary> 
static async Task ProcessResultsAsync(this SqlDataReader reader, Action<object[]> fieldsHandler) 
{ 
    // Set up async functions for the reader 
    var shouldLoopAsync = (Func<Task<bool>>)reader.ReadAsync; 
    var getAsync = new Func<SqlDataReader, Func<Task<object[]>>>(_reader => 
    { 
     var fieldCount = -1; 
     return() => Task.Run(() => 
     { 
      Interlocked.CompareExchange(ref fieldCount, _reader.FieldCount, -1); 
      var fields = new object[fieldCount]; 
      _reader.GetValues(fields); 
      return fields; 
     }); 
    })(reader); 

    // Turn the async functions into an IObservable 
    var observable = ToObservable(shouldLoopAsync, getAsync); 

    // Process the fields as they become available 
    var finished = new ManualResetEventSlim(); // This will be our signal for when the observable completes 
    using (observable.Subscribe(
     onNext: fieldsHandler, // Invoke the handler for each set of fields 
     onCompleted: finished.Set // Set the gate when the observable completes 
    )) // Don't forget best practice of disposing IDisposables 
     // Asynchronously wait for the gate to be set 
     await Task.Run((Action)finished.Wait); 
} 

(Zauważ, że getAsync można uprościć w powyższym bloku kodu, ale podoba mi się to, jak wyraźne jest ono w odniesieniu do zamknięcie, który jest tworzony)

... i wreszcie:

// Get a SqlDataReader 
var reader = await transaction.GetReaderAsync(commandText, parameters); 
// Do something with the records 
await reader.ProcessResultsAsync(fields => { /* Code here to process each record */ });