Turning Conduits into Pipes

Question

Does this look right to you? What are its shortcomings? Strengths?

import Control.Monad
import Control.Monad.Trans.Class
import Control.Pipe.Common
import Data.Conduit

c2p :: (Resource m, Monad m) => Conduit a m b -> Pipe a b m ()
c2p = do
  PreparedConduit push close <- lift $ runResourceT $ prepareConduit c
  loop push close
  where loop push close = do
          input <- await
          stepResult <- lift $ runResourceT $ push input
          case stepResult of
            Producing output -> do
              mapM_ yield output
              loop push close
            Finished _ output -> do
              mapM_ yield output
              lift $ runResourceT $ close
              return ()

Answer 1

I can see three problems with this approach:

1) Calling runResourceT at each step releases all the scarse resources allocated by the original conduit. This is not what you want, since a conduit can depend on the availability of those resources until the point where it terminates.

2) You cannot call close after push returns Finished. This is explicitly stated in the documentation of PreparedConduit.

3) You need to call close on the conduit when the upstream pipe terminates. Unfortunately, I think this is impossible to do with vanilla pipes. That's why I am currently experimenting with what I call "guarded pipes". You can find them here, together with some other utilities ported over from conduits.

Using guarded pipes, you can write something like:

c2p :: Resource m => Conduit a m b -> Pipe a b (ResourceT m) ()
c2p c = do
  PreparedConduit push close <- lift $ prepareConduit c
  loop push close
  where
    loop push close = do
      input <- tryAwait
      case input of
        Nothing -> do
          output <- lift $ close
          mapM_ yield output
        Just input -> do
          stepResult <- lift $ push input
          case stepResult of
            Producing output -> do
              mapM_ yield output
              loop push close
            Finished _ output ->
              mapM_ yield output

You can try it, for example, with:

main = runResourceT . runPipe $
  fileProducer "sample.txt.gz" >+> c2p CZ.ungzip >+> fileConsumer "sample.txt"

where fileProducer and fileConsumer are also contained in the above pipes-extra package.