Haskell Conduit侧通道传递组合子：是否存在或可构建该组合子？

Great question! There isn't a built-in combinator in the core conduit package that directly does exactly what you're asking for—transforming a ConduitT i o m r into a ConduitT i ([i], o) m r while passing inputs through to the original conduit. But the good news is you can build this easily using existing core combinators, specifically zipAccum, which handles state tracking alongside your conduit's normal processing.

Building the Combinator #

The key here is tracking the input(s) that lead to each output o. The zipAccum combinator lets you maintain a state value as inputs flow through, and pair that state with each output from the inner conduit. For your use case, we'll use a list of inputs as our state, updating it every time a new input comes in.

Here's the implementation:

import Data.Conduit (ConduitT, zipAccum)

-- Tracks all inputs received so far, pairing them with each output from the inner conduit
traceInputs :: Monad m => ConduitT i o m r -> ConduitT i ([i], o) m r
traceInputs inner = fmap snd $ zipAccum updateState [] inner
  where
    updateState acc i = pure (acc ++ [i])  -- Append new input to our state list

The zipAccum function handles passing inputs to the inner conduit, updating our state list with each new input, and pairing the current state with every output from the inner conduit. The fmap snd just discards the final accumulated state from the result, keeping only the original conduit's return value r.

Customizing for Your Testing Needs #

Depending on your exact testing requirement, you can tweak the state update logic:

• If you only want the single input that produced each output (e.g., for a 1:1 conduit like map), change the state update to replace the list instead of appending:

  updateState _ i = pure [i]
  

• If you're dealing with windowed or batch processing, adjust the state to only keep relevant inputs (e.g., the last 3 inputs processed).

Using It for Testing #

Once you have traceInputs, you can chain it with your testing logic to validate each ([i], o) pair. For example, let's test a conduit that sums inputs and emits the running total:

import Data.Conduit (yieldMany, runConduit, mapM_)
import Control.Monad.IO.Class (liftIO)

runningSum :: Num i => ConduitT i i m ()
runningSum = scanl (+) 0

testRunningSum :: IO ()
testRunningSum = runConduit $
  yieldMany [1, 2, 3, 4]
  .| traceInputs runningSum
  .| mapM_ validatePair
  where
    validatePair (inputs, total) = do
      let expected = sum inputs
      if total == expected
        then liftIO $ putStrLn $ "✅ Valid: " ++ show inputs ++ " → " ++ show total
        else liftIO $ putStrLn $ "❌ Invalid: " ++ show inputs ++ " → " ++ show total ++ " (expected " ++ show expected ++ ")"

Running this would output:

✅ Valid: [1] → 1
✅ Valid: [1,2] → 3
✅ Valid: [1,2,3] → 6
✅ Valid: [1,2,3,4] → 10

This setup lets you easily verify that each output from your original conduit aligns with the inputs that generated it.

内容的提问来源于stack exchange，提问作者Justin L.