Streaming JSON array downsampler written in (probably bad) Haskell

gistfile1.hs

import Data.JsonStream.Parser
import qualified Data.ByteString.Char8 as B
import qualified Data.ByteString.Lazy.Char8 as BL
import qualified Data.Aeson.Types as AT
import qualified Data.Aeson.Encode as AE
import Pipes
import qualified Pipes.Prelude as P

import System.IO (isEOF, stdin)
import System.Random
import Control.Monad (unless, when)

jsonArrayGrouper :: Pipe B.ByteString B.ByteString IO ()
jsonArrayGrouper = do_parse parse_output
  where parse_output :: (ParseOutput AT.Value)
        parse_output = runParser (arrayOf value)
        do_parse :: ParseOutput AT.Value -> Pipe B.ByteString B.ByteString IO ()
        do_parse output = case output of
            ParseYield value new_output -> do
              -- lift $ putStrLn "emitting new json value"
              yield $ BL.toStrict $ AE.encode value
              do_parse new_output
            ParseNeedData cont -> do
              -- lift $ putStrLn "in need of new data"
              input <- await
              -- lift $ putStrLn $ "injecting " ++ (B.unpack input) ++ " into the parser"
              do_parse $ cont input
            ParseDone remaining -> do
              lift $ putStrLn "parsing done"
              return ()
            ParseFailed err -> lift $ putStrLn $ "PARSING ERROR: " ++ err

stdinProducer :: Producer B.ByteString IO ()
stdinProducer = do
  eof <- lift isEOF
  unless eof $ do
    str <- lift (B.hGet stdin 100)
    yield str
    stdinProducer

randomSampler :: Pipe a a IO ()
randomSampler = do
  val <- lift (randomIO :: IO Float)
  tmp <- await
  when (val < 0.01) $ yield tmp
  randomSampler

bsWriter :: Consumer B.ByteString IO ()
bsWriter = do
  val <- await
  lift $ B.putStrLn val
  bsWriter

main =do
  runEffect $ stdinProducer >-> jsonArrayGrouper >-> randomSampler >-> bsWriter