问题标签 [scalaz-stream]

我有一个状态机，我想使用 scalaz-stream Process1 对其进行建模。

状态机对客户端和服务器之间的消息流进行建模。

一组基本的数据类型可能是：

在我的脑海中，这将由 a Process1[I, O]、 wheretype I = Message和type O = scalaz.State[Message, ServerState].

这就像一团迷雾——我可以看到解决方案的轮廓，但它的坚定定义却让我无法理解。

目前我有一些看起来像这样的东西

我知道这是错误的，但我不知道状态转换需要在哪里生活。

当驱动程序处理内部状态时，进程是否应该接受Message并返回物理？ServerStateProcess1

我正在努力了解如何“携带”目前无法执行的消息。

例子：

最近我开始在 Scala/Akka 中使用 scalaz 流。我正在从 nosql 数据库中获取记录。我想将记录映射到消息项（通过translateItem: Item）并创建它们的包（1 个包 = 100 个项）。例如，有 500 条记录。

我试图List[Package]通过runLog. 日志的输出看起来不错。但我不需要日志记录开销。

但是如何将结果写入列表，例如？

或者我可能会返回 aFuture[List[Package]]并将其传递给 Akka actor 以获取 ( Await.result) 它。我可能会将 scalaz 转换Task为Future.

提前致谢。

我试过这样做，但没有奏效：

我有一个两部分的问题，所以让我先介绍一些背景。我知道可以做类似于我想要的事情：

还有其他一些p1我可以使用的东西可以给我下面的输出而不挂起（就像process1.awaitOption）？

如果是，我认为回答下一个问题会很容易。还有其他一些p1我可以使用的东西可以给我下面的输出而不挂起（就像process1.chunkAll）？

编辑：

补充问题以使其更易于理解。如果我有这样的循环：

结果可能是：

我希望现在很清楚我要做什么。问题是我无法控制循环，如果队列中没有值，我不能阻止它。

我对 scalaz-stream 和特别是 scalaz.stream.tcp 都很陌生。我正在尝试为自己的教育目的做一个非常简单的服务器。我将请求解析为命令，执行它们以产生响应，然后将响应写回客户端。我遇到的问题是我想将每个收到的命令记录到标准输出。

这是我传递给 tcp.server 的内部进程：

（我没有在任何地方指定类型的习惯；我这样做只是为了尝试处理事情。我打算删除这些。）

上面的代码实际上可以正确编译和运行，但我觉得它不是很干净或惯用。具体来说，我对loggedCmds 部分不满意。我想通过 .observer 或使用 writer.logged/mapW/drainW 使用 io.stdOutLines，但无论我尝试什么，我似乎都无法正确排列类型。我总是遇到任务和连接之间的类型冲突。tcp.lift 似乎有助于输入流，但它似乎不适用于接收器。是否有更清洁/更好的方法来执行 loggedCmds 部分（FWIW：我愿意对上述任何代码进行更正或改进）。

我应该注意，如果我只是通过 io.stdOutLines 将结果发送到标准输出，我没有问题（“通过”在这种情况下似乎有效，我在示例中看到过），这只是当我想发送流到 io.stdOutLines并继续使用流响应客户端。

在以下scalaz-stream（取自文档）的使用示例中，如果输入和/或输出是 gzip 压缩文件，我需要更改什么？换句话说，我该如何使用compress？

假设我的输入文件myInput.txt如下所示：

也就是说，有文档以 . 分隔~~~。所需的输出如下：

我该怎么做？以下似乎很不自然，而且我失去了标题：

在我的应用程序中，我有多达 N 个并行工作的消费者和一个生产者。消费者从生产者那里获取资源，完成他们的工作，将结果附加到 anupdateQueue并请求更多资源。Producer 最初有一些可用资源，并且可以通过应用来自updateQueue. 在将新资源发送给消费者之前应用所有可用更新非常重要。我尝试使用以下生成器，每当消费者提出请求时“批量”请求更新，并在 a 中留出新资源（消费者不需要，但稍后可能会被其他消费者请求）ticketQueue：

它不起作用 - 最初可用的资源是从 发出的ticketQueue.dequeue，然后它似乎在wye, 日志记录中阻塞：

当最初在ticketQueue. 但是，如果我将其更改为

它按预期工作（尽管没有“在发出新资源之前应用更新”保证）。如何确保在正确的时间应用更新？

编辑：我已经使用以下代码解决了它：

然而，为什么原版def不起作用的问题仍然存在......

我有一个文件夹，其中包含多个文件，例如filetype1_ddMMyyyy_hhmm， filetype2_ddMMyyyy_hhmm

每天，可能有多个不同小时的文件，我只需要解析时间最长的文件。在非反应式流世界中，该算法可以实现为 groupBy 日期，它在 scalaz-stream 中的等价物是什么？

如何加快以下scalaz-stream代码的速度？目前处理 70MB 的文本大约需要 5 分钟，所以我可能做错了，因为普通的 scala 等价物需要几秒钟。

（跟进另一个问题）