问题标签 [iterate]

我目前正试图围绕 and 的想法来Enumerators思考Iteratees。我决定从查看 Play 2.0 的 iteratee 库开始，我已将其添加到我的测试项目中，并在我的build.sbt文件中添加了以下行。（我使用的是 Scala 2.10）（此处的文档）

我的目标是在文件的字节上创建一个枚举器，并最终为其附加一些解析逻辑，但是当我尝试看似简单的事情时，我得到了一个异常。我的代码如下所示：

发生的事情是（我假设是）文件中的所有字节都被打印出来，然后我得到一个IllegalStateException说法，“承诺已经完成”。

由于堆栈跟踪没有指向我的代码中的任何地方，而且这是一个不熟悉的领域，我不知道出了什么问题。任何人都可以为这个问题提供一些见解或解决方案吗？

我在 Haskell 中有下一个代码，这对于读取文件的第一行很好，但是我需要读取目录中文件的所有内容（递归地读取许多文件）。我正在尝试更改 firstLineE 函数，我不明白如何更改行： EIO.enumFile 1024 filename $ joinI $ ((mapChunks B.pack) ><> EC.enumLinesBS) 。你有这方面的一些文件，或者你能帮我举一些例子吗？

我正在查看文档，但 Iteratee 对我来说非常新：

http://www.mew.org/~kazu/proj/enumerator/

http://blog-mno2.csie.org/blog/2011/11/19/yet-another-guide-to-understand-iteratee-for-haskell-programmers/

我见过的 Scala 中 Iteratee 模式的 3 个描述都包括 3 个输入案例。例如，来自詹姆斯：

更多详细信息请参见James、Runar、Josh的博客。

我的问题很简单：为什么需要Empty输入案例？

迭代模式定义了值流的生产者和消费者之间的关系。直观地说，如果任何输入为空，“运行”迭代器的生产者应该简单地折叠该空项，并且在非空输入可用之前不调用迭代器。

我注意到基于 pull 的 iteratees 类比，更熟悉的迭代器，没有定义一个空的情况，尽管元素可能已经在迭代器的“内部”被过滤掉了。

虽然上述所有博客都提到需要一个 Empty 输入，但他们没有明确讨论为什么不能完全消除它。我注意到显示的示例迭代器将空输入视为无操作。

我真的很想要一个带有代码的示例，说明一个看似合理的“现实世界”问题，需要 Empty 输入消息来解决。

介绍

我在许多项目中使用Scalaz 7的迭代器，主要用于处理大型文件。我想开始切换到 Scalaz流，它旨在替换 iteratee 包（坦率地说，它缺少很多部分并且使用起来有点痛苦）。

流基于机器（迭代思想的另一种变体），它也已在 Haskell 中实现。我使用了一点 Haskell 机器库，但是机器和流之间的关系并不完全明显（至少对我而言），流库的文档仍然有点稀疏。

这个问题是关于一个简单的解析任务，我希望看到使用流而不是迭代器来实现。如果没有其他人超过我，我会自己回答这个问题，但我确信我不是唯一一个正在进行（或至少考虑）这种转变的人，因为无论如何我都需要完成这个练习，我想我还不如在公共场合做。

任务

假设我有一个文件，其中包含已标记化并用词性标记的句子：

每行有一个标记，单词和词性由一个空格分隔，空行表示句子边界。我想解析这个文件并返回一个句子列表，我们不妨将其表示为字符串元组的列表：

像往常一样，如果我们遇到无效输入或文件读取异常，我们希望优雅地失败，我们不想担心手动关闭资源等。

一个迭代的解决方案

首先是一些通用的文件读取内容（它确实应该是 iteratee 包的一部分，它目前不提供任何远程这个高级别的东西）：

然后我们的句子阅读器：

最后是我们的解析动作：

我们可以证明它有效：

我们完成了。

我想要的是

或多或少是使用 Scalaz 流而不是迭代器实现的相同程序。

我正在使用带有 scala 的 scalaz 6.0。我正在使用 iteratees 从输入流中读取。

这是我拥有的名为 simple.txt 的简单文件。

这
是
一个测试

