问题标签 [scalaz-stream]

我无法运行一个简单的 scalaz-stream 示例，从 TCP 读取并写入标准输出。

它只是坐在那里，不打印任何东西。

我也尝试过各种安排使用to, 总是用tcp.lift咒语来获得 a Process[Connection, A]，包括

这甚至不编译。

我需要wye将源流和打印流放在一起吗？我在原始拉取请求中找到的一个tcp示例nio似乎表明了这一点，但似乎wye不再存在于 上Process，因此不幸的是，混乱盛行。

编辑事实证明，除了 Paul 解释的类型问题之外，您还需要“手动”运行内部进程，例如通过执行p.map(_.run.run).run.run. 我不认为这是这样做的惯用方式，但它确实有效。

使用 scalaz-stream 是否可以拆分/分叉然后重新加入流？

例如，假设我有以下功能

使用 scalaz-stream，在此示例中，结果将如您所料 - 一个从 1 到 10 的数字元组传递到接收器。

但是，如果我们替换streamOfNumbers为需要 IO 的东西，它实际上会执行两次 IO 操作。

使用 aTopic我可以创建一个 pub/sub 进程，它可以正确地复制流中的元素，但它不会缓冲 - 它只是尽可能快地使用整个源，而不管接收器消耗它的速度如何。

我可以将它包装在一个有界队列中，但是最终结果感觉比它需要的要复杂得多。

有没有一种更简单的方法可以在 scalaz-stream 中拆分流而无需从源中重复 IO 操作？

让我们从一些 scalaz-stream 文档中举一个例子，但有一个理论上的扭曲。

在这种情况下，文件很可能包含一些非双精度字符串，并且fahrenheitToCelsius会抛出一些NumberFormatException. 假设在这种情况下，我们可能希望记录此错误并忽略它以进行进一步的流处理。惯用的做法是什么？我见过一些例子，但它们通常collectFrom是流。

I have copied an example from http4s:

This yields a compilation error: "Expression of type Unit doesn't conform to expected type _A"

I just want to printline "Terminated!" when the sink gets terminated

I want to implement an client app that first send an request to server then wait for its reply(similar to http)

My client process may be

Here is the api

Then, how to implement the remain part of ask ?

我编写了一个 echo 应用程序send并receive '\0'终止了字符串

https://gist.github.com/jilen/10a664cd588af10b7d09

我在不同的终端上Foo.runServer()运行Foo.runClient()

并从客户端控制台输入数字1 2 3 ...，但客户端没有收到回复。

我的回声应用程序有什么问题？

我创建了一个流程来帮助我解析日志文件。日志文件需要使用字符串标签进行标记。从来没有记录可以提供这个字符串标签，所以我需要在每个日志事件中维护一些状态，以便每个事件都有一个标签。如果事件缺少标签，我想使用找到的最后一个标签。这使我能够解析如下所示的日志并具有与消息行关联的“标签”，即使它们是 eventindex04 记录标头的一部分。我可以将事件包装在一个信封中并在信封中包含逻辑，但是我到处都有一个看起来很重的信封。

我把这个工作正常的过程放在一起：

这感觉应该表示为扫描/折叠，但我无法使其工作。这可以重写为扫描/折叠吗？

我想通过 db.stream(yourquery) 通过 scalaz-stream 流式传输从 3.0.0 查询返回的数据。

看起来 reactive-streams.org 使用了不同库实现的 API 和数据流模型。

在从 scalaz-stream 流程回流到 slick 发布者的背压的情况下，您如何做到这一点？

上下文：我正在尝试编写一个Process1[ByteVector, spray.http.HttpResponsePart]with output ChunkedResponseStart(bytes), MessageChunk(bytes), MessageChunk(bytes), ..., ChunkedResponseEnd。我还没有完全理解 scalaz-stream 及其词汇。

如何编写一个可以以n不同方式处理第一个块的进程？

我想出了这个（以字符串为例）：

什么是惯用的并且可能是通常可组合的写作方式headerChunkAndRest？

有没有一种惯用的方式来分块和连接？

我发现的方式（字节示例）：

1.

但在这种情况下，并不真正需要中间体Vector(from )。chunk

2. 基于 process1.chunk 的复制/粘贴：

Process有没有办法通过组合现有的'es来做同样的事情？

一个附带问题：可以repeat用来代替++ go? 是不是和上一个一样：