Conduit Streaming Framework

The Data.Conduit.Process module provides a high-level, type-safe interface for interacting with external processes via streaming. It addresses common issues in System.Process such as lazy I/O and race conditions when checking exit codes.

Key features:

• Provides Sources and Sinks for standard input, output, and error.
• Uses a StreamingProcessHandle to avoid deadlocks when checking exit codes.
• Designed to work seamlessly with the async library for concurrent stream handling.

Important Requirements:

• Use the multithreaded runtime (-threaded): The underlying waitForProcess function can block all threads on a single-threaded runtime.
• Process Configuration: For general parameters like working directory or environment variables, use standard System.Process facilities (e.g., the proc or shell functions).

import Data.Conduit.Process (proc, streamingProcess, waitForStreamingProcess)

ResourceT is a monad transformer that creates a region of code where you can safely allocate and release resources. Unlike the standard bracket pattern, which is designed for nested, statically known resource lifecycles, ResourceT allows for interleaved allocation and deallocation. This is particularly useful when the number or order of resources depends on runtime data (e.g., reading a stream and opening a new file for every chunk of data).

To use ResourceT, you typically use runResourceT to execute a block of code and allocate to register a resource with a cleanup action. You can also manually trigger the release of a resource using the releaseKey returned by allocate.

import Control.Monad.Trans.Resource
import Control.Monad.IO.Class

main :: IO ()
main = runResourceT $ do
    -- allocate returns a (ReleaseKey, resource)
    (releaseKey, resource) <- allocate
        (putStrLn "Allocating resource" >> return 42)
        (\i -> putStrLn $ "Freeing resource: " ++ show i)
    
    -- Use the resource
    liftIO $ print resource
    
    -- Manually release the resource early if needed
    release releaseKey

The most critical concept in Conduit is that everything is driven by downstream. A component in the pipeline only executes if a downstream component requests data via await.

• If the downstream component is return (), it never calls await, so the entire pipeline terminates immediately without running upstream components.
• If a component in the middle of the pipeline is return (), it will exit immediately without yielding anything, causing any downstream components to receive Nothing and also exit.
• To consume all values from a pipeline and discard them, use sinkNull instead of return ().

import Conduit

-- This will produce no output because return () exits immediately
main :: IO ()
main = runConduit $ yieldMany [1..10] .| iterMC print .| return ()

-- This will print all values because sinkNull drives the pipeline
main2 :: IO ()
main2 = runConduit $ yieldMany [1..10] .| iterMC print .| sinkNull

ZipConduit is a newtype wrapper that allows you to combine multiple transformers. It works by draining all yielded values from all internal ZipConduits until they are all awaiting, then grabbing the next value from upstream and feeding it to all of them simultaneously. This is particularly useful for splitting a single stream into multiple specialized sub-streams and then recombining the results into a single transformer.

To use it, wrap your individual conduits in ZipConduit and combine them using the *> operator, then use getZipConduit to extract the resulting transformer.

import Conduit

tagger :: Monad m => ConduitT Int (Either Int Int) m ()
tagger = mapC $ \i -> if even i then Left i else Right i

evens, odds :: Monad m => ConduitT Int String m ()
evens  = mapC $ \i -> "Even number: " ++ show i
odds   = mapC $ \i -> "Odd  number: " ++ show i

left :: Either l r -> Maybe l
left = either Just (const Nothing)

right :: Either l r -> Maybe r
right = either (const Nothing) Just

inside :: Monad m => ConduitT (Either Int Int) String m ()
inside = getZipConduit
    $ ZipConduit (concatMapC left  .| evens)
   *> ZipConduit (concatMapC right .| odds)

main :: IO ()
main = runConduit $ enumFromToC 1 10 .| tagger .| inside .| mapM_C putStrLn