so response

sotime.hs

{-# LANGUAGE OverloadedStrings #-}

import Data.Monoid
import Data.AdditiveGroup
import qualified Data.ByteString as B
import qualified Data.ByteString.Builder as BB
import qualified Data.ByteString.Lazy as BL
--import Data.Time.Clock
import Data.Thyme.Clock
import Data.Thyme.Clock.POSIX
import Control.Applicative
import Control.Monad.State.Strict
import Pipes
import Pipes.Lift
import qualified Pipes.ByteString as PB
-- import qualified Pipes.Prelude as P
-- import qualified Pipes.Group as G
import System.IO

batched :: Pipe B.ByteString BL.ByteString IO () 
batched = do
    startTime <- liftIO getPOSIXTime 
    evalStateP (startTime,mempty) $ forever $ do
        incoming <- BB.byteString <$> await
        currentTime <- liftIO getPOSIXTime
        -- (POSIXTime,BB.Builder) inner state
        (baseTime,accum) <- get
        if (currentTime ^-^ baseTime > timeLimit) 
            then do
                yield $ BB.toLazyByteString accum
                put $ (currentTime, incoming)
            else do
                put $ (baseTime, accum <> incoming)
    where
        timeLimit = fromSeconds 0.1

main :: IO ()
main = do
    runEffect $ (yield "boo" >> yield "baa") >-> PB.stdout
    putStrLn $ ""