Tee with conduit iternals

[snoyberg]

Tee with conduit iternals

[nh2]

I managed to make a small prototype:

module Main where

import Control.Monad.IO.Class
import qualified Data.Conduit as C
import Data.Conduit.Internal
import qualified Data.Conduit.List as CL
import Data.Void

main :: IO ()
main = do
    let printTee x    = liftIO $ putStrLn $ "tee " ++ show x
    let printNormal x = liftIO $ putStrLn $ "normal " ++ show x
    let ping          = liftIO $ putStrLn "ping"

    putStrLn "\n     -- Test 1 --"
    CL.sourceList [1..3::Int] C.=$= tee (CL.mapM_ printTee)
                              C.$$      (CL.mapM_ printNormal)

    putStrLn "\n     -- Test 2 --"
    CL.sourceList [1..3::Int] C.=$= tee (C.await >>= printTee)
                              C.$$      (CL.mapM_ printNormal)

    putStrLn "\n     -- Test 3 --"
    CL.sourceList [1..3::Int] C.=$= tee ((C.await >>= printTee) >> ping)
                              C.$$      (CL.mapM_ printNormal)

    putStrLn "\n     -- Test 4 --"
    CL.sourceList [1..3::Int] C.=$= tee ((C.await >>= printTee) >> ping >> (C.await >>= printTee))
                              C.$$      (CL.mapM_ printNormal)



-- | Forks a stream, like the Unix `tee` utility.
-- Like GNU tee, it yields each value first to the outer pipe, then to the inner one.
tee :: Monad m => Sink a m () -> ConduitM a a m ()
tee (ConduitM sStart) = ConduitM $ go sStart
  where
    go s = case s of
      HaveOutput _nextPipeCannotHappen mFinalizer void -> absurd void -- TODO where run finalizer?
      NeedInput f _fEnd -> NeedInput (\i -> yield i >> go (f i)) (\() -> Done ())
      Done () -> idP -- The tee pipe stopped accepting; we don't care.
                     -- The main pipe then just passes values through.
      PipeM mPipe -> PipeM $ do pip <- mPipe
                                return $ go pip
      Leftover pip x -> yield x >> go pip -- TODO Is it right that we insert the x here?

It has a simpler type signature than the tee in Data.Conduit.Extra.Pipes, and doesn't have the lift problems I mentioned (it composes).

Can you see any problem with this?

[snoyberg]

I think the leftover handling is incorrect; that clause should probably be:

Leftover pip x -> Leftover (go pip) x

though I haven't thought about it too much. I'd also probably prefer running the Sink to completion just in case it has any finalization it needs to perform. Other than that, seems good.

[jwiegley]

Looks like a much better tee than what I wrote.

[snoyberg]

@jwiegley Do you want to pull this code into conduit-extra then?

[jwiegley]

@snoyberg Will do.

[nh2]

@snoyberg Where did it end up?

[jwiegley]

I did not get around to moving this into conduit-extra, so if that's still needed let me know. I can move that in today actually if it hasn't been already.

[nh2]

@jwiegley I don't need it right now, but I've found it very useful in the past, and would appreciate it to be somewhere upstream rather than in my own code.

Also what @snoyberg said above would have to be addressed.

[jwiegley]

OK, one issue I see: If NeedInput is resolved and passes a value down to the sink and downstream, and the sink then uses leftover, downstream will see the leftover value again.

@snoyberg @nh2 Here is my version based on the comments above:

-- | Forks a stream, like the Unix `tee` utility.
--
-- This like GNU tee, but yields each value first to the outer pipe, then to
-- the inner one.
tee :: Monad m => Sink a m () -> ConduitM a a m ()
tee (ConduitM sStart) = ConduitM $ go sStart
  where
    go s = case s of
      -- The sink passed to tee cannot generate values.
      HaveOutput _ _ o -> absurd o

      -- Whenever a value is needed by the sink, we ask upstream and then
      -- yield the value to both the sink and downstream.  Note that if
      -- upstream terminates early, we pass the result value to the sink.
      NeedInput f lc   -> NeedInput (\i -> yield i >> go (f i)) (go . lc)

      -- Once the sink is Done, switch to being a transparent pass-through.
      Done ()          -> idP

      -- If an action must be executed to determined the next @Pipe@, do so
      -- and then process it with this function.
      PipeM mPipe      -> PipeM $ liftM go mPipe

      -- A leftover from the sink will be passed again to both the sink and
      -- downstream, meaning downstream will see the same result twice even
      -- though it never used 'leftover' itself.
      Leftover pip x   -> Leftover (go pip) x

[snoyberg]

I don't like the behavior of Leftover. I think it would make more sense to add a call to injectLeftovers around sStart.

[jwiegley]

@snoyberg Yes, that sounds nicer. Here is the result:

-- | Forks a stream, like the Unix `tee` utility.
--
-- This is like GNU tee, but yields each value first to the outer pipe, then
-- to the inner one.
tee :: Monad m => Sink a m () -> ConduitM a a m ()
tee (ConduitM sStart) = ConduitM $ go (injectLeftovers sStart)
  where
    go s = case s of
      -- The sink passed to tee cannot generate values.
      HaveOutput _ _ o -> absurd o

      -- Whenever a value is needed by the sink, we ask upstream and then
      -- yield the value to both the sink and downstream.  Note that if
      -- upstream terminates early, we pass the result value to the sink.
      NeedInput f lc   -> NeedInput (\i -> yield i >> go (f i)) (go . lc)

      -- Once the sink is Done, switch to being a transparent pass-through.
      Done ()          -> idP

      -- If an action must be executed to determined the next @Pipe@, do so
      -- and then process it with this function.
      PipeM mPipe      -> PipeM $ liftM go mPipe

      -- There can be no leftovers from the sink, since we called
      -- 'injectLeftovers' above.
      Leftover _ x    -> absurd x

[snoyberg]

Actually, this now looks like just a special case of passthroughSink. The latter function provides the ability for the Sink to return arbitrary return values and have them consumed by some finalizer function.

[jwiegley]

Ok, then I guess we leave this as closed and defer to passthroughSink. I was unaware of that function.