Jet.Internal

Tampering with the internals lets you write invalid Jets that don't respect stop signals from consumers, so be careful.

Also, the internals expose Line and ByteBundle as thin coats of paint over lazy text and lazy bytestring, respectively.

Documentation

>>> :set -XTypeApplications
>>> :set -XImportQualifiedPost
>>> :set -XScopedTypeVariables
>>> :set -XLambdaCase
>>> :set -XNumDecimals
>>> import Jet (Jet, (&))
>>> import Jet qualified as J
>>> import Control.Foldl qualified as L
>>> import Control.Concurrent
>>> import Data.IORef
>>> import Data.Text qualified as T

newtype Jet a

A Jet is a sequence of values produced through IO effects.

It allows consuming the elements as they are produced and doesn't force them to be present in memory all at the same time, unlike functions like replicateM from base.

Constructors

Jet

Fields

• runJet :: forall s. (s -> Bool) -> (s -> a -> IO s) -> s -> IO s

Instances

MonadIO Jet

>>> liftIO (putStrLn "foo") <> liftIO (putStrLn "bar") & J.toList
foo
bar
[(),()]

Defined in Jet.Internal

Methods

liftIO :: IO a -> Jet a

Alternative Jet

Same as Monoid.

Defined in Jet.Internal

Methods

empty :: Jet a

(<|>) :: Jet a -> Jet a -> Jet a

some :: Jet a -> Jet [a]

many :: Jet a -> Jet [a]

Applicative Jet

Similar to the instance for pure lists, that generates combinations.

>>> (,) <$> J.each "ab" <*> J.each "cd" & J.toList
[('a','c'),('a','d'),('b','c'),('b','d')]

Defined in Jet.Internal

Methods

pure :: a -> Jet a

(<*>) :: Jet (a -> b) -> Jet a -> Jet b

liftA2 :: (a -> b -> c) -> Jet a -> Jet b -> Jet c

(*>) :: Jet a -> Jet b -> Jet b

(<*) :: Jet a -> Jet b -> Jet a

Functor Jet

Maps over the yielded elements. (<&>) can be used to put the function last.

>>> J.each "aa" <&> succ & J.toList
"bb"

Defined in Jet.Internal

Methods

fmap :: (a -> b) -> Jet a -> Jet b

(<$) :: a -> Jet b -> Jet a

Monad Jet

Similar to the instance for pure lists, that does search.

>>> :{
do string <- J.each ["ab","cd"]
   J.each string
&
J.toList
:}
"abcd"

Defined in Jet.Internal

Methods

(>>=) :: Jet a -> (a -> Jet b) -> Jet b

(>>) :: Jet a -> Jet b -> Jet b

return :: a -> Jet a

MonadPlus Jet

Same as Monoid

Defined in Jet.Internal

Methods

mzero :: Jet a

mplus :: Jet a -> Jet a -> Jet a

MonadFail Jet

A failed pattern-match in a do-block produces mzero.

>>> :{
do Just c <- J.each [Nothing, Just 'a', Nothing, Just 'b']
   pure c
& J.toList
:}
"ab"

Defined in Jet.Internal

Methods

fail :: String -> Jet a

Monoid (Jet a)

mempty is the empty Jet.

>>> mempty <> J.each "ab" <> mempty & J.toList
"ab"

Defined in Jet.Internal

Methods

mempty :: Jet a

mappend :: Jet a -> Jet a -> Jet a

mconcat :: [Jet a] -> Jet a

Semigroup (Jet a)

Jet concatenation.

>>> J.each "ab" <> J.each "cd" & J.toList
"abcd"

Defined in Jet.Internal

Methods

(<>) :: Jet a -> Jet a -> Jet a

sconcat :: NonEmpty (Jet a) -> Jet a

stimes :: Integral b => b -> Jet a -> Jet a

run :: Jet a -> (s -> Bool) -> (s -> a -> IO s) -> s -> IO s

Go through the elements produced by a Jet, while threading an state s and possibly performing some effect.

The caller is the one who chooses the type of the state s, and must pass an initial value for it. The state is kept in weak-head normal form.

The caller must also provide a predicate on the state that informs the Jet when to stop producing values: whenever the predicate returns True.

consume :: Jet a -> (s -> a -> IO s) -> s -> IO s

Like run, but always goes through all elements produced by the Jet.

Equivalent to run (const False).

for :: Jet a -> (a -> IO b) -> Jet b

for_ :: Jet a -> (a -> IO b) -> IO ()

traverse :: (a -> IO b) -> Jet a -> Jet b

