缓冲流侦听器的值以使用每个单个流值提供异步函数

如何解决缓冲流侦听器的值以使用每个单个流值提供异步函数

我创建了一个流侦听器来将服务器数据与设备上的本地文件同步。在启动时，它会查找比设备上看到的数据更新的数据。但是，在启动时，这或多或少会同时返回一堆数据。

在函数完成第一次写入过程之前，这些数据似乎会调用我的函数多次将任何新的流事件写入本地文件。

听众：

// =================================================================== SYNC DATA
  void syncCustomers() async {
    DateTime lastCustomerChange = DateTime(2000,1,1);
    final List<Customer> customersLocal = await getAllLocalCustomers();
    if (customersLocal != null)
      for (var c in customersLocal) {
        if (lastCustomerChange.isBefore(c.changeDate))
          lastCustomerChange = c.changeDate;
      }
    _customeRSSubscription = customerCollection.where('changeDate',isGreaterThan: lastCustomerChange.toIso8601String())
        .snapshots().listen(
          (QuerySnapshot snapshot){snapshot.docs.forEach((document) {
            storeCustomerLocally(Customer.fromEntity(CustomerEntity.fromSnapshot(document)));
        });}
    );

存储流事件

  Future<void> storeCustomerLocally(Customer c) async {
    const JsonEncoder encoder = JsonEncoder.withIndent('  ');
    final cFile = await _getLocalFile('customers.json');

    List<Customer> customers = await getAllLocalCustomers();
    if (customers == null) {
      customers = [c];
    } else {
      customers.add(c);
    }
    final cJsonString = encoder.convert(customers);
    await cFile.writeAsstring(cJsonString);
    print('a');
    return vectorFile.writeAsstring(cJsonString);
  }

我在这里删除了一些代码，以便于阅读。基本上我将每个客户的数据分成 2 个文件，这就是为什么我在函数末尾有一行 await cFile... 和 return vectorFile...。启动时，仅使用最后一个流值到达打印语句。对于前面的语句，函数在到达 await 语句时重新启动，从而仅将最后一个流值存储在文件中。

更新 ================================

基于 pskinks 有用的输入，我将监听器修改为

final _customersAsyncStream = customerCollection.where('changeDate',isGreaterThan: lastCustomerChange.toIso8601String())
        .snapshots().asyncMap((snapshot) async {
          await snapshot.docs.forEach((document) async {
      await storeCustomerLocally(Customer.fromEntity(CustomerEntity.fromSnapshot(document)));
    });
    }).drain();

使用存储每个流事件的函数，我还尝试返回一个 bool 作为 Future 结果（在上面的代码中它是无效的），看看这是否有区别 - 没有。

对于每次调用该函数的事件，与我最初的问题相反，第一次运行时会创建一个文档。但是，它是空的，在 await cFile... 语句之后，它会使用存储函数重新启动。然后在解析所有事件后，它再次开始循环打印语句（如调试器中所示）。然后输出文件也填充了数据。

除了那个文件中的一些奇怪的 JSON 错误与我存储的 JSON 字符串不对应之外，我需要该文件中每个事件循环的数据作为下一个流事件循环的输入。

我真的不明白为什么每个函数调用都在那个 await 语句处分开、中断和继续。