Conduit with aeson / attoparsec, how to exit without exception, without exception, when the source no longer has data

Question

I use aeson/ attoparsecand conduit/ conduit-httpconnected conduit-attoparsecto parse JSON data from a file / web server. My problem is that my pipeline always throws this exception ...

ParseError {errorContexts = ["demandInput"], errorMessage = "not enough bytes", errorPosition = 1:1}

... as soon as the socket closes or we press EOF. Analysis and transmission of resulting data structures through a pipeline, etc. It works very well, but it always ends up sinkParserthrowing this exception. I call him like this ...

j <- CA.sinkParser json

... inside my channel that parses ByteStrings in my message structures.

How can I make it just exit the pipeline as soon as there is no data (no more than top-level expressions)? Is there a decent way to detect / distinguish this exception without looking at the error lines?

Thanks!

EDIT: Example:

{-# LANGUAGE OverloadedStrings #-}

module Main where

import Control.Applicative
import qualified Data.ByteString as B
import qualified Data.ByteString.Char8 as B8
import qualified Data.Conduit.Attoparsec as CA
import Data.Aeson
import Data.Conduit
import Data.Conduit.Binary
import Control.Monad.IO.Class

data MyMessage = MyMessage String deriving (Show)

parseMessage :: (MonadIO m, MonadResource m) => Conduit B.ByteString m B.ByteString
parseMessage = do
    j <- CA.sinkParser json
    let msg = fromJSON j :: Result MyMessage
    yield $ case msg of
        Success r -> B8.pack $ show r
        Error   s -> error s
    parseMessage

main :: IO ()
main =
    runResourceT $ do
        sourceFile "./input.json" $$ parseMessage =$ sinkFile "./out.txt"

instance FromJSON MyMessage where
    parseJSON j =
        case j of
        (Object o) -> MyMessage <$> o .: "text"
        _          -> fail $ "Expected Object - " ++ show j

Input example (input.json):

{"text":"abc"}
{"text":"123"}

Outputs:

out: ParseError {errorContexts = ["demandInput"], errorMessage = "not enough bytes", errorPosition = 3:1}

and out.txt:

MyMessage "abc"MyMessage "123"

Answer 1

This is an ideal use case for conduitParserEither:

parseMessage :: (MonadIO m, MonadResource m) => Conduit B.ByteString m B.ByteString
parseMessage =
    CA.conduitParserEither json =$= awaitForever go
  where
    go (Left s) = error $ show s
    go (Right (_, msg)) = yield $ B8.pack $ show msg ++ "\n"

If you are in FP Haskell Center, you can clone my solution in the IDE .