Apply an effectful transformation to each element in a Jet.

>>> :{
J.each "abc"
& J.traverse (\c -> let c' = succ c in putStrLn ([c] ++ " -> " ++ [c']) *> pure c')
& J.toList
:}
a -> b
b -> c
c -> d
"bcd"

traverse_ :: (a -> IO b) -> Sink a

drain :: Sink a

Go through the Jet only for the IO effects, discarding all yielded elements.

each :: forall a f. Foldable f => f a -> Jet a

Build a Jet from any Foldable container

>>> J.each [True,False] & J.toList
[True,False]

repeat :: a -> Jet a

>>> J.repeat True & J.take 2 & J.toList
[True,True]

repeatIO :: IO a -> Jet a

>>> J.repeatIO (putStrLn "hi" *> pure True) & J.take 2 & J.toList
hi
hi
[True,True]

replicate :: Int -> a -> Jet a

>>> J.replicate 2 True & J.toList
[True,True]

replicateIO :: Int -> IO a -> Jet a

>>> J.replicateIO 2 (putStrLn "hi" *> pure True) & J.toList
hi
hi
[True,True]

Don't confuse this with Control.Monad.replicateM :: Int -> Jet a -> Jet [a] which has a combinatorial behavior.

iterate :: (a -> a) -> a -> Jet a

>>> J.iterate succ (1 :: Int) & J.take 2 & J.toList
[1,2]

iterateIO :: (a -> IO a) -> a -> Jet a

>>> J.iterateIO (\x -> putStrLn "hi" *> pure (succ x)) (1 :: Int) & J.take 2 & J.toList
hi
[1,2]

unfold :: (b -> Maybe (a, b)) -> b -> Jet a

>>> J.unfold (\case [] -> Nothing ; c : cs -> Just (c,cs)) "abc" & J.toList
"abc"

unfoldIO :: (b -> IO (Maybe (a, b))) -> b -> Jet a

>>> :{
J.unfoldIO (\x -> do putStrLn "hi"
                     pure $ case x of
                        [] -> Nothing
                        c : cs -> Just (c,cs))
           "abc"
& J.toList
:}
hi
hi
hi
hi
"abc"

untilEOF :: (handle -> IO Bool) -> (handle -> IO a) -> handle -> Jet a

>>> j = J.untilEOF System.IO.hIsEOF System.IO.hGetLine :: Handle -> Jet String

untilNothing :: IO (Maybe a) -> Jet a

>>> :{
do ref <- newIORef "abc"
   let pop = atomicModifyIORef ref (\case [] -> ([], Nothing)
                                          x : xs -> (xs, Just x))
   J.untilNothing pop & J.toList
:}
"abc"

toList :: Jet a -> IO [a]

Convert to a regular list. This breaks streaming.

>>> J.each "abc" & J.toList
"abc"

Alternatively, we can use fold in combination with list form the foldl library:

>>> L.purely (J.fold (J.each "abc")) L.list
"abc"

which is more verbose, but more composable.

length :: Jet a -> IO Int

Returns the number of elements yielded by the Jet, exhausting it in the process.

>>> J.each "abc" & J.length
3

Alternatively, we can use fold in combination with length form the foldl library:

>>> L.purely (J.fold (J.each "abc")) L.length
3

which is more verbose, but more composable.

data Pair a b

Constructors

Pair !a !b

Instances

(Show a, Show b) => Show (Pair a b)

Defined in Jet.Internal

Methods

showsPrec :: Int -> Pair a b -> ShowS

show :: Pair a b -> String

showList :: [Pair a b] -> ShowS

pairExtract :: Pair a b -> b

pairEnv :: Pair a b -> a

data Triple a b c

Constructors

Triple !a !b !c

tripleExtract :: Triple a b c -> c

drop :: Int -> Jet a -> Jet a

>>> J.each "abc" & J.drop 2 & J.toList
"c"

data DropState

Constructors

StillDropping

DroppingNoMore

dropWhile :: (a -> Bool) -> Jet a -> Jet a

>>> J.each [1..5] & J.dropWhile (<3) & J.toList
[3,4,5]

dropWhileIO :: (a -> IO Bool) -> Jet a -> Jet a

take :: Int -> Jet a -> Jet a

>>> J.each "abc" & J.take 2 & J.toList
"ab"

limit :: Int -> Jet a -> Jet a

Synonym for take.

data TakeState

Constructors

StillTaking

TakingNoMore

takeWhile :: (a -> Bool) -> Jet a -> Jet a

>>> J.each [1..] & J.takeWhile (<5) & J.toList
[1,2,3,4]

