@@ -16,6 +16,7 @@ module System.IO.Streams.Csv.Decode
1616 , decodeStreamWith
1717 , decodeStreamByName
1818 , decodeStreamByNameWith
19+ , onlyValidRecords
1920 ) where
2021
2122--------------------------------------------------------------------------------
@@ -23,9 +24,7 @@ import Data.ByteString (ByteString)
2324import qualified Data.ByteString as BS
2425import Data.Csv hiding (Parser, decodeWith, decodeByNameWith)
2526import Data.Csv.Incremental
26−import Data.Either (lefts, rights)
2727import Data.IORef
28−import Data.List (intercalate)
2928import System.IO.Streams (InputStream, makeInputStream)
3029import qualified System.IO.Streams as Streams
3130
@@ -35,19 +34,26 @@ import qualified System.IO.Streams as Streams
3534--
3635-- Equivalent to @decodeStreamWith defaultDecodeOptions@.
3736decodeStream :: (FromRecord a)
38− => HasHeader -- ^ Whether to skip a header or not.
39− -> InputStream ByteString -- ^ Upstream.
40− -> IO (InputStream a) -- ^ An @InputStream@ which produces records.
37+ => HasHeader
38+ -- ^ Whether to skip a header or not.
39+ -> InputStream ByteString
40+ -- ^ Upstream.
41+ -> IO (InputStream (Either String a))
42+ -- ^ An @InputStream@ which produces records.
4143decodeStream = decodeStreamWith defaultDecodeOptions
4244
4345--------------------------------------------------------------------------------
4446-- | Create an @InputStream@ which decodes CSV records from the given
4547-- upstream data source.
4648decodeStreamWith :: (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.
49+ => DecodeOptions
50+ -- ^ CSV decoding options.
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.
5157decodeStreamWith ops hdr input = do
5258 queue <- newIORef []
5359 parser <- newIORef $ Just (decodeWith ops hdr)
@@ -59,17 +65,22 @@ decodeStreamWith ops hdr input = do
5965--
6066-- Equivalent to @decodeStreamByNameWith defaultDecodeOptions@.
6167decodeStreamByName :: (FromNamedRecord a)
62− => InputStream ByteString -- ^ Upstream.
63− -> IO (InputStream a) -- ^ An @InputStream@ which produces records.
68+ => InputStream ByteString
69+ -- ^ Upstream.
70+ -> IO (InputStream (Either String a))
71+ -- ^ An @InputStream@ which produces records.
6472decodeStreamByName = decodeStreamByNameWith defaultDecodeOptions
6573
6674--------------------------------------------------------------------------------
6775-- | Create an @InputStream@ which decodes CSV records from the given
6876-- upstream data source. Data should be preceded by a header.
6977decodeStreamByNameWith :: (FromNamedRecord a)
70− => DecodeOptions -- ^ CSV decoding options.
71− -> InputStream ByteString -- ^ Upstream.
72− -> IO (InputStream a) -- ^ An @InputStream@ which produces records.
78+ => DecodeOptions
79+ -- ^ CSV decoding options.
80+ -> InputStream ByteString
81+ -- ^ Upstream.
82+ -> IO (InputStream (Either String a))
83+ -- ^ An @InputStream@ which produces records.
7384decodeStreamByNameWith ops input = go (decodeByNameWith ops) where
7485 -- Dispatch on the HeaderParser type.
7586 go (FailH _ e) = bomb e
@@ -79,12 +90,32 @@ decodeStreamByNameWith ops input = go (decodeByNameWith ops) where
7990 parser <- newIORef (Just p)
8091 makeInputStream (dispatch queue parser input)
8192
93+--------------------------------------------------------------------------------
94+-- | Creates a new @InputStream@ which only sends valid CSV records
95+-- downstream. The first invalid record will throw an exception.
96+onlyValidRecords :: InputStream (Either String a)
97+ -- ^ Upstream.
98+ -> IO (InputStream a)
99+ -- ^ An @InputStream@ which only produces valid
100+ -- records.
101+onlyValidRecords input = makeInputStream $ do
102+ upstream <- Streams.read input
103+
104+ case upstream of
105+ Nothing -> return Nothing
106+ Just (Left err) -> bomb err -- FIXME: replace with throwIO
107+ Just (Right x) -> return (Just x)
108+
82109--------------------------------------------------------------------------------
83110-- | Internal function which feeds data to the CSV parser.
84−dispatch :: IORef [a] -- ^ List of queued CSV records.
85− -> IORef (Maybe (Parser a)) -- ^ Current CSV parser state.
86− -> InputStream ByteString -- ^ Upstream.
87− -> IO (Maybe a) -- ^ Data feed downstream.
111+dispatch :: IORef [Either String a]
112+ -- ^ List of queued CSV records.
113+ -> IORef (Maybe (Parser a))
114+ -- ^ Current CSV parser state.
115+ -> InputStream ByteString
116+ -- ^ Upstream.
117+ -> IO (Maybe (Either String a))
118+ -- ^ Data feed downstream.
88119dispatch queueRef parserRef input = do
89120 queue <- readIORef queueRef
90121
@@ -107,13 +138,9 @@ dispatch queueRef parserRef input = do
107138 more f = Streams.read input >>=
108139 writeIORef parserRef . Just . maybe (f BS.empty) f
109140
110− -- Feed more data downstream or fail if some records didn't parse
111− -- correctly. The elements are wrapped in an @Either@ which
112− -- indicates decoding failures.
113− feed xs = if not $ null (lefts xs)
114− then bomb $ intercalate "\n" (lefts xs)
115− else writeIORef queueRef (rights xs) >>
116− dispatch queueRef parserRef input
141+ -- Feed records downstream.
142+ feed xs = writeIORef queueRef xs >>
143+ dispatch queueRef parserRef input
117144
118145--------------------------------------------------------------------------------
119146-- | FIXME: use a proper exception.