join.hs

import Data.Conduit
import Control.Monad.Trans.Resource
import Control.Monad

joinSource :: Resource m => Source m (Source m a) -> Source m a
joinSource s = Source $ do
  ps <- genSource s
  innerSources <- newRef []
  let pull = do xs <- readRef innerSources
                case xs of
                  (x:xss) -> do SourceResult st vs <- sourcePull x
                                case st of
                                  StreamClosed -> writeRef innerSources xss
                                  StreamOpen -> return ()
                                return $ SourceResult StreamOpen vs
                  [] -> do SourceResult st xs' <- sourcePull ps
                           writeRef innerSources =<< mapM genSource xs'
                           return $ SourceResult st []
      close = do mapM_ sourceClose =<< readRef innerSources
                 sourceClose ps
  return $ PureSource {
      sourcePull = pull
    , sourceClose = close
    }