takeWhileIO :: (a -> IO Bool) -> Jet a -> Jet a

filter :: (a -> Bool) -> Jet a -> Jet a

>>> J.each "abc" & J.filter (=='a') & J.toList
"a"

filterIO :: (a -> IO Bool) -> Jet a -> Jet a

mapAccum :: (a -> b -> (a, c)) -> a -> Jet b -> Jet c

Behaves like a combination of fmap and foldl; it applies a function to each element of a structure passing an accumulating parameter from left to right.

The resulting Jet has the same number of elements as the original one.

Unlike mapAccumL, it doesn't make the final state available.

>>> J.each [1,2,3,4] & J.mapAccum (\a b -> (a + b,a)) 0 & J.toList
[0,1,3,6]

mapAccumIO :: (a -> b -> IO (a, c)) -> a -> Jet b -> Jet c

data Touched

Constructors

NotYetTouched

AlreadyTouched

intersperse :: a -> Jet a -> Jet a

>>> J.each "abc" & J.intersperse '-' & J.toList
"a-b-c"

zip :: Foldable f => f a -> Jet b -> Jet (a, b)

>>> J.each "abc" & J.zip [1..] & J.toList
[(1,'a'),(2,'b'),(3,'c')]

>>> J.each [1..] & J.zip "abc" & J.toList
[('a',1),('b',2),('c',3)]

zipWith :: Foldable f => (a -> b -> c) -> f a -> Jet b -> Jet c

zipIO :: Foldable f => f (IO a) -> Jet b -> Jet (a, b)

zipWithIO :: Foldable f => (a -> b -> IO c) -> f (IO a) -> Jet b -> Jet c

Zips a list of IO actions with a Jet, where the combining function can also have effects.

If the list of actions is exhausted, the Jet stops:

>>> J.each [1..] <&> show & zipWithIO (\c1 c2 -> putStrLn (c1 ++ c2)) [pure "a", pure "b"] & J.toList
a1
b2
[(),()]

withFile :: FilePath -> IOMode -> Jet Handle

Opens a file and makes the Handle available to all following statements in the do-block.

Notice that it's often simpler to use the JetSource (for reading) and JetSink (for writing) instances of File.

bracket

Arguments

:: IO a

allocator

-> (a -> IO b)

finalizer

-> Jet a

>>> :{
do r <- J.bracket (putStrLn "allocating" *> pure "foo") (\r -> putStrLn $ "deallocating " ++ r)
   liftIO $ putStrLn $ "using resource " ++ r
& drain
:}
allocating
using resource foo
deallocating foo

bracket_

Arguments

:: IO a

allocator

-> IO b

finalizer

-> Jet ()

bracketOnError

Arguments

:: IO a

allocator

-> (a -> IO b)

finalizer

-> Jet a

finally :: IO a -> Jet ()

Notice how the finalizer runs even when we limit the Jet:

>>> :{
do J.finally (putStrLn "hi") -- protects statements below
   liftIO (putStrLn "hey")
   J.each "abc"
& J.limit 2
& J.toList
:}
hey
hi
"ab"

But if the protected Jet is not consumed at all, the finalizer might not run.

>>> :{
do J.finally (putStrLn "hi") -- protects statements below
   liftIO (putStrLn "hey")
   J.each "abc"
& J.limit 0
& J.toList
:}
""

onException :: IO a -> Jet ()

control :: (forall x. (resource -> IO x) -> IO x) -> Jet resource

Lift a control operation (like bracket) for which the callback uses the allocated resource.

BEWARE: the control operation shouldn't do weird things like executing the callback twice.

control_ :: (forall x. IO x -> IO x) -> Jet ()

Lift a control operation (like finally) for which the callback doesn't use the allocated resource.

BEWARE: the control operation shouldn't do weird things like executing the callback twice.

fold :: Jet a -> (s -> a -> s) -> s -> (s -> r) -> IO r

>>> L.purely (J.fold (J.each "abc")) ((,) <$> L.list <*> L.length)
("abc",3)

foldIO :: Jet a -> (s -> a -> IO s) -> IO s -> (s -> IO r) -> IO r

>>> L.impurely (J.foldIO (J.each "abc")) (L.FoldM (\() c -> putStrLn [c]) (pure ()) pure *> L.generalize L.length)
a
b
c
3

data ChunkSize

Constructors

DefaultChunkSize

ChunkSize Int

ChunkSize1K

ChunkSize4K

ChunkSize8K

ChunkSize16K

ChunkSize1M

