jsonl-conduit 0.1.2 → 0.1.3

raw patch · 4 files changed

+129/−17 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +4/−0
• jsonl-conduit.cabal +2/−1
• src/JSONL/Conduit.hs +54/−16
• test/JSONL/ConduitSpec.hs +69/−0

@@ -1,3 +1,7 @@+0.1.3++add sourceFileCE : all decoding error messages are passed inside Left values+ 0.1.2  add sourceFileC_

@@ -1,5 +1,5 @@ name:                jsonl-conduit-version:             0.1.2+version:             0.1.3 synopsis:            Conduit interface to JSONL-encoded data description:         Streaming interface to JSONL-encoded files and bytestrings homepage:            https://github.com/unfoldml/jsonl@@ -35,6 +35,7 @@   type:                exitcode-stdio-1.0   hs-source-dirs:      test   main-is:             Spec.hs+  other-modules:       JSONL.ConduitSpec   build-depends:       base                      , aeson                      , bytestring

@@ -11,8 +11,11 @@   , appendFileC   -- * Decode   , jsonFromLBSC+  , jsonFromLBSCE   -- ** I/O   , sourceFileC+  , sourceFileCE+  -- * Tokenize only   , sourceFileC_   ) where @@ -28,7 +31,7 @@ import qualified Data.ByteString.Builder as BBS (toLazyByteString, Builder) import qualified Data.ByteString.Internal as BS (c2w) import qualified Data.ByteString.Char8 as BS8 (span, drop, putStrLn, putStr)-import qualified Data.ByteString.Lazy as LBS (ByteString, drop, span, toStrict, fromStrict)+import qualified Data.ByteString.Lazy as LBS (ByteString, null, drop, span, toStrict, fromStrict) -- conduit import qualified Conduit as C (ConduitT, runConduit, sourceFile, sinkFile, await, yield, mapC, unfoldC, foldMapC, foldlC, printC, sinkIOHandle) import Conduit ( (.|) , MonadResource)@@ -75,27 +78,47 @@ parseChunk :: (Monad m, FromJSON a) => C.ConduitT BS.ByteString a m () parseChunk = go mempty   where+    progress acc = case chopDecode acc of+                     Left _ -> pure ()+                     Right (y, srest) -> do+                       C.yield y+                       go srest     go acc =       if not (BS.null acc) -- buffer is non empty-      then-        case chopDecode acc of-          Left _ -> pure ()-          Right (y, srest) -> do-            C.yield y-            go srest+      then progress acc       else do         mc <- C.await -- get data from upstream         case mc of           Nothing -> pure ()-          Just x -> do-            let-              acc' = acc <> x-            case chopDecode acc' of-              Left _ -> pure ()-              Right (y, srest) -> do-                C.yield y-                go srest+          Just x -> progress (acc <> x) +-- | Read a JSONL file and stream the decoded records+--+-- NB : decoding error messages are in 'Left' values+sourceFileCE :: (MonadResource m, FromJSON a) =>+                FilePath -- ^ path of JSONL file to be read+             -> C.ConduitT () (Either String a) m ()+sourceFileCE fpath = C.sourceFile fpath .|+                     parseChunkE++parseChunkE :: (Monad m, FromJSON a) => C.ConduitT BS.ByteString (Either String a) m ()+parseChunkE = go mempty+  where+    progress acc = case chopDecode acc of+                     Left e -> C.yield (Left e)+                     Right (y, srest) -> do+                       C.yield $ Right y+                       go srest+    go acc =+      if not (BS.null acc) -- buffer is non empty+      then progress acc+      else do+        mc <- C.await -- get data from upstream+        case mc of+          Nothing -> pure ()+          Just x -> progress (acc <> x)++ -- | The outgoing stream elements are the lines of the file, i.e guaranteed not to contain newline characters -- -- NB : In case it wasn't clear, no JSON parsing is done, only string copies@@ -143,6 +166,8 @@   -- | Source a `LBS.ByteString` for JSONL records+--+-- NB in case of a decoding error the stream is stopped jsonFromLBSC :: (FromJSON a, Monad m) => LBS.ByteString -> C.ConduitT Void a m () jsonFromLBSC = C.unfoldC mk   where@@ -150,7 +175,20 @@       Right x -> Just (x, srest)       Left _ -> Nothing       where-        (s, srest) = chopLBS lbs -- LBS.span (== BS.c2w '\n') lbs+        (s, srest) = chopLBS lbs++-- | Like 'jsonFromLBSC' but all decoding errors are passed in Left values+jsonFromLBSCE :: (FromJSON a, Monad m) => LBS.ByteString -> C.ConduitT i (Either String a) m ()+jsonFromLBSCE = C.unfoldC mk+  where+    mk lbs =+      let (s, srest) = chopLBS lbs+      in+        if LBS.null s+        then Nothing+        else case eitherDecode' s of+          Right x -> Just (Right x, srest)+          Left e -> Just (Left e, srest)   -- * utilities

