potoki 0.11.3 → 2
raw patch · 13 files changed
+12/−1157 lines, 13 filesdep −QuickCheckdep −attoparsecdep −basedep ~potoki-core
Dependencies removed: QuickCheck, attoparsec, base, base-prelude, bytestring, directory, foldl, hashable, potoki, profunctors, ptr, quickcheck-instances, random, rerebase, tasty, tasty-hunit, tasty-quickcheck, text, transformers, unagi-chan, unordered-containers, vector
Dependency ranges changed: potoki-core
Files
- library/Potoki/Consume.hs +0/−200
- library/Potoki/Fetch.hs +0/−106
- library/Potoki/Prelude.hs +0/−72
- library/Potoki/Produce.hs +0/−130
- library/Potoki/Transform.hs +10/−15
- library/Potoki/Transform/Attoparsec.hs +0/−84
- library/Potoki/Transform/Basic.hs +0/−133
- library/Potoki/Transform/ByteString.hs +0/−61
- library/Potoki/Transform/Concurrency.hs +0/−99
- library/Potoki/Transform/FileIO.hs +0/−30
- library/Potoki/Transform/State.hs +0/−43
- potoki.cabal +2/−46
- test/Main.hs +0/−138
library/Potoki/Consume.hs view
@@ -27,204 +27,4 @@ ) where -import Potoki.Prelude hiding (sum, head, fold, concat, last) import Potoki.Core.Consume-import qualified Potoki.Core.Fetch as A-import qualified Potoki.Core.Produce as H-import qualified Potoki.Core.Transform as J-import qualified Potoki.Core.IO as L-import qualified Potoki.Fetch as A-import qualified Data.ByteString as C-import qualified Data.Attoparsec.ByteString as E-import qualified Data.Attoparsec.Text as F-import qualified Data.Attoparsec.Types as I-import qualified Data.Text.IO as K-import qualified Control.Foldl as D-import qualified System.Directory as G-import qualified Potoki.Transform.Concurrency as B-import qualified Control.Monad.Trans.State.Strict as O---{-# INLINABLE head #-}-head :: Consume input (Maybe input)-head =- Consume (\ (A.Fetch fetchIO) -> fetchIO Nothing Just)--{-# INLINABLE last #-}-last :: Consume input (Maybe input)-last = - fold D.last --{-|-A faster alternative to "list",-which however constructs the list in the reverse order.--}-{-# INLINABLE reverseList #-}-reverseList :: Consume input [input]-reverseList =- Consume $ \ (A.Fetch fetchIO) -> build fetchIO []- where- build fetchIO !acc =- join (fetchIO (pure acc) (\ element -> build fetchIO (element : acc)))--{-# INLINABLE vector #-}-vector :: Consume input (Vector input)-vector =- foldInIO D.vectorM--{-# INLINABLE count #-}-count :: Consume input Int-count =- Consume $ \ (A.Fetch fetchIO) -> build fetchIO 0- where- build fetchIO !acc =- join (fetchIO (pure acc) (const (build fetchIO (succ acc))))--{-# INLINABLE concat #-}-concat :: Monoid monoid => Consume monoid monoid-concat =- Consume $ \ (A.Fetch fetchIO) -> build fetchIO mempty- where- build fetchIO !acc =- join (fetchIO (pure acc) (\ x -> build fetchIO (mappend acc x)))--{-# INLINABLE processInIO #-}-processInIO :: IO () -> (element -> IO ()) -> Consume element ()-processInIO stop process =- Consume (\ fetch -> L.fetchAndHandleAll fetch stop process)--{-# INLINABLE printBytes #-}-printBytes :: Consume ByteString ()-printBytes =- processInIO (putChar '\n') C.putStr--{-# INLINABLE printText #-}-printText :: Consume Text ()-printText =- processInIO (putChar '\n') K.putStr--{-# INLINABLE printString #-}-printString :: Consume String ()-printString =- processInIO (putChar '\n') putStr--{-|-Overwrite a file.--* Exception-free-* Automatic resource management--}-{-# INLINABLE writeBytesToFile #-}-writeBytesToFile :: FilePath -> Consume ByteString (Either IOException ())-writeBytesToFile path =- Consume $ \ fetch ->- try $ withFile path WriteMode $ \ handle ->- do- hSetBuffering handle NoBuffering- L.fetchAndHandleAll fetch (return ()) (C.hPut handle)--{-|-Append to a file.--* Exception-free-* Automatic resource management--}-{-# INLINABLE appendBytesToFile #-}-appendBytesToFile :: FilePath -> Consume ByteString (Either IOException ())-appendBytesToFile path =- Consume $ \ fetch ->- try $ withFile path AppendMode $ \ handle ->- do- hSetBuffering handle NoBuffering- L.fetchAndHandleAll fetch (return ()) (C.hPut handle)--{-# INLINABLE deleteFiles #-}-deleteFiles :: Consume FilePath (Either IOException ())-deleteFiles =- Consume $ \ fetch ->- try $ L.fetchAndHandleAll fetch (return ()) G.removeFile--{-# INLINABLE fold #-}-fold :: D.Fold input output -> Consume input output-fold (D.Fold step init finish) =- Consume $ \ (A.Fetch fetch) -> build fetch init- where- build fetch !accumulator =- join (fetch (pure (finish accumulator)) (\ !input -> build fetch (step accumulator input)))--{-# INLINABLE foldInIO #-}-foldInIO :: D.FoldM IO input output -> Consume input output-foldInIO (D.FoldM step init finish) =- Consume $ \ (A.Fetch fetch) -> build fetch =<< init- where- build fetch !accumulator =- join (fetch (finish accumulator) (\ !input -> step accumulator input >>= build fetch))--{-# INLINABLE folding #-}-folding :: D.Fold a b -> Consume a c -> Consume a (b, c)-folding (D.Fold step init extract) (Consume consumeIO) =- Consume $ \ fetch -> do- foldStateRef <- newIORef init- consumptionResult <-- consumeIO (A.handlingElements (\ element -> do- !newState <- flip step element <$> readIORef foldStateRef- writeIORef foldStateRef newState) fetch)- foldResult <- extract <$> readIORef foldStateRef- return (foldResult, consumptionResult)--{-# INLINABLE foldingInIO #-}-foldingInIO :: D.FoldM IO a b -> Consume a c -> Consume a (b, c)-foldingInIO (D.FoldM step init extract) (Consume consumeIO) =- Consume $ \ fetch -> do- foldStateRef <- newIORef =<< init- consumptionResult <-- consumeIO (A.handlingElements (\ element -> do- !newState <- flip step element =<< readIORef foldStateRef- writeIORef foldStateRef newState) fetch)- foldResult <- extract =<< readIORef foldStateRef- return (foldResult, consumptionResult)--{-# INLINE execState #-}-execState :: (a -> O.State s b) -> s -> Consume a s-execState stateFn initialState = - fold $ D.Fold (\currentState input -> snd $ O.runState (stateFn input) currentState) initialState id--{-# INLINABLE runParseResult #-}-runParseResult :: (Monoid input, Eq input) => (input -> I.IResult input output) -> Consume input (Either Text output)-runParseResult inputToResult =- Consume $ \ (A.Fetch fetchInput) ->- let- consume inputToResult =- join (fetchInput nil just)- where- nil =- just mempty- just !input =- case inputToResult input of- I.Partial newInputToResult -> consume newInputToResult- I.Done _ parsed -> return (Right parsed)- I.Fail _ contexts message -> return (Left resultMessage)- where- resultMessage =- if null contexts- then fromString message- else fromString (showString (intercalate " > " contexts) (showString ": " message))- in consume inputToResult--{-# INLINABLE parseBytes #-}-parseBytes :: E.Parser output -> Consume ByteString (Either Text output)-parseBytes =- runParseResult . E.parse--{-# INLINABLE parseText #-}-parseText :: F.Parser output -> Consume Text (Either Text output)-parseText =- runParseResult . F.parse--{-|-Execute a Consume concurrently and consume its results.--}-{-# INLINABLE concurrently #-}-concurrently :: Int -> Consume a b -> Consume b c -> Consume a c-concurrently amount consume1 consume2 =- transform (B.concurrently amount (J.consume consume1)) consume2
− library/Potoki/Fetch.hs
@@ -1,106 +0,0 @@-module Potoki.Fetch-where--import Potoki.Prelude-import Potoki.Core.Fetch-import qualified Data.Attoparsec.Types as I-import qualified Data.Attoparsec.ByteString as K-import qualified Data.Attoparsec.Text as L-import qualified Data.HashMap.Strict as B-import qualified Data.Vector as C-import qualified Data.ByteString as D-import qualified Data.Text as A-import qualified Data.Text.IO as A---{-# INLINABLE handleBytes #-}-handleBytes :: Handle -> Fetch (Either IOException ByteString)-handleBytes =- handleBytesWithChunkSize ioChunkSize--{-# INLINABLE handleBytesWithChunkSize #-}-handleBytesWithChunkSize :: Int -> Handle -> Fetch (Either IOException ByteString)-handleBytesWithChunkSize chunkSize handle =- Fetch $ \ nil just -> do- chunk <- try (D.hGetSome handle chunkSize)- case chunk of- Right "" -> return nil- _ -> return (just chunk)--{-# INLINABLE handleText #-}-handleText :: Handle -> Fetch (Either IOException Text)-handleText handle =- Fetch $ \ nil just -> do- chunk <- try (A.hGetChunk handle)- case chunk of- Right "" -> return nil- _ -> return (just chunk)--{-# INLINABLE mapFilter #-}-mapFilter :: (input -> Maybe output) -> Fetch input -> Fetch output-mapFilter mapping (Fetch fetchIO) =- Fetch $ \ nil just ->- fix $ \ loop ->- join $ fetchIO (return nil) $ \ input ->- case mapping input of- Just output -> return (just output)- Nothing -> loop--{-# INLINABLE filter #-}-filter :: (input -> Bool) -> Fetch input -> Fetch input-filter predicate (Fetch fetchIO) =- Fetch $ \ nil just ->- fix $ \ loop ->- join $ fetchIO (return nil) $ \ input ->- if predicate input- then return (just input)- else loop--{-# INLINABLE just #-}-just :: Fetch (Maybe element) -> Fetch element-just (Fetch fetchIO) =- Fetch $ \ nil just ->- fix $ \ loop ->- join $ fetchIO (return nil) $ \ case- Just output -> return (just output)- Nothing -> loop--{-# INLINABLE takeWhile #-}-takeWhile :: (element -> Bool) -> Fetch element -> Fetch element-takeWhile predicate (Fetch fetchIO) =- Fetch $ \ nil just ->- fetchIO nil $ \ input ->- if predicate input- then just input- else nil--{-# INLINABLE infiniteMVar #-}-infiniteMVar :: MVar element -> Fetch element-infiniteMVar var =- Fetch $ \ nil just ->- fmap just (takeMVar var)--{-# INLINABLE finiteMVar #-}-finiteMVar :: MVar (Maybe element) -> Fetch element-finiteMVar var =- Fetch $ \ nil just ->- fmap (maybe nil just) (takeMVar var)--{-# INLINABLE vector #-}-vector :: IORef Int -> Vector element -> Fetch element-vector indexRef vector =- Fetch $ \ nil just -> do- index <- readIORef indexRef- if index < C.length vector- then do- writeIORef indexRef (succ index)- return (just (C.unsafeIndex vector index))- else return nil--{-# INLINABLE handlingElements #-}-handlingElements :: (element -> IO ()) -> Fetch element -> Fetch element-handlingElements xRay (Fetch fetchIO) =- Fetch $ \ nil just ->- join $ fetchIO- (return nil)- (\ element -> xRay element $> just element)
− library/Potoki/Prelude.hs
@@ -1,72 +0,0 @@-module Potoki.Prelude-( - module Exports,- ioChunkSize,- textString,- unsnoc,-)-where---- base---------------------------import Data.Functor.Compose as Exports-import System.IO as Exports-import Control.Arrow as Exports (first, second)---- base-prelude---------------------------import BasePrelude as Exports hiding (first, second)---- profunctors---------------------------import Data.Profunctor.Unsafe as Exports-import Data.Profunctor.Choice as Exports-import Data.Profunctor.Strong as Exports---- text---------------------------import Data.Text as Exports (Text)---- bytestring---------------------------import Data.ByteString as Exports (ByteString)---- unordered-containers---------------------------import Data.HashMap.Strict as Exports (HashMap)---- vector---------------------------import Data.Vector as Exports (Vector)---- hashable---------------------------import Data.Hashable as Exports (Hashable)------------------------------------------------------------------------------------import qualified Data.Text as A--{-# NOINLINE ioChunkSize #-}-ioChunkSize :: Int-ioChunkSize =- shiftL 2 12--textString :: Text -> String-textString =- A.unpack--{-# INLINABLE unsnoc #-}-unsnoc :: [a] -> Maybe ([a], a)-unsnoc list =- case process list of- (init, lastMaybe) -> fmap (\ last -> (init, last)) lastMaybe- where- process list =- case list of- head : tail -> case tail of- [] -> ([], Just head)- _ -> case process tail of- (init, lastMaybe) -> (head : init, lastMaybe)- _ -> ([], Nothing)-
library/Potoki/Produce.hs view
@@ -15,134 +15,4 @@ ) where -import Potoki.Prelude import Potoki.Core.Produce-import qualified Potoki.Fetch as A-import qualified Potoki.Core.Fetch as A-import qualified Potoki.Core.Consume as E-import qualified Potoki.Core.Transform as F-import qualified Data.Attoparsec.Types as I-import qualified Data.Attoparsec.ByteString as K-import qualified Data.Attoparsec.Text as L-import qualified Data.HashMap.Strict as B-import qualified Data.ByteString as D-import qualified Data.Vector as C-import qualified System.Directory as G---{-# INLINE vector #-}-vector :: Vector input -> Produce input-vector vector =- Produce $ do- indexRef <- newIORef 0- let- fetch =- A.Fetch $ \ nil just -> do- index <- readIORef indexRef- writeIORef indexRef $! succ index- return $ case (C.!?) vector index of- Just !input -> just input- Nothing -> nil- in return (fetch, return ())--{-# INLINE hashMapRows #-}-hashMapRows :: HashMap a b -> Produce (a, b)-hashMapRows =- list . B.toList--{-|-Read from a file by path.--* Exception-free-* Automatic resource management--}-{-# INLINABLE fileBytes #-}-fileBytes :: FilePath -> Produce (Either IOException ByteString)-fileBytes path =- accessingHandle (openBinaryFile path ReadMode) A.handleBytes--{-|-Read from a file by path.--* Exception-free-* Automatic resource management--}-{-# INLINABLE fileBytesAtOffset #-}-fileBytesAtOffset :: FilePath -> Int -> Produce (Either IOException ByteString)-fileBytesAtOffset path offset =- accessingHandle acquire A.handleBytes- where- acquire =- do- handle <- openBinaryFile path ReadMode- hSeek handle AbsoluteSeek (fromIntegral offset)- return handle--{-# INLINABLE accessingHandle #-}-accessingHandle :: IO Handle -> (Handle -> A.Fetch (Either IOException a)) -> Produce (Either IOException a)-accessingHandle acquireHandle fetch =- Produce (catchIOError normal failing)- where- normal =- do- handle <- acquireHandle- return (fetch handle, catchIOError (hClose handle) (const (return ())))- failing exception =- return (pure (Left exception), return ())--{-# INLINABLE stdinBytes #-}-stdinBytes :: Produce (Either IOException ByteString)-stdinBytes =- Produce (return (A.handleBytes stdin, return ()))--{-|-Sorted subpaths of the directory.--}-{-# INLINABLE directoryContents #-}-directoryContents :: FilePath -> Produce (Either IOException FilePath)-directoryContents path =- Produce (catchIOError success failure)- where- success =- do- subPaths <- G.listDirectory path- ref <- newIORef (map (Right . mappend path . (:) '/') (sort subPaths))- return (A.list ref, return ())- failure exception =- return (pure (Left exception), return ())--{-|-Read from a file by path.--* Exception-free-* Automatic resource management--}-{-# INLINABLE fileText #-}-fileText :: FilePath -> Produce (Either IOException Text)-fileText path =- Produce (catchIOError success failure)- where- success =- do- handle <- openFile path ReadMode- return (A.handleText handle, catchIOError (hClose handle) (const (return ())))- failure exception =- return (pure (Left exception), return ())--{-|-Read from MVar.-Nothing gets interpreted as the end of input.--}-{-# INLINE finiteMVar #-}-finiteMVar :: MVar (Maybe element) -> Produce element-finiteMVar var =- Produce (return (A.finiteMVar var, return ()))--{-|-Read from MVar.-Never stops.--}-{-# INLINE infiniteMVar #-}-infiniteMVar :: MVar element -> Produce element-infiniteMVar var =- Produce (return (A.infiniteMVar var, return ()))
library/Potoki/Transform.hs view
@@ -19,18 +19,19 @@ executeIO, mapInIO, -- * ByteString- module Potoki.Transform.ByteString,+ builderChunks,+ extractLines, -- * State- R.runState,- R.execState,- R.evalState,+ runState,+ execState,+ evalState, -- * Parsing- A.parseBytes,- A.parseText,+ parseBytes,+ parseText, -- * Concurrency- N.bufferize,- N.concurrently,- N.async,+ bufferize,+ concurrently,+ async, -- * File IO deleteFile, appendBytesToFile,@@ -41,10 +42,4 @@ where import Potoki.Core.Transform-import Potoki.Transform.Basic-import Potoki.Transform.FileIO-import Potoki.Transform.ByteString-import qualified Potoki.Transform.Attoparsec as A-import qualified Potoki.Transform.Concurrency as N-import qualified Potoki.Transform.State as R
− library/Potoki/Transform/Attoparsec.hs
@@ -1,84 +0,0 @@-module Potoki.Transform.Attoparsec-where--import Potoki.Prelude hiding (take, takeWhile, filter, drop)-import Potoki.Core.Transform-import Potoki.Transform.Basic-import qualified Potoki.Fetch as A-import qualified Potoki.Core.Fetch as A-import qualified Data.Attoparsec.ByteString as K-import qualified Data.Attoparsec.Text as L-import qualified Data.Attoparsec.Types as M---{-# INLINE mapWithParseResult #-}-mapWithParseResult :: forall input parsed. (Monoid input, Eq input) => (input -> M.IResult input parsed) -> Transform input (Either Text parsed)-mapWithParseResult inputToResult =- Transform $ \ inputFetch ->- do- unconsumedRef <- newIORef mempty- finishedRef <- newIORef False- return (A.Fetch (fetchParsed inputFetch finishedRef unconsumedRef))- where- fetchParsed :: A.Fetch input -> IORef Bool -> IORef input -> forall x. x -> (Either Text parsed -> x) -> IO x- fetchParsed (A.Fetch inputFetchIO) finishedRef unconsumedRef nil just =- do- finished <- readIORef finishedRef- if finished- then return nil- else do- unconsumed <- readIORef unconsumedRef- if unconsumed == mempty- then- join $ inputFetchIO- (return nil)- (\input -> do- if input == mempty- then return nil- else matchResult (inputToResult input))- else do- writeIORef unconsumedRef mempty- matchResult (inputToResult unconsumed)- where- matchResult =- \case- M.Partial inputToResult ->- consume inputToResult- M.Done unconsumed parsed ->- do- writeIORef unconsumedRef unconsumed- return (just (Right parsed))- M.Fail unconsumed contexts message ->- do- writeIORef unconsumedRef unconsumed- writeIORef finishedRef True- return (just (Left resultMessage))- where- resultMessage =- if null contexts- then fromString message- else fromString (showString (intercalate " > " contexts) (showString ": " message))- consume inputToResult =- join $ inputFetchIO- (do- writeIORef finishedRef True- matchResult (inputToResult mempty))- (\input -> do- when (input == mempty) (writeIORef finishedRef True)- matchResult (inputToResult input))--{-|-Lift an Attoparsec ByteString parser.--}-{-# INLINE parseBytes #-}-parseBytes :: K.Parser parsed -> Transform ByteString (Either Text parsed)-parseBytes parser =- mapWithParseResult (K.parse parser)--{-|-Lift an Attoparsec Text parser.--}-{-# INLINE parseText #-}-parseText :: L.Parser parsed -> Transform Text (Either Text parsed)-parseText parser =- mapWithParseResult (L.parse parser)
− library/Potoki/Transform/Basic.hs
@@ -1,133 +0,0 @@-module Potoki.Transform.Basic-where--import Potoki.Prelude hiding (take, takeWhile, filter, drop)-import Potoki.Core.Transform-import qualified Potoki.Fetch as A-import qualified Potoki.Core.Fetch as A-import qualified Data.HashSet as C-import qualified Data.Vector as P---{-# INLINE mapFilter #-}-mapFilter :: (input -> Maybe output) -> Transform input output-mapFilter mapping =- Transform (pure . A.mapFilter mapping)--{-# INLINE filter #-}-filter :: (input -> Bool) -> Transform input input-filter predicate =- Transform (pure . A.filter predicate)--{-# INLINE just #-}-just :: Transform (Maybe input) input-just =- Transform (pure . A.just)--{-# INLINE takeWhile #-}-takeWhile :: (input -> Bool) -> Transform input input-takeWhile predicate =- Transform (pure . A.takeWhile predicate)--{-# INLINE drop #-}-drop :: Int -> Transform input input-drop amount =- Transform $ \ (A.Fetch fetchIO) -> do- countRef <- newIORef amount- return $ A.Fetch $ \ nil just -> fix $ \ loop -> do- count <- readIORef countRef- if count > 0- then do- writeIORef countRef $! pred count- loop- else fetchIO nil just--{-# INLINE list #-}-list :: Transform [a] a-list =- Transform $ \ (A.Fetch fetchListIO) -> do- bufferRef <- newIORef []- return $ A.Fetch $ \ nil just -> do- buffer <- readIORef bufferRef- case buffer of- head : tail -> do- writeIORef bufferRef tail- return (just head)- _ ->- let- fetchElementIO =- let- nilIO =- return nil- justIO input =- case input of- head : tail -> do- writeIORef bufferRef tail- return (just head)- _ -> do- writeIORef bufferRef []- return nil- in join (fetchListIO nilIO justIO)- in fetchElementIO--{-# INLINABLE vector #-}-vector :: Transform (Vector a) a-vector =- Transform $ \ (A.Fetch fetchVectorIO) -> do- indexRef <- newIORef 0- vectorRef <- newIORef mempty- return $ A.Fetch $ \ nil just -> fix $ \ loop -> do- vector <- readIORef vectorRef- index <- readIORef indexRef- if index < P.length vector- then do- writeIORef indexRef (succ index)- return (just (P.unsafeIndex vector index))- else join $ fetchVectorIO (return nil) $ \ vector -> do- writeIORef vectorRef vector- writeIORef indexRef 0- loop--{-# INLINE distinctBy #-}-distinctBy :: (Eq comparable, Hashable comparable) => (element -> comparable) -> Transform element element-distinctBy f =- Transform $ \ (A.Fetch fetch) -> do- stateRef <- newIORef mempty- return $ A.Fetch $ \ nil just -> fix $ \ loop -> join $ fetch (return nil) $ \ !input -> do- let comparable = f input- !set <- readIORef stateRef- if C.member comparable set- then loop- else do- writeIORef stateRef $! C.insert comparable set- return (just input)--{-# INLINE distinct #-}-distinct :: (Eq element, Hashable element) => Transform element element-distinct = distinctBy id--{-# INLINE mapInIO #-}-mapInIO :: (a -> IO b) -> Transform a b-mapInIO io =- Transform $ \ (A.Fetch fetch) ->- return $ A.Fetch $ \ nil just ->- join $ fetch (return nil) $ (fmap . fmap) just io--{-# INLINE ioTransform #-}-ioTransform :: IO (Transform a b) -> Transform a b-ioTransform io =- Transform $ \ fetch -> do- Transform transformIO <- io- transformIO fetch--{-|-Useful for debugging--}-traceWithCounter :: (Int -> String) -> Transform a a-traceWithCounter show =- ioTransform $ do- counter <- newIORef 0- return $ mapInIO $ \ x -> do- n <- atomicModifyIORef' counter (\ n -> (succ n, n))- putStrLn (show n)- return x
− library/Potoki/Transform/ByteString.hs
@@ -1,61 +0,0 @@-module Potoki.Transform.ByteString-where--import Potoki.Prelude hiding (filter)-import Potoki.Core.Transform-import Potoki.Transform.Basic-import Potoki.Transform.State-import qualified Potoki.Fetch as A-import qualified Potoki.Core.Fetch as A-import qualified Potoki.Core.Produce as H-import qualified Ptr.Poking as C-import qualified Ptr.ByteString as D-import qualified Data.ByteString as B-import qualified Data.ByteString.Builder as E-import qualified Data.ByteString.Lazy as F-import qualified Control.Monad.Trans.State.Strict as O---{-# INLINE builderChunks #-}-builderChunks :: Transform E.Builder ByteString-builderChunks =- produce (H.list . F.toChunks . E.toLazyByteString)--{-|-Convert freeform bytestring chunks into chunks,-which are strictly separated by newline no matter how long they may be.--}-extractLines :: Transform ByteString ByteString-extractLines =- lineList >>> filter (not . null) >>> list- -- lineList >>> list- where- lineList =- Transform $ \ (A.Fetch fetchIO) ->- do- stateRef <- newIORef Nothing- return $ A.Fetch $ \ nil just -> join $ fetchIO- (do- state <- readIORef stateRef- case state of- Just poking -> do- writeIORef stateRef Nothing- return (just [D.poking poking])- Nothing -> return nil)- (\ chunk ->- case B.split 10 chunk of- firstInput : tail -> do- state <- readIORef stateRef- let- newPoking =- fold state <> C.bytes firstInput- in case unsnoc tail of- Just (init, last) ->- do- writeIORef stateRef (Just (C.bytes last))- return (just (D.poking newPoking : init))- Nothing ->- do- writeIORef stateRef (Just newPoking)- return (just [])- _ -> return (just []))
− library/Potoki/Transform/Concurrency.hs
@@ -1,99 +0,0 @@-module Potoki.Transform.Concurrency-(- bufferize,- concurrently,- async,-)-where--import Potoki.Prelude hiding (take, takeWhile, filter)-import Potoki.Core.Transform-import qualified Potoki.Fetch as A-import qualified Potoki.Core.Fetch as A-import qualified Control.Concurrent.Chan.Unagi.Bounded as B---{-# INLINE bufferize #-}-bufferize :: Int -> Transform element element-bufferize size =- Transform $ \ (A.Fetch fetch) -> do- (inChan, outChan) <- B.newChan size- forkIO $ fix $ \ loop ->- join $ fetch- (B.writeChan inChan Nothing)- (\ !element -> B.writeChan inChan (Just element) >> loop)- return $ A.Fetch $ \ nil just -> fmap (maybe nil just) (B.readChan outChan)--{-|-Identity Transform, which ensures that the inputs are fetched synchronously.--Useful for concurrent transforms.--}-{-# INLINABLE sync #-}-sync :: Transform a a-sync =- Transform $ \ (A.Fetch fetch) -> do- activeVar <- newMVar True- return $ A.Fetch $ \ nil just -> do- active <- takeMVar activeVar- if active- then join $ fetch- (do- putMVar activeVar False- return nil)- (\ !element -> do- putMVar activeVar True- return (just element))- else do- putMVar activeVar False- return nil--{-|-Execute the transform on the specified amount of threads.-The order of the outputs produced is indiscriminate.--}-{-# INLINABLE concurrently #-}-concurrently :: Int -> Transform input output -> Transform input output-concurrently workersAmount transform =- if workersAmount == 1- then transform- else- sync >>>- concurrentlyUnsafe workersAmount transform--{-# INLINE concurrentlyUnsafe #-}-concurrentlyUnsafe :: Int -> Transform input output -> Transform input output-concurrentlyUnsafe workersAmount (Transform syncTransformIO) =- Transform $ \ fetch -> do- outChan <- newEmptyMVar- replicateM_ workersAmount $ forkIO $ do- A.Fetch fetchIO <- syncTransformIO fetch- fix $ \ loop -> join $ fetchIO- (putMVar outChan Nothing)- (\ !result -> putMVar outChan (Just result) >> loop)- activeWorkersAmountVar <- newMVar workersAmount- return $ A.Fetch $ \ nil just -> fix $ \ loop -> do- activeWorkersAmount <- takeMVar activeWorkersAmountVar- if activeWorkersAmount <= 0- then return nil- else do- fetchResult <- takeMVar outChan- case fetchResult of- Just result -> do- putMVar activeWorkersAmountVar activeWorkersAmount- return (just result)- Nothing -> do- putMVar activeWorkersAmountVar (pred activeWorkersAmount)- loop--{-|-A transform, which fetches the inputs asynchronously on the specified number of threads.--}-async :: Int -> Transform input input-async workersAmount =- Transform $ \ (A.Fetch fetchIO) -> do- chan <- newEmptyMVar - replicateM_ workersAmount $ forkIO $ fix $ \ loop -> do- fetchResult <- fetchIO Nothing Just- putMVar chan fetchResult- return (A.finiteMVar chan)
− library/Potoki/Transform/FileIO.hs
@@ -1,30 +0,0 @@-module Potoki.Transform.FileIO-where--import Potoki.Prelude hiding (take, takeWhile, filter, drop)-import Potoki.Core.Transform-import Potoki.Transform.Basic-import qualified Data.ByteString as J-import qualified Data.Text.IO as Q-import qualified System.Directory as I---{-# INLINE deleteFile #-}-deleteFile :: Transform FilePath (Either IOException ())-deleteFile =- mapInIO (try . I.removeFile)--{-# INLINE appendBytesToFile #-}-appendBytesToFile :: Transform (FilePath, ByteString) (Either IOException ())-appendBytesToFile =- mapInIO $ \ (path, bytes) ->- try $ - withFile path AppendMode $ \ handle -> - J.hPut handle bytes--{-# INLINABLE writeTextToFile #-}-writeTextToFile :: Transform (FilePath, Text) (Either IOException ())-writeTextToFile =- mapInIO $ \ (path, text) ->- try $ - Q.writeFile path text
− library/Potoki/Transform/State.hs
@@ -1,43 +0,0 @@-module Potoki.Transform.State-where--import Potoki.Prelude-import Potoki.Core.Transform-import qualified Potoki.Fetch as A-import qualified Potoki.Core.Fetch as A-import qualified Data.ByteString as B-import qualified Control.Monad.Trans.State.Strict as O---{-|-Notice that you can control the emission of output of each step-by producing a list of outputs and then composing the transform with-the "list" transform.--}-{-# INLINE runState #-}-runState :: (a -> O.State s b) -> s -> Transform a (s, b)-runState stateFn initialState =- Transform $ \ (A.Fetch fetchIO) -> do- stateRef <- newIORef initialState- return $ A.Fetch $ \ nil just -> do- let- nilIO =- return nil- justIO input =- do- currentState <- readIORef stateRef- case O.runState (stateFn input) currentState of- (output, newState) -> do- writeIORef stateRef newState- return (just (newState, output))- in join (fetchIO nilIO justIO)--{-# INLINE evalState #-}-evalState :: (a -> O.State s b) -> s -> Transform a b-evalState stateFn initialState =- runState stateFn initialState >>> arr snd--{-# INLINE execState #-}-execState :: (a -> O.State s b) -> s -> Transform a s-execState stateFn initialState =- runState stateFn initialState >>> arr fst
potoki.cabal view
@@ -1,7 +1,7 @@ name: potoki version:- 0.11.3+ 2 synopsis: Simple streaming in IO description:@@ -72,50 +72,6 @@ Potoki.IO Potoki.Produce Potoki.Transform- other-modules:- Potoki.Fetch- Potoki.Prelude- Potoki.Transform.Attoparsec- Potoki.Transform.Basic- Potoki.Transform.ByteString- Potoki.Transform.Concurrency- Potoki.Transform.FileIO- Potoki.Transform.State build-depends:- attoparsec >=0.13 && <0.15,- base >=4.7 && <5,- base-prelude <2,- bytestring ==0.10.*,- directory >=1.3 && <2,- foldl >=1.3 && <2,- hashable >=1 && <2,- potoki-core >=1.5 && <1.6,- profunctors >=5.2 && <6,- ptr >=0.16.2 && <0.17,- text >=1 && <2,- transformers >=0.5 && <0.6,- unagi-chan >=0.4 && <0.5,- unordered-containers >=0.2 && <0.3,- vector >=0.12 && <0.13+ potoki-core >=2 && <2.1 -test-suite test- type:- exitcode-stdio-1.0- hs-source-dirs:- test- main-is:- Main.hs- default-extensions:- Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples- default-language:- Haskell2010- build-depends:- attoparsec,- potoki,- QuickCheck >=2.8.1 && <3,- quickcheck-instances >=0.3.11 && <0.4,- random >=1.1 && <2,- rerebase >=1.1 && <2,- tasty >=0.12 && <0.13,- tasty-hunit >=0.9 && <0.11,- tasty-quickcheck >=0.9 && <0.10
− test/Main.hs
@@ -1,138 +0,0 @@-module Main where--import Prelude hiding (first, second)-import Control.Arrow-import Test.QuickCheck.Instances-import Test.Tasty-import Test.Tasty.Runners-import Test.Tasty.HUnit-import Test.Tasty.QuickCheck-import qualified Potoki.IO as C-import qualified Potoki.Consume as D-import qualified Potoki.Transform as A-import qualified Potoki.Produce as E-import qualified Data.Attoparsec.ByteString.Char8 as B-import qualified Data.ByteString as F-import qualified Data.Vector as G-import qualified System.Random as H---main =- defaultMain $- testGroup "All tests" $- [- testCase "vector to list" $ do- result <- C.produceAndConsume (E.vector (G.fromList [1,2,3])) (D.list)- assertEqual "" [1,2,3] result- ,- testCase "just" $ do- result <- C.produceAndConsume (E.list [Just 1, Nothing, Just 2]) (D.transform A.just D.list)- assertEqual "" [1,2] result- ,- testCase "transform,consume,take" $ do- let- transform = A.consume (D.transform (A.take 3) D.list)- consume = D.transform transform D.list- produceAndConsume list = C.produceAndConsume (E.list list) (consume)- assertEqual "" [[1,2,3], [4,5,6], [7,8]] =<< produceAndConsume [1,2,3,4,5,6,7,8]- assertEqual "" [[1,2,3], [4,5,6], [7,8,9]] =<< produceAndConsume [1,2,3,4,5,6,7,8,9]- assertEqual "" [] =<< produceAndConsume ([] :: [Int])- ,- testCase "File reading" $ do- let produce =- E.transform (arr (either (const Nothing) Just) >>> A.just) $- E.fileBytes "samples/1"- result <- C.produceAndConsume produce (fmap F.length D.concat)- assertEqual "" 17400 result- ,- transform- ,- parsing- ]--transform :: TestTree-transform =- testGroup "Transform" $- [- testCase "Order" $ do- let- list = [Left 1, Left 2, Right 'z', Left 2, Right 'a', Left 1, Right 'b', Left 0, Right 'x', Left 4, Left 3]- transform = left (A.consume (D.transform (A.take 2) D.sum))- result <- C.produceAndConsume (E.list list) (D.transform transform D.list)- assertEqual "" [Left 3, Right 'z', Left 2, Right 'a', Left 1, Right 'b', Left 0, Right 'x', Left 7] result- ,- testCase "Interrupted order" $ do- let- list = [Left 1, Left 2, Right 'a']- transform = left (A.consume (D.transform (A.take 3) D.sum))- result <- C.produceAndConsume (E.list list) (D.transform transform D.list)- assertEqual "" [Left 3, Right 'a'] result- ,- testCase "Distinct" $ do- let- list = [1,2,3,2,3,2,1,4,1] :: [Int]- result <- C.produceAndConsume (E.list list) (D.transform A.distinct D.list)- assertEqual "" [1,2,3,4] result- ,- testCase "Distinct By" $ do- let- list = [(1, ""),(2, ""),(3, ""),(2, ""),(3, ""),(2, ""),(1, ""),(4, ""),(1, "")] :: [(Int, String)]- result <- C.produceAndConsume (E.list list) (D.transform (A.distinctBy fst) D.list)- assertEqual "" [(1, ""),(2, ""),(3, ""),(4, "")] result- ,- testCase "Concurrently" $ do- let- list = [1..20000]- produce = E.list list- transform =- A.concurrently 12 $- arr (\ x -> H.randomRIO (0, 100) >>= threadDelay >> return x) >>>- A.executeIO- consume = D.transform transform D.list- result <- C.produceAndConsume produce consume- assertBool "Is dispersed" (list /= result)- assertEqual "Contains no duplicates" 0 (length result - length (nub result))- assertEqual "Equals the original once sorted" list (sort result)- ,- testProperty "Line" $ \ chunks ->- let- expected =- mconcat chunks- actual =- unsafePerformIO (C.produceAndConsume produce consume)- where- produce =- E.list chunks- consume =- rmap (mconcat . intersperse "\n") $- D.transform A.extractLines D.list- in expected === actual- ]--parsing :: TestTree-parsing =- testGroup "Parsing" $- [- testCase "Sample 1" $ do- let parser = B.double <* B.char ','- transform = arr (either (const Nothing) Just) >>> A.just >>> A.parseBytes parser- produce = E.transform transform (E.fileBytes "samples/1")- result <- C.produceAndConsume produce D.count- assertEqual "" 4350 result- ,- testCase "Sample 1 greedy" $ do- let parser = B.sepBy B.double (B.char ',')- transform = arr (either (const Nothing) Just) >>> A.just >>> A.parseBytes parser- produce = E.transform transform (E.fileBytes "samples/1")- result <- C.produceAndConsume produce D.list- assertEqual "" [Right 4350] (fmap (fmap length) result)- ,- testCase "Split chunk" $- let- produce = E.list ["1", "2", "3"]- parser = B.anyChar- transform = A.parseBytes parser >>> arr (either (const Nothing) Just) >>> A.just- consume = D.transform transform D.count- in do- assertEqual "" 3 =<< C.produceAndConsume produce consume- ]