ChunkSize2M

Instances

Show ChunkSize

Defined in Jet.Internal

Methods

showsPrec :: Int -> ChunkSize -> ShowS

show :: ChunkSize -> String

showList :: [ChunkSize] -> ShowS

chunkSize :: ChunkSize -> Int

class JetSource a source where

Helper multi-parameter typeclass for creating Jet values out of a variety of common sources.

Because there's no functional dependency, sometimes we need to use TypeApplications to give the compiler a hint about the type of elements we want to produce. For example, here we want Lines and not, say, ByteStrings:

>>> action = J.jet @Line (File "foo.txt") & J.sink J.stdout

Methods

jet :: source -> Jet a

Instances

JetSource ByteString Handle

Defined in Jet.Internal

Methods

jet :: Handle -> Jet ByteString

JetSource Line Handle

Uses the default system locale.

Defined in Jet.Internal

Methods

jet :: Handle -> Jet Line

JetSource a Handle => JetSource a File

Defined in Jet.Internal

Methods

jet :: File -> Jet a

bytes :: ChunkSize -> Handle -> Jet ByteString

accumByteLengths :: Jet ByteString -> Jet (Int, ByteString)

data AmIContinuing

Constructors

Continuing

NotContinuing

Instances

Show AmIContinuing

Defined in Jet.Internal

Methods

showsPrec :: Int -> AmIContinuing -> ShowS

show :: AmIContinuing -> String

showList :: [AmIContinuing] -> ShowS

bytesOverBuckets :: [Int] -> Splitter ByteString ByteString

Splits a stream of bytes into groups bounded by maximum byte sizes. When one group "fills up", the next one is started.

When the list of buckets sizes is exhausted, all incoming bytes are put into the same unbounded group.

Useful in combination with recast.

newtype ByteBundle

A sequence of bytes that we might want to keep together.

Constructors

ByteBundle ByteString

Instances

Monoid ByteBundle

Defined in Jet.Internal

Methods

mempty :: ByteBundle

mappend :: ByteBundle -> ByteBundle -> ByteBundle

mconcat :: [ByteBundle] -> ByteBundle

Semigroup ByteBundle

Defined in Jet.Internal

Methods

(<>) :: ByteBundle -> ByteBundle -> ByteBundle

sconcat :: NonEmpty ByteBundle -> ByteBundle

stimes :: Integral b => b -> ByteBundle -> ByteBundle

Show ByteBundle

Defined in Jet.Internal

Methods

showsPrec :: Int -> ByteBundle -> ShowS

show :: ByteBundle -> String

showList :: [ByteBundle] -> ShowS

JetSink ByteBundle Handle

Defined in Jet.Internal

Methods

sink :: Handle -> Sink ByteBundle

JetSink ByteBundle [BoundedSize File]

Distributes incoming bytes through a sequence of files. Once a file is full, we start writing the next one.

Each ByteBundle value is garanteed to be written to a single file. If a file turns out to be too small for even a single ByteBundle value, a BucketOverflow exception is thrown.

Defined in Jet.Internal

Methods

sink :: [BoundedSize File] -> Sink ByteBundle

bundle :: Foldable f => f ByteString -> ByteBundle

Constructs a ByteBundle out of the bytes of some Foldable container.

bundleLength :: ByteBundle -> Int

Length in bytes.

bundleBytes :: ByteBundle -> Jet ByteString

data BucketOverflow

Exception thrown when we try to write too much data in a size-bounded destination.

Constructors

BucketOverflow

Instances

Exception BucketOverflow

Defined in Jet.Internal

Methods

toException :: BucketOverflow -> SomeException

fromException :: SomeException -> Maybe BucketOverflow

displayException :: BucketOverflow -> String

backtraceDesired :: BucketOverflow -> Bool

Show BucketOverflow

Defined in Jet.Internal

Methods

showsPrec :: Int -> BucketOverflow -> ShowS

show :: BucketOverflow -> String

showList :: [BucketOverflow] -> ShowS

byteBundlesOverBuckets :: [Int] -> Splitter ByteBundle ByteString

Splits a stream of ByteBundles into groups bounded by maximum byte sizes. Bytes belonging to the same ByteBundle are always put in the same group. When one group "fills up", the next one is started.

When the list of buckets sizes is exhausted, all incoming bytes are put into the same unbounded group.

Useful in combination with recast.

THROWS:

• BucketOverflow exception if the size bound of a group turns out to be too small for holding even a single ByteBundle value.

decodeUtf8 :: Jet ByteString -> Jet Text

THROWS:

• UnicodeException

