向下游发信号通知上游耗尽

Question

题

使用Haskell 管道库,我正在尝试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

A pipe like signalExhausted can't be defined, but a function equivalent to (>-> signalExhausted) can.

>-> is a specialized version of the pull category. Execution is driven by the downstream proxies pulling data from upstream proxies. The downstream proxy sends an empty request () upstream and blocks until a response holding a value comes back from the upstream proxy. When the upstream proxy is exhausted and doesn't have any more values to send back, it returns. You can see the return that matters for these examples in the definition of each.

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

The downstream proxy needs a value to continue running, but there's no value the pipes library can possibly provide it, so the downstream proxy never runs again. Since it never runs again, there's no way it can modify or react to the data.

There are two solutions to this problem. The simplest is to map Value over the upstream pipe and add a yield Exhausted after it's done.

import Pipes
import qualified Pipes.Prelude as P

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

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

This does exactly what you're looking for except the function signalExhausted takes the place of (>-> 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]

The more general solution to this problem is to stop the upstream proxy from returning and instead signal downstream when it is exhausted. I demonstrated how to do so in an 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 <\\)

This replaces each respond with respond . Right and replaces return with forever . respond . left, sending returns downstream along with responses.

returnDownstream is more general than what you are looking for. We can demonstrate how to use it to recreate signalExhausted. returnDownstream transforms a pipe that returns into one that never returns, and instead forwards its return value downstream as the Left value of an Either.

signalExhausted p = returnDownstream p >-> respondLeftOnce

respondLeftOnce is an example downstream proxy. The downstream proxy can discern between regular values held in Right and the return value held in 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