haskell - 向下游发信号表明上游已用尽

Question

问题

使用 Haskell pipes库，我正在尝试定义 Pipe 具有以下类型:

signalExhausted :: Monad m => Pipe a (Value a) m r 

Value数据类型定义为:

data Value a = Value a | Exhausted

管道应遵守以下规律:

toList (each [] >-> signalExhausted) ==                 [Exhausted]
toList (each xs >-> signalExhausted) == map Value xs ++ [Exhausted]

换句话说，管道应该等价于 Pipes.Prelude.map Value。 ，除了它应该产生一个额外的 Exhausted在处理完所有上游值之后，让下游有机会执行一些最终操作。

可以这样 Pipe被定义？

例子

> let xs = words "hubble bubble toil and trouble"
> toList $ each xs >-> signalExhausted
[Value "hubble", Value "bubble", Value "toil", Value "and", Value "trouble", Exhausted]

笔记

我知道 pipes-parse 库提供函数 draw和 parseForever .这些看起来很有用，但我不太明白如何将它们组合成 Pipe符合上述规范。

Answer 1

像 signalExhausted 这样的管道无法定义，但函数等效于 (>-> signalExhausted)能够。

>-> 是 pull category 的专用版本.执行由下游代理从上游代理中提取数据驱动。下游代理发送空请求()上游并阻塞，直到持有值的响应从上游代理返回。当上游代理用尽并且没有任何值要发回时，它return s。你可以看到return这对于 each 的定义中的这些示例很重要.

each = F.foldr (\a p -> yield a >> p) (return ())
-- what to do when the data's exhausted ^                       

下游代理需要一个值才能继续运行，但管道库无法提供任何值，因此下游代理永远不会再次运行。由于它永远不会再次运行，因此它无法修改或对数据使用react。

这个问题有两种解决方案。最简单的就是 map Value在上游管道上方并添加 yield Exhausted完成后。

import Pipes
import qualified Pipes.Prelude as P

data Value a = Value a | Exhausted
    deriving (Show)

signalExhausted p = p >-> P.map Value >> yield Exhausted

除了函数 signalExhausted 之外，这正是您正在寻找的。取代 (>-> signalExhausted) .

let xs = words "hubble bubble toil and trouble"
print . P.toList . signalExhausted $ each xs

[Value "hubble",Value "bubble",Value "toil",Value "and",Value "trouble",Exhausted]

这个问题的更通用的解决方案是阻止上游代理返回，而是在它耗尽时向下游发出信号。我在 answer to a related question 中演示了如何做到这一点。 .

import Control.Monad
import Pipes.Core

returnDownstream :: Monad m => Proxy a' a b' b m r -> Proxy a' a b' (Either r b) m r'
returnDownstream = (forever . respond . Left =<<) . (respond . Right <\\)

这将替换每个 respond与 respond . Right并替换 return与 forever . respond . left ，将返回与响应一起发送到下游。
returnDownstream比你要找的更一般。我们可以演示如何使用它来重新创建 signalExhausted . returnDownstream将返回的管道转换为永不返回的管道，并将其返回值作为 Left 转发到下游 Either 的值.

signalExhausted p = returnDownstream p >-> respondLeftOnce

respondLeftOnce是一个示例下游代理。下游代理可以辨别 Right 中保存的常规值。以及保存在 Left 中的返回值.

respondLeftOnce :: Monad m => Pipe (Either e a) (Value a) m ()
respondLeftOnce = go
    where
        go = do
            ea <- await
            case ea of
                Right a -> yield (Value a) >> go                    
                Left  _ -> yield Exhausted       -- The upstream proxy is exhausted; do something else