encodeUtf8 :: Jet Text -> Jet ByteString

newtype Line

A line of text.

While it is guaranteed that the Lines coming out of the lines function do not contain newlines, that invariant is not otherwise enforced.

Constructors

Line_ Text

Instances

Monoid Line

Defined in Jet.Internal

Methods

mempty :: Line

mappend :: Line -> Line -> Line

mconcat :: [Line] -> Line

Semigroup Line

Defined in Jet.Internal

Methods

(<>) :: Line -> Line -> Line

sconcat :: NonEmpty Line -> Line

stimes :: Integral b => b -> Line -> Line

IsString Line

Defined in Jet.Internal

Methods

fromString :: String -> Line

Show Line

Defined in Jet.Internal

Methods

showsPrec :: Int -> Line -> ShowS

show :: Line -> String

showList :: [Line] -> ShowS

Eq Line

Defined in Jet.Internal

Methods

(==) :: Line -> Line -> Bool

(/=) :: Line -> Line -> Bool

Ord Line

Defined in Jet.Internal

Methods

compare :: Line -> Line -> Ordering

(<) :: Line -> Line -> Bool

(<=) :: Line -> Line -> Bool

(>) :: Line -> Line -> Bool

(>=) :: Line -> Line -> Bool

max :: Line -> Line -> Line

min :: Line -> Line -> Line

JetSink Line Handle

Uses the default system locale. Adds newlines.

Defined in Jet.Internal

Methods

sink :: Handle -> Sink Line

JetSource Line Handle

Uses the default system locale.

Defined in Jet.Internal

Methods

jet :: Handle -> Jet Line

pattern Line :: StrictText -> Line

Unidirectional pattern that allows converting a Line into a Text during pattern-matching.

lineToText :: Line -> Text

Converts a Line back to text, without adding the newline.

lineToUtf8 :: Line -> ByteBundle

Converts a Line to an utf8-encdoed ByteBundle, without adding the newline.

textToLine :: Text -> Line

newline :: Text

Data.Text.singleton '\n'

textToUtf8 :: Text -> ByteBundle

lineContains :: Text -> Line -> Bool

lineBeginsWith :: Text -> Line -> Bool

prefixLine :: Text -> Line -> Line

Adds the Text to the beginning of the Line.

stringToLine :: String -> Line

isEmptyLine :: Line -> Bool

emptyLine :: Line

data NewlineForbidden

Exception thrown when we find newlines in functions which don't accept them.

A direct copy of the NewlineForbidden exception from the turtle package.

Constructors

NewlineForbidden

Instances

Exception NewlineForbidden

Defined in Jet.Internal

Methods

toException :: NewlineForbidden -> SomeException

fromException :: SomeException -> Maybe NewlineForbidden

displayException :: NewlineForbidden -> String

backtraceDesired :: NewlineForbidden -> Bool

Show NewlineForbidden

Defined in Jet.Internal

Methods

showsPrec :: Int -> NewlineForbidden -> ShowS

show :: NewlineForbidden -> String

showList :: [NewlineForbidden] -> ShowS

removeTrailingCarriageReturn :: Text -> Text

lines :: Jet Text -> Jet Line

unlines :: Jet Line -> Jet Text

downstream :: (s -> Bool) -> (s -> x -> IO s) -> [x] -> s -> IO s

type Sink a = Jet a -> IO ()

A function that consumes a Jet totally or partially, without returning a result.

class JetSink a target where

Helper multi-parameter typeclass for creating Jet-consuming functions out of a variety of common destinations.

>>> J.each ["aaa","bbb","ccc"] <&> J.stringToLine & J.sink J.stdout
aaa
bbb
ccc

Methods

sink :: target -> Sink a

Instances

JetSink ByteString Handle

Defined in Jet.Internal

Methods

sink :: Handle -> Sink ByteString

JetSink ByteBundle Handle

Defined in Jet.Internal

Methods

sink :: Handle -> Sink ByteBundle

JetSink Line Handle

Uses the default system locale. Adds newlines.

Defined in Jet.Internal

Methods

sink :: Handle -> Sink Line

JetSink Text Handle

Uses the default system locale.

Defined in Jet.Internal

Methods

sink :: Handle -> Sink Text

JetSink a Handle => JetSink a File

Defined in Jet.Internal

Methods

sink :: File -> Sink a

JetSink ByteString [BoundedSize File]

Distributes incoming bytes through a sequence of files. Once a file is full, we start writing the next one.

Defined in Jet.Internal

Methods

sink :: [BoundedSize File] -> Sink ByteString

