如何避免在下一个完成之前调用 onComplete？答案

【问题标题】：How to avoid calling onComplete before on next is finished?如何避免在下一个完成之前调用 onComplete？
【发布时间】：2017-03-07 10:03:54
【问题描述】：

假设我有一个数据服务类，它可以批量获取数据并将其数据块发送给订阅者。

public class DataService {
   public IObservable<IList<T>> QuerySegmentedObservable<T>(string tableName)  where T : TableEntity, new(){

       return Observable.Create<IList<T>>(async (observer, token) =>{
           TableContinuationToken continuationToken = null;
           do{
               var currentSegment = CallData();
               observer.OnNext(currentSegment.Results);

               continuationToken = currentSegment.ContinuationToken;
           } while (continuationToken != null);
           observer.OnCompleted();
       } 
   }
 }

我正在订阅这个 observable，如下所示。

 public async Task<bool> MyMethod()
        {
            var tcs = new TaskCompletionSource<bool>();
            var observable = _dataService.QuerySegmentedObservable<TSource>(_sourceTableName);

            var dataCount = 0;

             _databaseService.OpenConnection();

            observable.Subscribe(async data =>
            {                
                await _databaseService.DoSomething(data);

                dataCount += data.Count;

                Console.WriteLine($"Processing - {dataCount}");
            },
            err =>
            {
                Console.WriteLine($"Error - {err.Message}");
                tcs.SetResult(false);
            },
            () =>
            {    
                _databaseService.CloseConnection();
                Console.WriteLine($"Finished");
                tcs.SetResult(true);
            }
            );

            return await tcs.Task;
        }

问题是在最后一个 OnNext() 完成之前调用了 OnComplete()。所以，我在完成我在订阅（）中所做的任务之前关闭了连接；

有什么办法可以解决吗？谢谢。

【问题讨论】：

尝试从 onNext 块中删除异步操作。
_databaseService的类型是什么？
@En 它只是一个将数据插入数据库或执行可能需要几秒钟到 40 秒的操作的类。
@redent84 我以为新版的rx支持async/await，不是吗？

标签： c# system.reactive reactive-programming

【解决方案1】：

Rx 确实支持运算符中的 async/await。您在订阅中使用它。所以（希望）你可以把你的代码改成这样：

public async Task<bool> MyMethod()
{
    var tcs = new TaskCompletionSource<bool>();
    _databaseService.OpenConnection();
    var dataCount = 0;
    _dataService.QuerySegmentedObservable<TSource>(_sourceTableName)
        .SelectMany(async data =>
        {
            await _databaseService.DoSomething(data);
            return data;
        })
        //.Finally(() => _databaseService.CloseConnection()) //This would be called on OnComplete and OnError, just like try-finally
        .Subscribe(data =>
            {
                dataCount += data.Count;

                Console.WriteLine($"Processing - {dataCount}");
            },
            err =>
            {
                Console.WriteLine($"Error - {err.Message}");
                tcs.SetResult(false);
            },
            () =>
            {
                _databaseService.CloseConnection(); //Maybe move this to a Finally call?
                Console.WriteLine($"Finished");
                tcs.SetResult(true);
            }
        );

    return await tcs.Task;
}

我无法真正测试它，所以我希望这能让你走上正确的道路。如果您需要更多帮助，请发一个更好的MCVE。

【讨论】：