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.

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.

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

Equivalent to run (const False).

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

Build a Jet from any Foldable container

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

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

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

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

>>> 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.

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

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

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

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

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

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

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.

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.

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"

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

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

Synonym for take.

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

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

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

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]

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

>>> 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)]

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
[(),()]

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.

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

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
:}
""

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

"morally", all control operations compatible with this library should execute the callback only once, which means that they should have a linear type. But because linear types are not widespread, they usually are given a less precise non-linear type. If you know what you are doing, use this function to give them a linear type.

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

Line unsafeCoerceControl, for when the callback doesn't use the allocated resource.

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

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

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

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

Length in bytes.

THROWS:

• UnicodeException

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.

Data.Text.singleton '\n'

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

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

Adds the Text to the beginning of the Line.

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"

Configuration record for the worker pool.

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

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

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

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

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.

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"]

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

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

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

By default, False.

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.

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.

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

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

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

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

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

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

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.

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

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.

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.

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.

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.

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)

Constructor for Combiners values.

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.

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

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

Useful with recast.