如何編寫一個管道向下遊發送從上游接收到的內容列表？

Question

我有一個艱難的時間來寫這個簽名pipe：

toOneBigList :: (Monad m, Proxy p) =>() -> Pipe p a [a] m r 

應該簡單地採取從上游所有a S和列表中的下游發送。

我所有的嘗試看起來都是根本打破的。

任何人都可以指向正確的方向嗎？

Answer 1

有兩個pipes爲基礎的解決方案，我就讓你選擇你更喜歡哪一個。

注意：目前還不清楚爲什麼您在下游接口上輸出列表，而不是直接將其作爲結果返回。

管道式

第一個，這是非常接近conduit基礎的解決方案採用了即將到來的pipes-pase，這基本上是完整的，只是需要的文檔。你可以在Github上找到latest draft。

使用pipes-parse，溶液是相同的conduit解決方案，彼得得到：

import Control.Proxy 
import Control.Proxy.Parse 

combine 
    :: (Monad m, Proxy p) 
    =>() -> Pipe (StateP [Maybe a] p) (Maybe a) [a] m() 
combine() = loop [] 
    where 
    loop as = do 
     ma <- draw 
     case ma of 
      Nothing -> respond (reverse as) 
      Just a -> loop (a:as) 

draw就像conduit的await：它從任一剩菜緩衝器（這是StateP一部分）或請求一個值如果緩衝區爲空，則從上游啓動。 Nothing表示文件結束。

你可以用不具有使用wrap函數從pipes-parse文件信號，其具有類型的端部的管：

wrap :: (Monad m, Proxy p) => p a' a b' b m r -> p a' a b' (Maybe b) m s 

經典管道風格

第二種替代方案是簡單一點。如果你想摺疊一個給定的管道，你可以這樣做，直接使用WriterP：

import Control.Proxy 
import Control.Proxy.Trans.Writer 

foldIt 
    :: (Monad m, Proxy p) => 
    (() -> Pipe p a b m()) ->() -> Pipe p a [b] m() 
foldIt p() = runIdentityP $ do 
    r <- execWriterK (liftP . p >-> toListD >-> unitU)() 
    respond r 

這是怎麼回事的更高級別的描述，但它需要在管道作爲傳遞一個明確的說法。這取決於你喜歡哪一個。

順便說一下，這就是爲什麼我問你爲什麼要發送一個下游價值。以上是簡單得多，如果你回到摺疊列表：

foldIt p = execWriterK (liftP . p >-> toListD) 

的liftP甚至可能不是必要的，如果p在其代理類型完全多態性。我只是把它作爲一個預防措施。

獎金方案

原因pipes-parse不提供是，它總是一個管反模式組的結果到一個列表。 pipes有幾個很好的功能，可以不必將輸入分組到列表中，即使您試圖產生多個列表。例如，使用respond組成，你可以有一個代理產生它會遍歷流的子集，然後注射使用該子集的處理程序：

example :: (Monad m, Proxy p) =>() -> Pipe p a (() -> Pipe p a a m()) m r 
example() = runIdentityP $ forever $ do 
    respond $ \() -> runIdentityP $ replicateM_ 3 $ request() >>= respond 

printIt :: (Proxy p, Show a) =>() -> Pipe p a a IO r 
printIt() = runIdentityP $ do 
    lift $ putStrLn "Here we go!" 
    printD() 

useIt :: (Proxy p, Show a) =>() -> Pipe p a a IO r 
useIt = example />/ (\p -> (p >-> printIt)()) 

這裏有一個如何使用它的一個例子：

>>> runProxy $ enumFromToS 1 10 >-> useIt 
Here we go! 
1 
2 
3 
Here we go! 
4 
5 
6 
Here we go! 
7 
8 
9 
Here we go! 
10 

這意味着即使需要對元素進行分組，也不需要將單個元素放入內存中。

Answer 2

我只給出部分答案，也許別人會有更好的答案。

據我所知，標準管道沒有檢測管道另一部分何時終止的機制。終止的第一個管道產生管道的最終結果，所有其他管道都被丟棄。因此，如果你有一個永久消耗輸入的管道（最終產生一個列表），那麼當它的上游完成時，它將沒有機會演出併產生輸出。 （這是故意的，因此上下游部分都是相互對立的。）也許這是在管道頂部的一些圖書館中解決的。

情況與conduit不同。它具有consume函數，該函數將所有輸入組合到列表中並返回（不輸出）它。就像寫一個你所需要的，是在最後輸出列表的功能，並不難：

import Data.Conduit 

combine :: (Monad m) => Conduit a m [a] 
combine = loop [] 
    where 
    loop xs = await >>= maybe (yield $ reverse xs) (loop . (: xs))