@@ -0,0 +1,120 @@
3+This file is part of the Haskell package cassava-streams. It is
4+subject to the license terms in the LICENSE file found in the
5+top-level directory of this distribution and at
6+git://pmade.com/cassava-streams/LICENSE. No part of cassava-streams
7+package, including this file, may be copied, modified, propagated, or
8+distributed except according to the terms contained in the LICENSE
13+--------------------------------------------------------------------------------
14+module System.IO.Streams.Csv.Decode
18+ , decodeStreamByNameWith
21+--------------------------------------------------------------------------------
22+import Data.ByteString (ByteString)
23+import qualified Data.ByteString as BS
24+import Data.Csv hiding (Parser, decodeWith, decodeByNameWith)
25+import Data.Csv.Incremental
26+import Data.Either (lefts, rights)
28+import System.IO.Streams (InputStream, makeInputStream)
29+import qualified System.IO.Streams as Streams
31+--------------------------------------------------------------------------------
32+-- | Create an @InputStream@ which decodes CSV records from the given
33+-- upstream data source.
35+-- Equivalent to @decodeStreamWith defaultDecodeOptions@.
36+decodeStream :: (FromRecord a)
37+ => HasHeader -- ^ Whether to skip a header or not.
38+ -> InputStream ByteString -- ^ Upstream.
39+ -> IO (InputStream a) -- ^ An @InputStream@ which produces records.
40+decodeStream = decodeStreamWith defaultDecodeOptions
42+--------------------------------------------------------------------------------
43+-- | Create an @InputStream@ which decodes CSV records from the given
44+-- upstream data source.
45+decodeStreamWith :: (FromRecord a)
46+ => DecodeOptions -- ^ CSV decoding options.
47+ -> HasHeader -- ^ Whether to skip a header or not.
48+ -> InputStream ByteString -- ^ Upstream.
49+ -> IO (InputStream a) -- ^ An @InputStream@ which produces records.
50+decodeStreamWith ops hdr input = do
52+ parser <- newIORef (decodeWith ops hdr)
53+ makeInputStream (dispatch queue parser input)
55+--------------------------------------------------------------------------------
56+-- | Create an @InputStream@ which decodes CSV records from the given
57+-- upstream data source. Data should be preceded by a header.
59+-- Equivalent to @decodeStreamByNameWith defaultDecodeOptions@.
60+decodeStreamByName :: (FromNamedRecord a)
61+ => InputStream ByteString -- ^ Upstream.
62+ -> IO (InputStream a) -- ^ An @InputStream@ which produces records.
63+decodeStreamByName = decodeStreamByNameWith defaultDecodeOptions
65+--------------------------------------------------------------------------------
66+-- | Create an @InputStream@ which decodes CSV records from the given
67+-- upstream data source. Data should be preceded by a header.
68+decodeStreamByNameWith :: (FromNamedRecord a)
69+ => DecodeOptions -- ^ CSV decoding options.
70+ -> InputStream ByteString -- ^ Upstream.
71+ -> IO (InputStream a) -- ^ An @InputStream@ which produces records.
72+decodeStreamByNameWith ops input = go (decodeByNameWith ops) where
73+ -- Dispatch on the HeaderParser type.
74+ go (FailH _ e) = bomb e
75+ go (PartialH f) = Streams.read input >>= go . maybe (f BS.empty) f
79+ makeInputStream (dispatch queue parser input)
81+--------------------------------------------------------------------------------
82+-- | Internal function which feeds data to the CSV parser.
83+dispatch :: IORef [a] -- ^ List of queued CSV records.
84+ -> IORef (Parser a) -- ^ Current CSV parser state.
85+ -> InputStream ByteString -- ^ Upstream.
86+ -> IO (Maybe a) -- ^ Data feed downstream.
87+dispatch queueRef parserRef input = do
88+ queue <- readIORef queueRef
92+ parser <- readIORef parserRef
95+ Many xs f -> more f >> feed xs
99+ writeIORef queueRef xs
103+ -- Send more data to the CSV parser. If there is no more data
104+ -- from upstream then send an empty @ByteString@.
106+ bytes <- Streams.read input
107+ writeIORef parserRef (maybe (f BS.empty) f bytes)
109+ -- Feed more data downstream or fail if some records didn't parse
110+ -- correctly. The elements are wrapped in an @Either@ which
111+ -- indicates decoding failures.
112+ feed xs = if not $ null (lefts xs)
113+ then bomb $ concat (lefts xs)
114+ else writeIORef queueRef (rights xs) >>
115+ dispatch queueRef parserRef input
117+--------------------------------------------------------------------------------
118+-- | FIXME: use a proper exception.
119+bomb :: String -> IO a