Potoki.Transform

Documentation

data Transform input output

Potoki integration

consume :: Consume input output -> Transform input output

produce :: (input -> Produce output) -> Transform input output

Basics

ioTransform :: IO (Transform a b) -> Transform a b

take :: Int -> Transform input input

takeWhile :: (input -> Bool) -> Transform input input

drop :: Int -> Transform input input

mapFilter :: (input -> Maybe output) -> Transform input output

filter :: (input -> Bool) -> Transform input input

just :: Transform (Maybe input) input

list :: Transform [a] a

vector :: Vector vector a => Transform (vector a) a

batch :: Vector vector a => Int -> Transform a (vector a)

Chunk the stream to vector batches of the given size.

It's useful in combination with concurrently in cases where the lifted transform's iteration is too light. Actually, there is a composed variation of concurrently, which utilizes it: concurrentlyWithBatching.

distinctBy :: (Eq comparable, Hashable comparable) => (element -> comparable) -> Transform element element

distinct :: (Eq element, Hashable element) => Transform element element

executeIO :: Transform (IO a) a

Execute the IO action.

mapInIO :: (a -> IO b) -> Transform a b

reportProgress :: (Text -> IO ()) -> Transform a a

handleProgressAndCountOnInterval :: NominalDiffTime -> (Int -> Int -> IO ()) -> Transform a a

ByteString

builderChunks :: Transform Builder ByteString

extractLines :: Transform ByteString ByteString

Convert freeform bytestring chunks into chunks, which are strictly separated by newline no matter how long they may be.

extractLinesWithoutTrail :: Transform ByteString ByteString

extractLinesConcurrently :: Int -> Transform ByteString ByteString

State

runState :: (input -> State state output) -> state -> Transform input (output, state)

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.

execState :: (input -> State state output) -> state -> Transform input state

evalState :: (input -> State state output) -> state -> Transform input output

Parsing

scan :: Scanner a -> Transform ByteString (Either Text a)

parseBytes :: Parser parsed -> Transform ByteString (Either Text parsed)

Lift an Attoparsec ByteString parser.

parseText :: Parser parsed -> Transform Text (Either Text parsed)

Lift an Attoparsec Text parser.

parseLineBytesConcurrently :: NFData a => Int -> Parser a -> Transform ByteString (Either Text a)

Lift an Attoparsec ByteString parser to a transform, which parses the lines concurrently.

parseNonEmptyLineBytesConcurrently :: NFData a => Int -> Parser a -> Transform ByteString (Either Text a)

Lift an Attoparsec ByteString parser to a transform, which parses the lines concurrently.

Concurrency

bufferize :: NFData element => Int -> Transform element element

bufferizeFlushing :: Int -> Transform input [input]

concurrently :: NFData output => Int -> Transform input output -> Transform input output

Execute the transform on the specified amount of threads. The order of the outputs produced is indiscriminate.

concurrentlyInOrder :: NFData b => Int -> Transform a b -> Transform a b

unsafeConcurrently :: NFData output => Int -> Transform input output -> Transform input output

async :: NFData input => Int -> Transform input input

A transform, which fetches the inputs asynchronously on the specified number of threads.

concurrentlyWithBatching :: (NFData a, NFData b) => Int -> Int -> Transform a b -> Transform a b

concurrentlyInOrderWithBatching :: NFData b => Int -> Int -> Transform a b -> Transform a b

File IO

deleteFile :: Transform FilePath (Either IOException ())

appendBytesToFile :: Transform (FilePath, ByteString) (Either IOException ())

writeTextToFile :: Transform (FilePath, Text) (Either IOException ())

Debugging

count :: Transform a Int

mapInIOWithCounter :: (Int -> a -> IO b) -> Transform a b

handleCount :: (Int -> IO ()) -> Transform a a

handleCountOnInterval :: NominalDiffTime -> (Int -> IO ()) -> Transform a a

Provides for progress monitoring by means of periodic measurement.

traceWithCounter :: (Int -> String) -> Transform a a

Useful for debugging