我的 iteratee 将建立一个 io monad 来打印这些行

当我使用 enumeratorM

我检索到正确的输出。

当我尝试使用

我只会将“This”返回到控制台。getLines 使用标准输入流。我已经向 iteratee 添加了调试语句，并且 getLines 似乎在第一行之后发送 EOF() 并且我无法解决它。

在 play WS 库中，我使用此调用来处理流式 http 响应：

我正在传递类似的东西：_ => (Iteratee.foreach(chunk => println(chunk)))

一切正常，但在某些时候连接似乎关闭了，我不知道如何处理。.mapDone当 Iteratee 完成时，我尝试添加以打印出一些东西，但它从未发生过。

在此获取请求中，我如何检测到连接已关闭并处理该事件？

Iteratee是否可以在 Java中使用 Play 的s？我找不到任何关于Iteratee在Java中使用的示例和文档，只有 Scala。我猜让 Iteratees 使用 PLay API 在 Java 中工作会有点混乱的代码（很多匿名Funtion1<?,>）......

如果可能的话，我想创建一个 App 控制器，它可以接受通过 HTTPs 分块传输编码上传的多部分文件上传，并将这些消息块向下游解析到 S3 存储。关于如何在 Java 中解决这个问题的任何想法？

干杯。

我一直在使用 Scalaz 7 迭代器来处理恒定堆空间中的大型（即无界）数据流。

在代码中，它看起来像这样：

现在我想并行执行处理，一次处理P块数据。我仍然必须限制堆空间，但可以合理地假设有足够的堆来存储P块数据和计算的累积结果。

我知道Task该类并考虑映射枚举器以创建任务流：

但我仍然不确定如何管理非确定性。在消费流时，如何确保P任务尽可能运行？

第一次尝试：

我的第一个解决方案是折叠流并创建一个 ScalaFuture来处理每个块。但是，该程序因 GC 开销错误而崩溃（可能是因为它在尝试创建所有Futures 时将所有块拉入内存）。相反，当已经有P个任务在运行时，迭代者需要停止使用输入，并在这些任务中的任何一个完成时再次恢复。

第二次尝试：

我的下一个尝试是将流分组为P大小的部分，并行处理每个部分，然后在继续下一部分之前加入：

虽然这不会充分利用可用的处理器（特别是因为处理每个处理器所需的时间Chunk可能差别很大），但这将是一个改进。但是，枚举group对象似乎泄漏了内存——堆使用率突然飙升。

我正在使用 scala 和 Play 编写流式网络广播框架。我依靠 Iteratees 进行实际的流式传输，但我遇到了一个问题，试图阻止贪婪的客户端过快地下载数据，并为所有客户端消耗流。为此，我一直在尝试创建一个 Enumeratee，它将限制 Enumerator 生成数据的速度。这是我的 Enumeratee 的样子

这个想法是我使用计时器任务来创建一个 throttlingIteratee 并将其与 Enumeratee.grouped 函数配对。这似乎工作得很好，但我无法弄清楚块大小使用什么值。我想让它以与音频播放大致相同的速率产生块。我的音频文件以 82kpbs 编码，我尝试以字节为单位进行计算，但我得出的值似乎太小了，而且音频播放的速度比数据流传输的速度快。

我的问题是两方面的。我采用的基本方法是一个好的方法吗？如果是，我该如何根据音频文件的比特率设置块大小。

背景

如this question中所述，我正在使用Scalaz 7迭代来处理恒定堆空间中的大型（即无界）数据流。

我的代码如下所示：

问题

我似乎遇到了内存泄漏，但我对 Scalaz/FP 不够熟悉，无法知道该错误是在 Scalaz 中还是在我的代码中。直观地说，我希望这段代码只需要（大约）P倍的Chunk-size 空间。

注意：我发现了一个类似的问题，其中OutOfMemoryError遇到了 an ，但我的代码没有使用consume.

测试

我进行了一些测试以尝试隔离问题。zipWithIndex总而言之，只有在使用和时才会出现泄漏group。

测试代码：

问题

• 我的代码中有错误吗？
• 我怎样才能使它在恒定的堆空间中工作？