Haskell的Pipe/Conduit是什麼?

02-01

謝邀，Pipe的 primitive version：

data Pipe a b r = Pure r | Await (a -&> Pipe a b r) | Yield b (Pipe a b r)

(&<-&<) :: Pipe b c r -&> Pipe a b r -&> Pipe a c r Pure r &<-&< _ = Pure r Yield b p1 &<-&< p2 = Yield b (p1 &<-&< p2) Await f &<-&< Yield b p2 = f b &<-&< p2 p1 &<-&< Await f = Await $ a -&> p1 &<-&< f a _ &<-&< Pure r = Pure r cat :: Pipe a a r cat = Await $ a -&> Yield a cat

Pure為終止計算；

Yield為emit 計算結果b，以及下一步計算(Pipe a b r)，即：Send output data；

Await等待輸入a，返回一個計算(Pipe a b r)，即：Receive input data。

又因為：

cat &<-&< p = p -- Right identity p &<-&< cat = p -- Left identity (p1 &<-&< p2) &<-&< p3 = p1 &<-&< (p2 &<-&< p3) -- Associativity

故稱之為Pipe category

用stages composition去取代functions composition；可以將計算細化為不同的stages，函數組合變成了狀態/階段的組合，more flexible

Pipes.Tutorial

port to Scala: arjanblokzijl/scala-pipes

樓上基本說的都很清楚啦，我就說說Conduit和Pipe的區別吧。可以先去看看MonadResource/ResourceT (mtl style transformer)，一個嵌入了一張可變釋放表的Reader。

Conduit內置了和MonadResource的互動，也就是addCleanup這個函數，它可以在某個中間的component跑完的時候，或者下游發生異常提前終止stream的時候，去執行cleanup動作。然後基於這個函數就可以構造一些方便資源管理的函數了（bracketP之類的）。
Conduit有個專門處理LeftOver的構造函數，用來把這次沒處理完的值返回出來（當然大部分情況下l和i是一樣的），這可以方便的實現io-streams的unRead，但是Pipe要做這個就得引入State了，把leftover記錄在s里，下一個component再去消費。

所以有人說Conduit臃腫，不過我覺得還好啦。

簡單來說，Conduit就是流計算的一種模型。每一步都有從上游接收和向下游發送兩個類型，以及計算結果類型（注意區別於向下游發送的數據）。每一步按單位進行計算，比如LazyByteString的Chunk，數組的元素等。目的就是避免每一步計算都一次性針對全部數據。