arrow-fx-coroutines / arrow.fx.coroutines.stream
Package arrow.fx.coroutines.stream
Types
| Chunk | Strict, finite sequence of values that allows index-based random access of elements.abstract class ~~Chunk~~<out O> |
| EmitterSyntax | interface EmitterSyntax<A> |
| ForStream | class ForStream |
| Pipe | A stream transformation can be represented as a function from stream to stream. This means that Pipe is also an alias for fun Stream<I>.name(): Stream<O>.typealias ~~Pipe~~<I, O> = (Stream<I>) -> Stream<O> |
| Pipe2 | A stream transformation that combines two streams in to a single stream, represented as a function from two streams to a single stream.typealias ~~Pipe2~~<I, I2, O> = (Stream<I>, Stream<I2>) -> Stream<O> |
| Pull | sealed class ~~Pull~~<out O, out R> |
| PullUncons | data class PullUncons<O> |
| PullUncons1 | data class PullUncons1<O> |
| ResourceOps | DSL boundary to access terminal operators as a Resource Allows for consume a Stream as a Resource, meaning the root scope of the Stream remains open until Resource.use returns.class ~~ResourceOps~~<O> |
| Scope | Implementation of Scope for the internal stream interpreter.class ~~Scope~~ |
| StepLeg | When merging multiple streams, this represents step of one leg.class StepLeg<O> |
| Stream | class ~~Stream~~<out O> : StreamOf<O> |
| StreamOf | typealias StreamOf<A> = Kind<ForStream, A> |
| ZipWithCont | typealias ZipWithCont<I, O> = (Either<Pair<Chunk<I>, Pull<I, Unit>>, Pull<I, Unit>>) -> Pull<O, Unit> |
Extensions for External Classes
| arrow.Kind | |
| kotlin.Function1 |
Functions
| append | fun <O, R> Pull<O, R>.append(post: () -> Pull<O, R>): Pull<O, R>Lazily appends s2 to the end of this stream. fun <O> Stream<O>.append(s2: () -> Stream<O>): Stream<O> |
| asResource | Opens DSL to consume Stream as a Resource.fun <O> Stream<O>.asResource(): ResourceOps<O> |
| cons | fun <O, B> Pull<O, B>.cons(c: Chunk<O>): Pull<O, B>fun <O, B> Pull<O, B>.cons(o: O): Pull<O, B>Prepends a Chunk onto the front of this stream. fun <O> Stream<O>.cons(c: Chunk<O>): Stream<O> |
| cons1 | Prepends a value onto the front of this stream.fun <O> Stream<O>.cons1(o: O): Stream<O> |
| copyToArray | fun <O> Chunk<O>.copyToArray(xs: Array<O>, start: Int = 0): Unitfun Chunk<Byte>.copyToArray(xs: ByteArray, start: Int = 0): Unitfun Chunk<Long>.copyToArray(xs: LongArray, start: Int = 0): Unitfun Chunk<Boolean>.copyToArray(xs: BooleanArray, start: Int = 0): Unitfun Chunk<Short>.copyToArray(xs: ShortArray, start: Int = 0): Unitfun Chunk<Int>.copyToArray(xs: IntArray, start: Int = 0): Unitfun Chunk<Float>.copyToArray(xs: FloatArray, start: Int = 0): Unitfun Chunk<Double>.copyToArray(xs: DoubleArray, start: Int = 0): Unit |
| drain | Runs all the effects of this Stream and ignores all emitted values.suspend fun <O> Stream<O>.drain(): Unit |
| drop | Drops the first n elements read by this Pull and returnsfun <O> Pull<O, Unit>.drop(n: Long): Pull<O, Unit> |
| dropThrough | Like dropWhile, but drops the first value which tests false.fun <O> Pull<O, Unit>.dropThrough(p: (O) -> Boolean): Pull<O, Unit> |
| dropWhile | Drops elements of the this stream until the predicate p fails, and returns the new stream. If defined, the first element of the returned stream will fail p.fun <O> Pull<O, Unit>.dropWhile(p: (O) -> Boolean): Pull<O, Unit> |
| dropWhile_ | fun <O> Pull<O, Unit>.dropWhile_(p: (O) -> Boolean, dropFailure: Boolean): Pull<O, Pull<O, Unit>> |
| echo1 | Reads a single element from the input and emits it to the output.fun <O> Pull<O, Unit>.echo1(): Pull<O, Pull<O, Unit>?> |
| echoChunk | Reads the next available chunk from the input and emits it to the output.fun <O> Pull<O, Unit>.echoChunk(): Pull<O, Pull<O, Unit>?> |
| emptyStream | fun <O> emptyStream(): Stream<O> |
| fetchN | Like unconsN, but leaves the buffered input unconsumed.fun <O> Pull<O, Unit>.fetchN(n: Int): Pull<O, Pull<O, Unit>?> |
| filterNotNull | Filters any null.fun <O : Any> Stream<O?>.filterNotNull(): Stream<O> |
| filterOption | Filters any arrow.core.None.fun <O> Stream<Option<O>>.filterOption(): Stream<O> |
| firstOrError | Runs the first effect of this Stream, raising a NoSuchElementException if the stream emitted no values and returns the value if emitted.suspend fun <O> Stream<O>.firstOrError(): O |
| firstOrNull | Awaits the next available element where the predicate returns true.fun <O> Pull<O, Unit>.firstOrNull(f: (O) -> Boolean): Pull<Nothing, PullUncons1<O>?>Runs the first effect of this Stream, and returns null if the stream emitted a value and returns the value if emitted.suspend fun <O> Stream<O>.firstOrNull(): O? |
| flatMap | fun <O, O2 : O, R, R2> Pull<O, R>.flatMap(f: (R) -> Pull<O2, R2>): Pull<O2, R2> |
| flatMapOutput | fun <O, O2> Pull<O, Unit>.flatMapOutput(f: (O) -> Pull<O2, Unit>): Pull<O2, Unit> |
| flatten | fun <O, R> Pull<O, Pull<O, R>>.flatten(): Pull<O, R>Flattens a stream of streams in to a single stream by concatenating each stream. fun <O> Stream<Stream<O>>.flatten(): Stream<O> |
| fold | Folds all inputs using an initial value z and supplied binary operator, and writes the final result to the output of the supplied Pull when the stream has no more values.fun <O, O2> Pull<O, Unit>.fold(initial: O2, f: (O2, O) -> O2): Pull<Nothing, O2> |
| fold1 | Folds all inputs using the supplied binary operator, and writes the final result to the output of the supplied Pull when the stream has no more values.fun <O> Pull<O, Unit>.fold1(f: (O, O) -> O): Pull<Nothing, O?>Folds all inputs using the supplied operator f, and emits a single-element stream, or the empty stream if the input is empty, or the never stream if the input is non-terminating. fun <O> Stream<O>.fold1(f: (O, O) -> O): Stream<O> |
| foldChunks | Folds all the effects of this stream in to a value by folding the output chunks together, starting with the provided init and combining the current value with each output chunk using fsuspend fun <O, B> Stream<O>.foldChunks(init: B, f: (B, Chunk<O>) -> B): B |
| foldMonoid | Folds this stream with the monoid for O.fun <O> Stream<O>.foldMonoid(MO: Monoid<O>): Stream<O> |
| forall | Writes a single true value if all input matches the predicate, false otherwise.fun <O> Pull<O, Unit>.forall(p: (O) -> Boolean): Pull<Nothing, Boolean> |
| handleErrorWith | fun <O, R> Pull<O, R>.handleErrorWith(f: (Throwable) -> Pull<O, R>): Pull<O, R>If this Stream terminates with Stream.raiseError, invoke [h](handle-error-with.html#arrow.fx.coroutines.stream$handleErrorWith(arrow.fx.coroutines.stream.Stream((arrow.fx.coroutines.stream.handleErrorWith.O)), kotlin.Function1((kotlin.Throwable, arrow.fx.coroutines.stream.Stream((arrow.fx.coroutines.stream.handleErrorWith.O)))))/h) and continue with result.fun |
| interleave | Deterministically interleaves elements, starting on the left, terminating when the end of either branch is reached naturally.fun <O> Stream<O>.interleave(that: Stream<O>): Stream<O> |
| interleaveAll | Deterministically interleaves elements, starting on the left, terminating when the ends of both branches are reached naturally.fun <O> Stream<O>.interleaveAll(that: Stream<O>): Stream<O> |
| interruptScope | Creates a scope that may be interrupted by calling scope#interrupt.fun <O> Pull<O, Unit>.interruptScope(): Pull<O, Unit> |
| intersperse | fun <A> Chunk<A>.intersperse(separator: A): Chunk<A>Emits the specified separator between every pair of elements in the source stream. fun <O> Stream<O>.intersperse(separator: O): Stream<O> |
| lastOrError | Runs all the effects of this Stream, raising a NoSuchElementException if the stream emitted no values, and returning the last value emitted otherwise.suspend fun <O> Stream<O>.lastOrError(): O |
| lastOrNull | Runs all the effects of this Stream, and returns null if the stream emitted no values and returning the last value emitted if values were emitted.suspend fun <O> Stream<O>.lastOrNull(): O? |
| map | fun <O, R, R2> Pull<O, R>.map(f: (R) -> R2): Pull<O, R2> |
| noneTerminate | Wraps the inner values in Option as Some and adds an None at the end of the Stream to signal completion. Note that this doesn’t actually limit an infinite stream.fun <O> Stream<O>.noneTerminate(): Stream<Option<O>> |
| onComplete | Run s2 after this, regardless of errors during this, then reraise any errors encountered during this.fun <O> Stream<O>.onComplete(s2: () -> Stream<O>): Stream<O> |
| parJoin | Non-deterministically merges a stream of streams (outer) in to a single stream, opening at most maxOpen streams at any point in time.fun <O> Stream<Stream<O>>.~~parJoin~~(maxOpen: Int, ctx: CoroutineContext = Dispatchers.Default): Stream<O> |
| parJoinUnbounded | Like parJoin but races all inner streams simultaneously without limit.fun <O> Stream<Stream<O>>.parJoinUnbounded(ctx: CoroutineContext = Dispatchers.Default): Stream<O> |
| peek | Like uncons but does not consume the chunk (i.e., the chunk is pushed back).fun <O> Pull<O, Unit>.peek(): Pull<Nothing, PullUncons<O>?> |
| peek1 | Like uncons1 but does not consume the element (i.e., the element is pushed back).fun <O> Pull<O, Unit>.peek1(): Pull<Nothing, PullUncons1<O>?> |
| Pipe | fun <I, O> ~~Pipe~~(pipe: (Stream<I>) -> Stream<O>): Pipe<I, O> |
| Pipe2 | fun <I, I2, O> ~~Pipe2~~(pipe: (Stream<I>, Stream<I2>) -> Stream<O>): Pipe2<I, I2, O> |
| prependTo | Prepends a chunk to the start of this chunk queue.infix fun <A> Chunk<A>.prependTo(q: Queue<A>): Queue<A>Prepends a Chunk onto the front of this stream. infix fun <O> Chunk<O>.prependTo(s: Stream<O>): Stream<O>Prepends a value onto the front of this stream. infix fun <O> O.prependTo(s: Stream<O>): Stream<O>infix fun <A> A.prependTo(fa: Iterable<A>): List<A> |
| reduce | fun <O> Stream<O>.reduce(f: (O, O) -> O): Stream<O> |
| reduceSemigroup | Reduces this stream with the Semigroup for O.fun <O> Stream<O>.reduceSemigroup(S: Semigroup<O>): Stream<O> |
| repartition | Repartitions the input with the function f. On each step f is applied to the input and all elements but the last of the resulting sequence are emitted. The last element is then appended to the next input using the Semigroup S.fun <O> Stream<O>.repartition(S: Semigroup<O>, f: (O) -> Chunk<O>): Stream<O> |
| repeat | Repeatedly invokes using, running the resultant Pull each time, halting when a pull returns null instead of nextStream.fun <O, B> Pull<O, Unit>.repeat(using: (Pull<O, Unit>) -> Pull<B, Pull<O, Unit>?>): Pull<B, Unit> |
| scan | Left fold which outputs all intermediate results.fun <O, O2> Stream<O>.scan(init: O2, f: (O2, O) -> O2): Stream<O2> |
| scan1 | Like [scan], but uses the first element of the stream as the seed.fun <O> Stream<O>.scan1(f: (O, O) -> O): Stream<O> |
| scanChunks | Like scan but f is applied to each chunk of the source stream. The resulting chunk is emitted and the result of the chunk is used in the next invocation of f. The final state value is returned as the result of the pull.fun <O, S, O2> Pull<O, Unit>.scanChunks(init: S, f: (S, Chunk<O>) -> Pair<S, Chunk<O2>>): Pull<O2, S>Like scan but f is applied to each chunk of the source stream. The resulting chunk is emitted and the result of the chunk is used in the next invocation of f.fun <O, S, O2> Stream<O>.scanChunks(init: S, f: (S, Chunk<O>) -> Pair<S, Chunk<O2>>): Stream<O2> |
| scanChunksOpt | More general version of scanChunks where the current state (i.e., S) can be inspected to determine if another chunk should be pulled or if the pull should terminate. Termination is signaled by returning None from f. Otherwise, a function which consumes the next chunk is returned wrapped in Some. The final state value is returned as the result of the pull.fun <O, S, O2> Pull<O, Unit>.scanChunksOpt(init: S, f: (S) -> ((Chunk<O>) -> Pair<S, Chunk<O2>>)?): Pull<O2, S>More general version of scanChunks where the current state (i.e., S) can be inspected to determine if another chunk should be pulled or if the stream should terminate. Termination is signaled by returning null from f. Otherwise, a function which consumes the next chunk.fun <O, S, O2> Stream<O>.scanChunksOpt(init: S, f: (S) -> ((Chunk<O>) -> Pair<S, Chunk<O2>>)?): Stream<O2> |
| scanMap | Alias for map(f).scanMonoid.fun <O, O2> Stream<O>.scanMap(MO2: Monoid<O2>, f: (O) -> O2): Stream<O2> |
| scanMonoid | Folds this stream with the monoid for O while emitting all intermediate results.fun <O> Stream<O>.scanMonoid(MO: Monoid<O>): Stream<O> |
| scope | fun <O> Pull<O, Unit>.scope(): Pull<O, Unit> |
| stepLeg | Like uncons, but instead of performing normal uncons, this will run the stream up to the first chunk available. Useful when zipping multiple streams (legs) into one stream. Assures that scopes are correctly held for each stream leg independently of scopes from other legs.fun <O> Pull<O, Unit>.stepLeg(): Pull<Nothing, StepLeg<O>?> |
| stop | suspend fun <O> stop(done: SignallingAtomic<Option<Option<Throwable>>>, outputQ: NoneTerminatedQueue<Chunk<O>>, rslt: Option<Throwable>): Unit |
| stream | Interpret this Pull to produce a Stream, introducing a scope.fun <O> Pull<O, Unit>.stream(): Stream<O> |
| take | Emits the first n elements of the input.fun <O> Pull<O, Unit>.take(n: Long): Pull<O, Pull<O, Unit>> |
| takeLast | Emits the last n elements of the input.fun <O> Pull<O, Unit>.takeLast(n: Int): Pull<Nothing, Queue<O>> |
| takeThrough | fun <O> Pull<O, Unit>.takeThrough(p: (O) -> Boolean): Pull<O, Pull<O, Unit>> |
| takeWhile | fun <O> Pull<O, Unit>.takeWhile(p: (O) -> Boolean): Pull<O, Pull<O, Unit>> |
| takeWhile_ | fun <O> Pull<O, Unit>.takeWhile_(p: (O) -> Boolean, takeFailure: Boolean): Pull<O, Pull<O, Unit>> |
| terminateOn | Halts the input stream when the condition is true.fun <O> Stream<O>.terminateOn(terminator: (O) -> Boolean): Stream<O> |
| terminateOnNone | Halts the input stream at the first arrow.core.None.fun <O> Stream<Option<O>>.terminateOnNone(): Stream<O> |
| terminateOnNull | Halts the input stream at the first null.fun <O> Stream<O>.terminateOnNull(): Stream<O> |
| toArray | fun <O> Chunk<O>.toArray(): Array<O> |
| toList | Runs all the effects of this Stream and collects all emitted values into a List. If the Stream doesn’t emit any values it returns emptyList.suspend fun <O> Stream<O>.toList(): List<O> |
| toSet | Runs all the effects of this Stream and collects all emitted values into a Set. If the Stream doesn’t emit any values it returns emptySet.suspend fun <O> Stream<O>.toSet(): Set<O> |
| transformWith | fun <O, R, R2> Pull<O, R>.transformWith(f: (Result<R>) -> Pull<O, R2>): Pull<O, R2> |
| uncons1OrNull | fun <O> Pull<O, Unit>.uncons1OrNull(): Pull<Nothing, PullUncons1<O>?> |
| unconsLimitOrNull | Like unconsOrNull, but returns a chunk of no more than n elements.fun <O> Pull<O, Unit>.unconsLimitOrNull(n: Int): Pull<Nothing, PullUncons<O>?> |
| unconsN | Like uncons, but returns a chunk of exactly n elements, splitting chunk as necessary.fun <O> Pull<O, Unit>.unconsN(n: Int, allowFewer: Boolean = false): Pull<Nothing, PullUncons<O>?> |
| unconsNonEmptyOrNull | Like unconsOrNull but skips over empty chunks, pulling until it can emit the first non-empty chunk.fun <O> Pull<O, Unit>.unconsNonEmptyOrNull(): Pull<Nothing, PullUncons<O>?> |
| unconsOrNull | fun <O> Pull<O, Unit>.unconsOrNull(): Pull<Nothing, PullUncons<O>?> |
| zipAll | Deterministically zips elements, terminating when the ends of both branches are reached naturally, padding the left branch with pad1 and padding the right branch with pad2 as necessary.fun <O, B> Stream<O>.zipAll(that: Stream<B>, pad1: O, pad2: B): Stream<Pair<O, B>> |
| zipAllWith | Deterministically zips elements with the specified function, terminating when the ends of both branches are reached naturally, padding the left branch with pad1 and padding the right branch with pad2 as necessary.fun <A, B, C> Stream<A>.zipAllWith(that: Stream<B>, pad1: A, pad2: B, f: (A, B) -> C): Stream<C> |
Companion Object Functions
| callback | Creates a Stream from the given suspended block callback, allowing to emit, set cancel effects and end the emission.fun <A> Stream.Companion.callback(f: suspend EmitterSyntax<A>.() -> Unit): Stream<A> |
| cancellable | Creates a cancellable Stream from the given suspended block that will evaluate the passed CancelToken if cancelled.fun <A> Stream.Companion.cancellable(f: suspend EmitterSyntax<A>.() -> CancelToken): Stream<A> |
| monoid | Monoid instance for Stream.fun <O> Stream.Companion.monoid(): Monoid<Stream<O>> |