Conduit Broadcast

Question

Una vista giorni fa, ho chiesto questo question. Ora ho bisogno di una pura versione a thread singolo di questa funzione:

Per ripetere, ho bisogno di una funzione che manda ogni valore ricevuto a ciascun sink e raccoglie i loro risultati. La firma tipo di funzione dovrebbe essere qualcosa di simile:

broadcast :: [Sink a m b] -> Sink a m [b] 

migliore Sven

---

P.S. Non è sequence, ho provato che:

> C.sourceList [1..100] $$ sequence [C.fold (+) 0, C.fold (+) 0] 
[5050, 0] 

risultato atteso:

[5050, 5050] 

---

P.P.S. zipSinks dà il risultato desiderato, ma funziona solo con le tuple:

> C.sourceList [1..100] $$ C.zipSinks (C.fold (+) 0) (C.fold (+) 0) 
(5050, 5050) 

Answer 1

In pratica tutto quello che dobbiamo fare è quello di reimplementare sequence, ma con zipSinks invece dell'operazione sequenza originale:

import Data.Conduit as C 
import Data.Conduit.List as C 
import Data.Conduit.Util as C 

fromPairs 
    :: (Functor f) 
    => f [a]      --^an empty list to start with 
    -> (f a -> f [a] -> f (a, [a])) --^a combining function 
    -> [f a]      --^input list 
    -> f [a]      --^combined list 
fromPairs empty comb = g 
    where 
    g []  = empty 
    g (x:xs) = uncurry (:) `fmap` (x `comb` g xs) 

Ora creazione broadcast è solo applicando fromPairs-zipSinks:

broadcast :: (Monad m) => [Sink a m b] -> Sink a m [b] 
broadcast = fromPairs (return []) zipSinks 

E possiamo fare qualcosa come

main = C.sourceList [1..100] $$ broadcast [C.fold (+) 0, C.fold (*) 1] 

---

Aggiornamento: Possiamo vedere che fromPairs sembra proprio sequenceA e così siamo in grado di spingere ancora di più l'idea. Definiamo un funtore applicativa zippare su tubi simili a ZipList:

import Control.Applicative 
import Control.Monad 
import Data.Conduit 
import Data.Conduit.Util 
import Data.Traversable (Traversable(..), sequenceA) 

newtype ZipSink i m r = ZipSink { getZipSink :: Sink i m r } 

instance Monad m => Functor (ZipSink i m) where 
    fmap f (ZipSink x) = ZipSink (liftM f x) 
instance Monad m => Applicative (ZipSink i m) where 
    pure = ZipSink . return 
    (ZipSink f) <*> (ZipSink x) = 
     ZipSink $ liftM (uncurry ($)) $ zipSinks f x 

Poi broadcast diventa semplice come

broadcast :: (Traversable f, Monad m) => f (Sink i m r) -> Sink i m (f r) 
broadcast = getZipSink . sequenceA . fmap ZipSink