@@ -1,3 +1,6 @@
1+{-# LANGUAGE DeriveDataTypeable #-}
2+{-# LANGUAGE ScopedTypeVariables #-}
3+
14{-
25
36This file is part of the Haskell package cassava-streams. It is
@@ -12,7 +15,8 @@ file.
1215
1316--------------------------------------------------------------------------------
1417module System.IO.Streams.Csv.Decode
15− ( decodeStream
18+ ( StreamDecodingError (..)
19+ , decodeStream
1620 , decodeStreamWith
1721 , decodeStreamByName
1822 , decodeStreamByNameWith
@@ -20,14 +24,24 @@ module System.IO.Streams.Csv.Decode
2024 ) where
2125
2226--------------------------------------------------------------------------------
27+import Control.Exception
2328import Data.ByteString (ByteString)
2429import qualified Data.ByteString as BS
2530import Data.Csv hiding (Parser, decodeWith, decodeByNameWith)
2631import Data.Csv.Incremental
2732import Data.IORef
33+import Data.Typeable
2834import System.IO.Streams (InputStream, makeInputStream)
2935import qualified System.IO.Streams as Streams
3036
37+--------------------------------------------------------------------------------
38+-- | Exception thrown when stream decoding cannot continue due to an
39+-- error.
40+newtype StreamDecodingError = StreamDecodingError String
41+ deriving (Typeable, Show)
42+
43+instance Exception StreamDecodingError
44+
3145--------------------------------------------------------------------------------
3246-- | Create an @InputStream@ which decodes CSV records from the given
3347-- upstream data source.
@@ -103,12 +117,12 @@ onlyValidRecords input = makeInputStream $ do
103117
104118 case upstream of
105119 Nothing -> return Nothing
106− Just (Left err) -> bomb err -- FIXME: replace with throwIO
120+ Just (Left err) -> bomb ("invalid CSV row: " ++ err)
107121 Just (Right x) -> return (Just x)
108122
109123--------------------------------------------------------------------------------
110124-- | Internal function which feeds data to the CSV parser.
111−dispatch :: IORef [Either String a]
125+dispatch :: forall a. IORef [Either String a]
112126 -- ^ List of queued CSV records.
113127 -> IORef (Maybe (Parser a))
114128 -- ^ Current CSV parser state.
@@ -124,9 +138,10 @@ dispatch queueRef parserRef input = do
124138 parser <- readIORef parserRef
125139 case parser of
126140 Nothing -> return Nothing
127− Just (Fail _ e) -> bomb e
128− Just (Many xs f) -> more f >> feed xs
129− Just (Done xs ) -> writeIORef parserRef Nothing >> feed xs
141+ Just (Fail _ e) -> bomb ("input data malformed: " ++ e)
142+ Just (Many [] f) -> more f
143+ Just (Many xs f) -> writeIORef parserRef (Just $ Many [] f) >> feed xs
144+ Just (Done xs ) -> writeIORef parserRef Nothing >> feed xs
130145
131146 (x:xs) -> do
132147 writeIORef queueRef xs
@@ -135,14 +150,17 @@ dispatch queueRef parserRef input = do
135150 where
136151 -- Send more data to the CSV parser. If there is no more data
137152 -- from upstream then send an empty @ByteString@.
138− more f = Streams.read input >>=
139− writeIORef parserRef . Just . maybe (f BS.empty) f
153+ more :: (ByteString -> Parser a) -> IO (Maybe (Either String a))
154+ more f = do bs <- Streams.read input
155+ writeIORef parserRef (Just $ maybe (f BS.empty) f bs)
156+ dispatch queueRef parserRef input
140157
141158 -- Feed records downstream.
142− feed xs = writeIORef queueRef xs >>
143− dispatch queueRef parserRef input
159+ feed :: [Either String a] -> IO (Maybe (Either String a))
160+ feed xs = do writeIORef queueRef xs
161+ dispatch queueRef parserRef input
144162
145163--------------------------------------------------------------------------------
146−-- | FIXME: use a proper exception.
164+-- | Throw an exception.
147165bomb :: String -> IO a
148−bomb = fail
166+bomb = throwIO . StreamDecodingError