@@ -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,45 +15,62 @@ file.
1215
1316--------------------------------------------------------------------------------
1417module System.IO.Streams.Csv.Decode
15− ( decodeStream
18+ ( StreamDecodingError (..)
19+ , decodeStream
1620 , decodeStreamWith
1721 , decodeStreamByName
1822 , decodeStreamByNameWith
23+ , onlyValidRecords
1924 ) where
2025
2126--------------------------------------------------------------------------------
27+import Control.Exception
2228import Data.ByteString (ByteString)
2329import qualified Data.ByteString as BS
2430import Data.Csv hiding (Parser, decodeWith, decodeByNameWith)
2531import Data.Csv.Incremental
26−import Data.Either (lefts, rights)
2732import Data.IORef
28−import Data.List (intercalate)
33+import Data.Typeable
2934import System.IO.Streams (InputStream, makeInputStream)
3035import qualified System.IO.Streams as Streams
3136
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+
3245--------------------------------------------------------------------------------
3346-- | Create an @InputStream@ which decodes CSV records from the given
3447-- upstream data source.
3548--
3649-- Equivalent to @decodeStreamWith defaultDecodeOptions@.
3750decodeStream :: (FromRecord a)
38− => HasHeader -- ^ Whether to skip a header or not.
39− -> InputStream ByteString -- ^ Upstream.
40− -> IO (InputStream a) -- ^ An @InputStream@ which produces records.
51+ => HasHeader
52+ -- ^ Whether to skip a header or not.
53+ -> InputStream ByteString
54+ -- ^ Upstream.
55+ -> IO (InputStream (Either String a))
56+ -- ^ An @InputStream@ which produces records.
4157decodeStream = decodeStreamWith defaultDecodeOptions
4258
4359--------------------------------------------------------------------------------
4460-- | Create an @InputStream@ which decodes CSV records from the given
4561-- upstream data source.
4662decodeStreamWith :: (FromRecord a)
47− => DecodeOptions -- ^ CSV decoding options.
48− -> HasHeader -- ^ Whether to skip a header or not.
49− -> InputStream ByteString -- ^ Upstream.
50− -> IO (InputStream a) -- ^ An @InputStream@ which produces records.
63+ => DecodeOptions
64+ -- ^ CSV decoding options.
65+ -> HasHeader
66+ -- ^ Whether to skip a header or not.
67+ -> InputStream ByteString
68+ -- ^ Upstream.
69+ -> IO (InputStream (Either String a))
70+ -- ^ An @InputStream@ which produces records.
5171decodeStreamWith ops hdr input = do
5272 queue <- newIORef []
53− parser <- newIORef (decodeWith ops hdr)
73+ parser <- newIORef $ Just (decodeWith ops hdr)
5474 makeInputStream (dispatch queue parser input)
5575
5676--------------------------------------------------------------------------------
@@ -59,32 +79,57 @@ decodeStreamWith ops hdr input = do
5979--
6080-- Equivalent to @decodeStreamByNameWith defaultDecodeOptions@.
6181decodeStreamByName :: (FromNamedRecord a)
62− => InputStream ByteString -- ^ Upstream.
63− -> IO (InputStream a) -- ^ An @InputStream@ which produces records.
82+ => InputStream ByteString
83+ -- ^ Upstream.
84+ -> IO (InputStream (Either String a))
85+ -- ^ An @InputStream@ which produces records.
6486decodeStreamByName = decodeStreamByNameWith defaultDecodeOptions
6587
6688--------------------------------------------------------------------------------
6789-- | Create an @InputStream@ which decodes CSV records from the given
6890-- upstream data source. Data should be preceded by a header.
6991decodeStreamByNameWith :: (FromNamedRecord a)
70− => DecodeOptions -- ^ CSV decoding options.
71− -> InputStream ByteString -- ^ Upstream.
72− -> IO (InputStream a) -- ^ An @InputStream@ which produces records.
92+ => DecodeOptions
93+ -- ^ CSV decoding options.
94+ -> InputStream ByteString
95+ -- ^ Upstream.
96+ -> IO (InputStream (Either String a))
97+ -- ^ An @InputStream@ which produces records.
7398decodeStreamByNameWith ops input = go (decodeByNameWith ops) where
7499 -- Dispatch on the HeaderParser type.
75100 go (FailH _ e) = bomb e
76101 go (PartialH f) = Streams.read input >>= go . maybe (f BS.empty) f
77102 go (DoneH _ p) = do
78103 queue <- newIORef []
79− parser <- newIORef p
104+ parser <- newIORef (Just p)
80105 makeInputStream (dispatch queue parser input)
81106
107+--------------------------------------------------------------------------------
108+-- | Creates a new @InputStream@ which only sends valid CSV records
109+-- downstream. The first invalid record will throw an exception.
110+onlyValidRecords :: InputStream (Either String a)
111+ -- ^ Upstream.
112+ -> IO (InputStream a)
113+ -- ^ An @InputStream@ which only produces valid
114+ -- records.
115+onlyValidRecords input = makeInputStream $ do
116+ upstream <- Streams.read input
117+
118+ case upstream of
119+ Nothing -> return Nothing
120+ Just (Left err) -> bomb ("invalid CSV row: " ++ err)
121+ Just (Right x) -> return (Just x)
122+
82123--------------------------------------------------------------------------------
83124-- | Internal function which feeds data to the CSV parser.
84−dispatch :: IORef [a] -- ^ List of queued CSV records.
85− -> IORef (Parser a) -- ^ Current CSV parser state.
86− -> InputStream ByteString -- ^ Upstream.
87− -> IO (Maybe a) -- ^ Data feed downstream.
125+dispatch :: forall a. IORef [Either String a]
126+ -- ^ List of queued CSV records.
127+ -> IORef (Maybe (Parser a))
128+ -- ^ Current CSV parser state.
129+ -> InputStream ByteString
130+ -- ^ Upstream.
131+ -> IO (Maybe (Either String a))
132+ -- ^ Data feed downstream.
88133dispatch queueRef parserRef input = do
89134 queue <- readIORef queueRef
90135
@@ -92,9 +137,11 @@ dispatch queueRef parserRef input = do
92137 [] -> do
93138 parser <- readIORef parserRef
94139 case parser of
95− Fail _ e -> bomb e
96− Many xs f -> more f >> feed xs
97− Done xs -> feed xs
140+ Nothing -> return Nothing
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
98145
99146 (x:xs) -> do
100147 writeIORef queueRef xs
@@ -103,17 +150,17 @@ dispatch queueRef parserRef input = do
103150 where
104151 -- Send more data to the CSV parser. If there is no more data
105152 -- from upstream then send an empty @ByteString@.
106− more f = Streams.read input >>= writeIORef parserRef . 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
107157
108− -- Feed more data downstream or fail if some records didn't parse
109− -- correctly. The elements are wrapped in an @Either@ which
110− -- indicates decoding failures.
111− feed xs = if not $ null (lefts xs)
112− then bomb $ intercalate "\n" (lefts xs)
113− else writeIORef queueRef (rights xs) >>
114− dispatch queueRef parserRef input
158+ -- Feed records downstream.
159+ feed :: [Either String a] -> IO (Maybe (Either String a))
160+ feed xs = do writeIORef queueRef xs
161+ dispatch queueRef parserRef input
115162
116163--------------------------------------------------------------------------------
117−-- | FIXME: use a proper exception.
164+-- | Throw an exception.
118165bomb :: String -> IO a
119−bomb = fail
166+bomb = throwIO . StreamDecodingError