生成多个 IEnumerable

Question

我有一段代码可以对资产进行计算。有数百万个，所以我想计算流中的所有内容。我当前的“管道”如下所示：

我有一个作为数据读取器执行的查询。

然后我的 Asset 类有一个接受 IDataReader 的构造函数；

Public Asset(IdataReader rdr){
  // logic that initiates fields
}

以及将 IDataReader 转换为 IEnumerable的方法。

public static IEnumerable<Asset> ToAssets(IDataReader rdr) {

    // make sure the reader is in the right formt
    CheckReaderFormat(rdr);

    // project reader into IEnumeable<Asset>
    while (rdr.Read()) yield return new Asset(rdr);

}

然后将其传递到执行实际计算的函数中，然后将其投影到 IEnumerable中。

然后获取一个包装器，将答案公开为 IDataReader，然后将其传递给 OracleBulkCopy，并将流写入数据库。

到目前为止，它就像一个魅力。由于设置的原因，我可以将 DataReader 替换为从文件读取的 IEnumerable，或者将结果写入文件等。所有这些都取决于我如何将类/函数串在一起。

现在：我可以计算一些东西，例如，除了正常的答案之外，我还可以有一个 DebugAnswer 类，它还输出一些用于调试的中间数字。所以我想做的是将 IEnumerable 投影到多个输出流中，这样我就可以在这些输出流上放置“侦听器”。这样我就不必多次检查数据。我怎样才能做到这一点？有点像有多个事件，然后仅在附加侦听器时才触发某些代码。

有时我也会写入数据库，但也会写入 zip 文件，只是为了保留结果的备份。那么我想在 IEnumerable 上有 2 个“监听器”。一个项目是作为 IDataReader，另一个项目是直接写入文件。

如何输出多个输出流以及如何在一个输出流上放置多个侦听器？是什么让我能够组成这样的数据流？

编辑

所以我想做一些伪代码：

foreach(Asset in Assets){
   if(DebugListener != null){
     // compute 
     DebugAnswer da = new DebugAnswer {result = 100};
     yield da to DebugListener;  // so instead of yield return yield to that stream

   }

   if(AnswerListener != null){
     // compute basic stuff 
     Answer a = new Answer { bla = 200 };
     yield a to AnswerListener;
   }
}

提前致谢，

Gert-Jan

Answer 1

您所描述的内容听起来有点像响应式框架通过 IObservable 接口提供的内容，但我不确定它是否允许多个订阅者访问单个订阅流。

更新

如果您查看IObservable的文档 ，它有一个很好的示例，说明如何使用单个对象的多个订阅者来完成您正在做的事情。

Answer 2

使用 Rx 重写的示例：

// The stream of assets
IObservable<Asset> assets = ...

// The stream of each asset projected to a DebugAnswer
IObservable<DebugAnswer> debugAnswers = from asset in assets
                                        select new DebugAnswer { result = 100 };

// Subscribe the DebugListener to receive the debugAnswers
debugAnswers.Subscribe(DebugListener);

// The stream of each asset projected to an Anwer
IObservable<Answer> answers = from asset in assets
                              select new Answer { bla = 200 };

// Subscribe the AnswerListener to receive the answers
answers.Subscribe(AnswerListener);

Answer 3

这正是响应式扩展（成为.NET的一部分）的工作从 4.0 开始，在 3.5 中作为库提供）。

Answer 4

您不需要多个“侦听器”，您只需要非破坏性甚至不一定可转换的管道组件。

IEnumerable<T> PassThroughEnumerable<T>(IEnumerable<T> source, Action<T> action) {
    foreach (T t in source) {
       Action(t);
       yield return t;
    }    
}

或者，当您在管道中处理时，只需引发一些要使用的事件。如果需要，您可以异步它们：

static IEnumerable<Asset> ToAssets(IDataReader rdr) {
   CheckReaderFormat(rdr);
   var h = this.DebugAsset;
   while (rdr.Read()) {
      var a = new Asset(rdr);
      if (h != null) h(a);
      yield return a;
   }
}

public event EventHandler<Asset> DebugAsset;

Answer 5

如果我的理解正确，应该可以替换或装饰包装器。 WrapperDecorator 可能会将调用转发到普通的 OracleBulkCopy （或您正在使用的任何内容）并添加一些自定义调试代码。

这对你有帮助吗？

马蒂亚斯