sinkfork 0.3

sinkfork.hs

import Data.Conduit
import qualified Data.Conduit.List as CL
import Control.Applicative ((<$>), (<*>), Applicative, pure, (<|>))
import Data.Functor.Identity (runIdentity)

sinkFork :: (Applicative m, Monad m) => Sink a m b -> Sink a m c -> Sink a m (b, c)
sinkFork (SinkM mlsink) rsink = SinkM $ do
    lsink <- mlsink
    return $ sinkFork lsink rsink
sinkFork lsink (SinkM mrsink) = SinkM $ do
    rsink <- mrsink
    return $ sinkFork lsink rsink
sinkFork (Processing lpush lclose) (Processing rpush rclose) = Processing
    (\i -> sinkFork (lpush i) (rpush i))
    ((,) <$> lclose <*> rclose)
sinkFork (Processing lpush lclose) (Done _ rres) = Processing
    (\i -> sinkFork (lpush i) (Done Nothing rres))
    ((,) <$> lclose <*> pure rres)
sinkFork (Done _ lres) (Processing rpush rclose) = Processing
    (\i -> sinkFork (Done Nothing lres) (rpush i))
    ((,) lres <$> rclose)
sinkFork (Done lleft lres) (Done rleft rres) = Done
    (lleft <|> rleft)
    (lres, rres)

main :: IO ()
main = print $ runIdentity $ CL.sourceList [1..10]
    $$ sinkFork
            (CL.fold (+) (0 :: Int))
            (CL.take 5)