JetSink ByteBundle [BoundedSize File]

Distributes incoming bytes through a sequence of files. Once a file is full, we start writing the next one.

Each ByteBundle value is garanteed to be written to a single file. If a file turns out to be too small for even a single ByteBundle value, a BucketOverflow exception is thrown.

Defined in Jet.Internal

Methods

sink :: [BoundedSize File] -> Sink ByteBundle

newtype File

FilePaths are plain strings. This newtype provides a small measure of safety over them.

Constructors

File

Fields

• getFilePath :: FilePath

Instances

Show File

Defined in Jet.Internal

Methods

showsPrec :: Int -> File -> ShowS

show :: File -> String

showList :: [File] -> ShowS

JetSink a Handle => JetSink a File

Defined in Jet.Internal

Methods

sink :: File -> Sink a

JetSource a Handle => JetSource a File

Defined in Jet.Internal

Methods

jet :: File -> Jet a

JetSink ByteString [BoundedSize File]

Distributes incoming bytes through a sequence of files. Once a file is full, we start writing the next one.

Defined in Jet.Internal

Methods

sink :: [BoundedSize File] -> Sink ByteString

JetSink ByteBundle [BoundedSize File]

Distributes incoming bytes through a sequence of files. Once a file is full, we start writing the next one.

Each ByteBundle value is garanteed to be written to a single file. If a file turns out to be too small for even a single ByteBundle value, a BucketOverflow exception is thrown.

Defined in Jet.Internal

Methods

sink :: [BoundedSize File] -> Sink ByteBundle

data BoundedSize x

The maximum size in bytes of some destination into which we write the bytes produced by a Jet.

Constructors

BoundedSize Int x

Instances

JetSink ByteString [BoundedSize File]

Distributes incoming bytes through a sequence of files. Once a file is full, we start writing the next one.

Defined in Jet.Internal

Methods

sink :: [BoundedSize File] -> Sink ByteString

JetSink ByteBundle [BoundedSize File]

Distributes incoming bytes through a sequence of files. Once a file is full, we start writing the next one.

Each ByteBundle value is garanteed to be written to a single file. If a file turns out to be too small for even a single ByteBundle value, a BucketOverflow exception is thrown.

Defined in Jet.Internal

Methods

sink :: [BoundedSize File] -> Sink ByteBundle

Read x => Read (BoundedSize x)

Defined in Jet.Internal

Methods

readsPrec :: Int -> ReadS (BoundedSize x)

readList :: ReadS [BoundedSize x]

readPrec :: ReadPrec (BoundedSize x)

readListPrec :: ReadPrec [BoundedSize x]

Show x => Show (BoundedSize x)

Defined in Jet.Internal

Methods

showsPrec :: Int -> BoundedSize x -> ShowS

show :: BoundedSize x -> String

showList :: [BoundedSize x] -> ShowS

makeAllocator :: BoundedSize File -> IO Handle

newtype DList a

Constructors

DList

Fields

• runDList :: [a] -> [a]

Instances

Monoid (DList a)

Defined in Jet.Internal

Methods

mempty :: DList a

mappend :: DList a -> DList a -> DList a

mconcat :: [DList a] -> DList a

Semigroup (DList a)

Defined in Jet.Internal

Methods

(<>) :: DList a -> DList a -> DList a

sconcat :: NonEmpty (DList a) -> DList a

stimes :: Integral b => b -> DList a -> DList a

makeDList :: [a] -> DList a

closeDList :: DList a -> [a]

singleton :: a -> DList a

traverseConcurrently :: (PoolConf -> PoolConf) -> (a -> IO b) -> Jet a -> Jet b

Process the values yielded by the upstream Jet in a concurrent way, and return the results in the form of another Jet as they are produced.

NB: this function might scramble the order of the returned values. Right now there isn't a function for unscrambling them.

>>> :{
 J.each [(3,'a'), (2,'b'), (1,'c')]
 & J.traverseConcurrently (numberOfWorkers 10) (\(d,c) -> threadDelay (d*1e5) *> pure c)
 & J.toList
:}
"cba"

What happens if we limit the resulting Jet and we reach that limit, or if we otherwise stop consuming the Jet before it gets exhausted? In those cases, all pending IO b tasks are cancelled.

>>> :{
 J.each [(9999,'a'), (2,'b'), (1,'c')]
 & J.traverseConcurrently (numberOfWorkers 10) (\(d,c) -> threadDelay (d*1e5) *> pure c)
 & J.take 2
 & J.toList
:}
"cb"

data PoolConf