@@ -0,0 +1,69 @@+{-# language DeriveGeneric #-}+{-# language OverloadedStrings #-}+{-# options_ghc -Wno-unused-imports #-}+module JSONL.ConduitSpec where++import Data.Functor.Identity (Identity(..))+import GHC.Generics (Generic)++-- aeson+import qualified Data.Aeson as A (FromJSON(..), ToJSON(..))+-- bytestring+import qualified Data.ByteString as BS (ByteString)+import qualified Data.ByteString.Builder as BBS (toLazyByteString, Builder)+import qualified Data.ByteString.Internal as BS (c2w)+import qualified Data.ByteString.Char8 as BS8 (span)+import qualified Data.ByteString.Lazy as LBS (ByteString, span, toStrict, fromStrict)+-- conduit+import qualified Conduit as C (ConduitT, runConduit, runConduitRes, sourceFile, sinkFile, await, yield, yieldMany, mapC, unfoldC, foldMapC, foldlC, sinkList)+import Conduit ( (.|) , MonadResource)+-- hspec+import Test.Hspec (Spec, describe, it, shouldBe, shouldSatisfy)++import JSONL.Conduit (jsonToLBSC, jsonFromLBSC, jsonFromLBSCE, sourceFileC, sinkFileC, sourceFileCE)++spec :: Spec+spec =+  describe "JSONL.Conduit" $ do+    it "encodes a list of JSONL records" $ do+      datsC `shouldBe` datsLBS+    it "sourceFileC : reads a JSONL file and decodes its contents" $ do+      rs <- readRecords+      rs `shouldBe` dats+    it "sourceFileCE : reads a JSONL file and decodes its contents" $ do+      rs <- readRecordsE+      rs `shouldBe` datsE+    it "jsonFromLBSCE : decode a JSONL string" $ do+      rs <- C.runConduit (jsonFromLBSCE datsLBS .| C.sinkList)+      rs `shouldBe` datsE+    it "jsonFromLBSCE : decode a JSONL string - one failing entry" $ do+      rs <- C.runConduit (jsonFromLBSCE datsLBS_ng .| C.sinkList)+      rs `shouldBe` datsE_ng++data Record = R String String Int Bool deriving (Eq, Show, Generic)+instance A.ToJSON Record+instance A.FromJSON Record++datsC :: LBS.ByteString+datsC = runIdentity $ C.runConduit $ C.yieldMany dats .| jsonToLBSC++datsE, datsE_ng :: [Either String Record]+datsE = map Right dats+datsE_ng = [Right (R "Gilbert" "2013" 24 True),Left "Error in $[2]: parsing Int failed, expected Number, but encountered Boolean"]++dats :: [Record]+dats = [R "Gilbert" "2013" 24 True,R "Alexa" "2013" 29 True, R "May" "2012B" 14 False, R "Deloise" "2012A" 19 True]++datsLBS, datsLBS_ng :: LBS.ByteString+datsLBS = "[\"Gilbert\",\"2013\",24,true]\n[\"Alexa\",\"2013\",29,true]\n[\"May\",\"2012B\",14,false]\n[\"Deloise\",\"2012A\",19,true]\n"+datsLBS_ng = "[\"Gilbert\",\"2013\",24,true]\n[\"Alexa\",\"2013\",false,true]\n"++-- data asset+readRecordsE :: IO [Either String Record]+readRecordsE = C.runConduitRes $ sourceFileCE "records" .| C.sinkList++readRecords :: IO [Record]+readRecords = C.runConduitRes $ sourceFileC "records" .| C.sinkList++dump :: IO ()+dump = C.runConduitRes $ C.yieldMany dats .| sinkFileC "records"