Data.Conduit.Algorithms.Async

Higher level async processing interfaces.

Documentation

conduitPossiblyCompressedFile :: (MonadUnliftIO m, MonadResource m, MonadThrow m) => FilePath -> ConduitT () ByteString m ()

If the filename indicates a gzipped file (or, on Unix, also a bz2 file), then it reads it and uncompresses it.

To ensure that the file is closed even if the downstream finishes early, consider using withPossiblyCompressedFile.

On Windows, attempting to read from a bzip2 file, results in error.

conduitPossiblyCompressedToFile :: (MonadUnliftIO m, MonadResource m) => FilePath -> ConduitT ByteString Void m ()

If the filename indicates a gzipped file (or, on Unix, also a bz2 file), then it compresses and write with the algorithm matching the filename

On Windows, attempting to write to a bzip2 file, results in error.

withPossiblyCompressedFile :: (MonadUnliftIO m, MonadResource m, MonadThrow m) => FilePath -> (ConduitT () ByteString m () -> m a) -> m a

If the filename indicates a supported compressed file (gzip, xz, and, on Unix, bzip2), then it reads it and uncompresses it.

Usage

     withPossiblyCompressedFile fname $ src ->
         src .| mySink

Unlike conduitPossiblyCompressedFile, this ensures that the file is closed even if the conduit terminates early.

On Windows, attempting to read from a bzip2 file, results in error.

asyncMapC

Arguments

:: (MonadIO m, NFData b)

=> Int

Maximum number of worker threads

-> (a -> b)

Function to execute

-> ConduitT a b m ()

This is like map, except that each element is processed in a separate thread (up to maxThreads can be queued up at any one time). Results are evaluated to normal form (not weak-head normal form!, i.e., the structure is deeply evaluated) to ensure that the computation is fully evaluated in the worker thread.

Note that there is some overhead in threading. It is often a good idea to build larger chunks of input before passing it to asyncMapC to amortize the costs. That is, when f is not a lot of work, instead of asyncMapC f, it is sometimes better to do

   CC.conduitVector 4096 .| asyncMapC (V.map f) .| CC.concat

where CC refers to Combinators

See unorderedAsyncMapC

asyncMapEitherC :: forall a m b e. (MonadIO m, NFData b, NFData e, MonadError e m) => Int -> (a -> Either e b) -> ConduitT a b m ()

asyncMapC with error handling. The inner function can now return an error (as a Left). When the first error is seen, it throwErrors in the main monad. Note that f may be evaluated for arguments beyond the first error (as some threads may be running in the background and already processing elements after the first error).

See asyncMapC

asyncGzipTo :: forall m. (MonadIO m, MonadUnliftIO m) => Handle -> ConduitT ByteString Void m ()

A simple sink which performs gzip compression in a separate thread and writes the results to h.

See also asyncGzipToFile

asyncGzipToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> ConduitT ByteString Void m ()

Compresses the output and writes to the given file with compression being performed in a separate thread.

See also asyncGzipTo

asyncGzipFrom :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> ConduitT () ByteString m ()

A source which produces the ungzipped content from the the given handle. Note that this "reads ahead" so if you do not use all the input, the Handle will probably be left at an undefined position in the file.

See also asyncGzipFromFile

asyncGzipFromFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> ConduitT () ByteString m ()

Open and read a gzip file with the uncompression being performed in a separate thread.

See also asyncGzipFrom

asyncBzip2To :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> ConduitT ByteString Void m ()

A simple sink which performs bzip2 compression in a separate thread and writes the results to h.

See also asyncBzip2ToFile

asyncBzip2ToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> ConduitT ByteString Void m ()

Compresses the output and writes to the given file with compression being performed in a separate thread.

See also asyncBzip2To

asyncBzip2From :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> ConduitT () ByteString m ()

A source which produces the bzipped2 content from the the given handle. Note that this "reads ahead" so if you do not use all the input, the Handle will probably be left at an undefined position in the file.

See also asyncBzip2FromFile

asyncBzip2FromFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> ConduitT () ByteString m ()

Open and read a bzip2 file with the uncompression being performed in a separate thread.

See also asyncBzip2From

asyncXzTo :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> ConduitT ByteString Void m ()

A simple sink which performs lzma/xz compression in a separate thread and writes the results to h.

See also asyncXzToFile

asyncXzToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> ConduitT ByteString Void m ()

Compresses the output and writes to the given file with compression being performed in a separate thread.

See also asyncXzTo

asyncXzFrom :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m, MonadThrow m) => Handle -> ConduitT () ByteString m ()

A source which produces the unxzipped content from the the given handle. Note that this "reads ahead" so if you do not use all the input, the Handle will probably be left at an undefined position in the file.

See also asyncXzFromFile

asyncXzFromFile :: forall m. (MonadResource m, MonadUnliftIO m, MonadThrow m) => FilePath -> ConduitT () ByteString m ()

Open and read a lzma/xz file with the uncompression being performed in a separate thread.

See also asyncXzFrom

unorderedAsyncMapC

Arguments

:: (MonadIO m, NFData b)

=> Int

Maximum number of worker threads

-> (a -> b)

Function to execute

-> ConduitT a b m ()

A version of asyncMapC which can reorder results in the stream

If the order of the results is not important, this function can lead to a better use of resources if some of the chunks take longer to complete.

See asyncMapC