Configuration record for the worker pool.

Constructors

PoolConf

Fields

• _inputQueueSize :: Int
• _numberOfWorkers :: Int
• _outputQueueSize :: Int

Instances

Show PoolConf

Defined in Jet.Internal

Methods

showsPrec :: Int -> PoolConf -> ShowS

show :: PoolConf -> String

showList :: [PoolConf] -> ShowS

defaultPoolConf :: PoolConf

inputQueueSize :: Int -> PoolConf -> PoolConf

Size of the waiting queue into the worker pool. The default is 1.

numberOfWorkers :: Int -> PoolConf -> PoolConf

The size of the worker pool. The default is 1.

outputQueueSize :: Int -> PoolConf -> PoolConf

Size of the queue holding results out of the working pool before they are yielded downstream. The default is 1.

defaults :: a -> a

An alias for id. Useful with functions like traverseConcurrently and throughProcess, for which it means "use the default configuration".

throughProcess :: (ProcConf -> ProcConf) -> CreateProcess -> Jet ByteString -> Jet ByteString

Feeds the upstream Jet to an external process' stdin and returns the process' stdout as another Jet. The feeding and reading of the standard streams is done concurrently in order to avoid deadlocks.

What happens if we limit the resulting Jet and we reach that limit, or if we otherwise stop consuming the Jet before it gets exhausted? In those cases, the external process is promptly terminated.

linesThroughProcess :: (ProcConf -> ProcConf) -> CreateProcess -> Jet Line -> Jet Line

Like throughProcess, but feeding and reading Lines using the default system encoding.

>>> :{
J.each ["aaa","bbb","ccc"]
<&> J.stringToLine
& linesThroughProcess defaults (shell "cat")
& J.toList
:}
["aaa","bbb","ccc"]

An example of not reading all the lines from a long-lived process that gets cancelled:

>>> :{
mempty
& linesThroughProcess defaults (shell "{ printf \"aaa\\nbbb\\nccc\\n\" ; sleep infinity ; }")
& J.limit 2
& J.toList
:}
["aaa","bbb"]

utf8LinesThroughProcess :: (ProcConf -> ProcConf) -> CreateProcess -> Jet Line -> Jet Line

Like throughProcess, but feeding and reading Lines encoded in UTF8.

throughProcess_ :: ProcConf_ a b -> CreateProcess -> Jet a -> Jet b

type ProcConf = ProcConf_ ByteString ByteString

Configuration record with some extra options in addition to those in CreateProcess.

data ProcConf_ a b

Constructors

ProcConf_

Fields

• _bufferStdin :: Bool
• _writeToStdIn :: Handle -> a -> IO ()
• _readFromStdout :: Handle -> IO b
• _readFromStderr :: Handle -> IO ()
• _handleExitCode :: ExitCode -> IO ()

defaultProcConf :: ProcConf

bufferStdin :: Bool -> ProcConf -> ProcConf

Should we buffer the process' stdin? Usually should be True for interactive scenarios.

By default, False.

readFromStderr :: (Handle -> IO ()) -> ProcConf -> ProcConf

Sets the function that reads a single line of output from the process stderr. It's called repeatedly until stderr is exhausted. The reads are done concurrently with the reads from stdout.

By default, lines of text are read using the system's default encoding.

This is a good place to throw an exception if we don't like what comes out of stderr.

handleExitCode :: (ExitCode -> IO ()) -> ProcConf -> ProcConf

Sets the function that handles the final ExitCode of the process.

The default behavior is to throw the ExitCode as an exception if it's not a success.

data AreWeInsideGroup foldState

Constructors

OutsideGroup

InsideGroup !foldState

data RecastState foldState

Constructors

RecastState !(AreWeInsideGroup foldState) [IO foldState]

recast :: Splitter a b -> Combiners b c -> Jet a -> Jet c

This is a complex, unwieldly, yet versatile function. It can be used to define grouping operations, but also for decoding and other purposes.

Groups are delimited in the input Jet using the Splitter, and the contents of those groups are then combined using Combiners. The result of each combiner is yielded by the return Jet.

If the list of combiners is finite and becomes exhausted, we stop splitting and the return Jet stops.

data Combiners a b where

A Combiners value knows how to process a sequence of groups, while keeping a (existentially hidden) state for each group.

Very much like a FoldM IO from the foldl library, but "restartable" with a list of starting states.

For converting one into the other, this function should do the trick:

\(L.FoldM step allocator coda) -> combiners step coda (Prelude.repeat allocator)

Constructors

Combiners :: forall s a b. (s -> a -> IO s) -> (s -> IO b) -> [IO s] -> Combiners a b

Instances

Functor (Combiners a)

Defined in Jet.Internal

Methods

fmap :: (a0 -> b) -> Combiners a a0 -> Combiners a b

(<$) :: a0 -> Combiners a b -> Combiners a a0

combiners

Arguments

:: forall {k} s a b (r :: k). (s -> a -> IO s)

Step function that threads the state s.

-> (s -> IO b)

Coda invoked when a group closes.

-> [IO s]

Actions that produce the initial states s for processing each group.

-> Combiners a b

Constructor for Combiners values.

withCombiners_

Arguments

:: (h -> a -> IO ())

Step function that accesses the resource h.

-> (h -> IO ())

Finalizer to run after closing each group, and also in the case of an exception.

-> [IO h]

Actions that allocate a sequence of resources h.

-> (Combiners a () -> IO r)

The Combiners value should be consumed linearly.

-> IO r

A simpler version of withCombiners that doen't thread a state; it merely allocates and deallocates the resource h.

withCombiners

Arguments

:: (h -> s -> a -> IO s)

Step function that accesses the resource h and threads the state s.

-> (h -> s -> IO b)

Coda invoked when a group closes.

-> (h -> IO ())

Finalizer to run after each coda, and also in the case of an exception.

-> [(IO h, h -> IO s)]

Actions that allocate a sequence of resources h and produce initial states s for processing each group.

-> (Combiners a b -> IO r)

The Combiners value should be consumed linearly.

-> IO r

Combiners thread a state s while processing each group. Sometimes, in addition to that, we want to allocate a resource h when we start processing a group, and deallocate it after we finish processing the group or an exception is thrown. The typical example is allocating a Handle for writing the elements of the group as they arrive.

combineIntoLists :: Combiners a [a]

Puts the elements of each group into a list that is kept in memory. This breaks streaming within the group.

Useful with recast.

type Splitter a b = MealyIO a (SplitStepResult b)

Delimits groups in the values yielded by a Jet, and can also transform those values.

data MealyIO a b where

A Mealy machine with an existentially hidden state.

Very much like a FoldM IO from the foldl library, but it emits an output at each step, not only at the end.

Constructors

MealyIO

Fields

• :: forall s a b. (s -> a -> IO (b, s))

  The step function which threads the state.

• -> (s -> IO b)

  The final output, produced from the final state.

• -> IO s

  An action that produces the initial state.

• -> MealyIO a b

Instances

Functor (MealyIO a)

Defined in Jet.Internal

Methods

fmap :: (a0 -> b) -> MealyIO a a0 -> MealyIO a b

(<$) :: a0 -> MealyIO a b -> MealyIO a a0

data SplitStepResult b

For each value coming from upstream, what has the Splitter learned?

• Perhaps we should continue some group we have already started in a previous step.
• Perhaps we have found entire groups that we should emit in one go, groups we know are already complete.
• Perhaps we should start a new group that will continue in the next steps.

Constructors

SplitStepResult

Fields

• continuationOfPreviouslyStartedGroup :: [b]

  The continued group will be "closed" if in the current step we emit an entire group or we begin a new group.

  INVARIANT: we should only continue a group if we have already opened a "new one" with one or more elements in an earlier step.

• entireGroups :: [[b]]

  It's ok if the groups we find are empty.

• startOfNewGroup :: [b]

  INVARIANT: when we are in the final step, we should not yield elements for the beginning of a new one.

Instances

Functor SplitStepResult

Defined in Jet.Internal

Methods

fmap :: (a -> b) -> SplitStepResult a -> SplitStepResult b

(<$) :: a -> SplitStepResult b -> SplitStepResult a

Monoid (SplitStepResult b)

Defined in Jet.Internal

Methods

mempty :: SplitStepResult b

mappend :: SplitStepResult b -> SplitStepResult b -> SplitStepResult b

mconcat :: [SplitStepResult b] -> SplitStepResult b

Semigroup (SplitStepResult b)

Defined in Jet.Internal

Methods

(<>) :: SplitStepResult b -> SplitStepResult b -> SplitStepResult b

sconcat :: NonEmpty (SplitStepResult b) -> SplitStepResult b

stimes :: Integral b0 => b0 -> SplitStepResult b -> SplitStepResult b

Show b => Show (SplitStepResult b)

Defined in Jet.Internal

Methods

showsPrec :: Int -> SplitStepResult b -> ShowS

show :: SplitStepResult b -> String

showList :: [SplitStepResult b] -> ShowS