arrow/util/async_generator.h

64 using AsyncGenerator = std::function<Future<T>()>;  variable
67 struct IterationTraits<AsyncGenerator<T>> {
70   static AsyncGenerator<T> End() { return AsyncGenerator<T>(); }
72   static bool IsEnd(const AsyncGenerator<T>& val) { return !val; }
82 Future<> VisitAsyncGenerator(AsyncGenerator<T> generator, Visitor visitor) {
107     AsyncGenerator<T> generator;
116 Future<> DiscardAllFromAsyncGenerator(AsyncGenerator<T> generator) {
123 Future<std::vector<T>> CollectAsyncGenerator(AsyncGenerator<T> generator) {
138     AsyncGenerator<T> generator_;
148   MappingGenerator(AsyncGenerator<T> source, std::function<Future<V>(const T&)> map)
170     State(AsyncGenerator<T> source, std::function<Future<V>(const T&)> map)
188     AsyncGenerator<T> source;
271 AsyncGenerator<V> MakeMappedGenerator(AsyncGenerator<T> source_generator, MapFn map) {
294 AsyncGenerator<T> MakeFlatMappedGenerator(AsyncGenerator<T> source_generator, MapFn map) {
303   SequencingGenerator(AsyncGenerator<T> source, ComesAfter compare, IsNext is_next,
348     State(AsyncGenerator<T> source, ComesAfter compare, IsNext is_next, T initial_value)
357     AsyncGenerator<T> source;
440 AsyncGenerator<T> MakeSequencingGenerator(AsyncGenerator<T> source_generator,
456     TransformingGeneratorState(AsyncGenerator<T> generator, Transformer<T, V> transformer)
517     AsyncGenerator<T> generator_;
524   explicit TransformingGenerator(AsyncGenerator<T> generator,
546 AsyncGenerator<V> MakeTransformedGenerator(AsyncGenerator<T> generator,
555   SerialReadaheadGenerator(AsyncGenerator<T> source_generator, int max_readahead)
588     State(AsyncGenerator<T> source, int max_readahead)
619     AsyncGenerator<T> source_;
663   explicit FutureFirstGenerator(Future<AsyncGenerator<T>> future)
671       return state_->future_.Then([state](const AsyncGenerator<T>& source) {
680     explicit State(Future<AsyncGenerator<T>> future) : future_(future), source_() {}
682     Future<AsyncGenerator<T>> future_;
683     AsyncGenerator<T> source_;
696 AsyncGenerator<T> MakeFromFuture(Future<AsyncGenerator<T>> future) {
709 AsyncGenerator<T> MakeSerialReadaheadGenerator(AsyncGenerator<T> source_generator,
724 AsyncGenerator<T> MakeAutoStartingGenerator(AsyncGenerator<T> generator) {
736     AsyncGenerator<T> source;
747   ReadaheadGenerator(AsyncGenerator<T> source_generator, int max_readahead)
788     State(AsyncGenerator<T> source_generator, int max_readahead)
799     AsyncGenerator<T> source_generator;
961 AsyncGenerator<T> MakeReadaheadGenerator(AsyncGenerator<T> source_generator,
970 AsyncGenerator<T> MakeVectorGenerator(std::vector<T> vec) {
994   explicit MergedGenerator(AsyncGenerator<AsyncGenerator<T>> source,
1033     explicit DeliveredJob(AsyncGenerator<T> deliverer_, Result<T> value_,
1037     AsyncGenerator<T> deliverer;
1043     State(AsyncGenerator<AsyncGenerator<T>> source, int max_subscriptions)
1054     Future<AsyncGenerator<T>> PullSource() {
1061     AsyncGenerator<AsyncGenerator<T>> source;
1063     std::vector<AsyncGenerator<T>> active_subscriptions;
1122     void operator()(const Result<AsyncGenerator<T>>& maybe_next) {
1137                   AsyncGenerator<T>(), maybe_next.status(), index));
1183 AsyncGenerator<T> MakeMergedGenerator(AsyncGenerator<AsyncGenerator<T>> source,
1189 Result<AsyncGenerator<T>> MakeSequencedMergedGenerator(
1190     AsyncGenerator<AsyncGenerator<T>> source, int max_subscriptions) {
1197   AsyncGenerator<AsyncGenerator<T>> autostarting_source = MakeMappedGenerator(
1199       [](const AsyncGenerator<T>& sub) { return MakeAutoStartingGenerator(sub); });
1200   AsyncGenerator<AsyncGenerator<T>> sub_readahead =
1217 AsyncGenerator<T> MakeConcatenatedGenerator(AsyncGenerator<AsyncGenerator<T>> source) {
1238   EnumeratingGenerator(AsyncGenerator<T> source, T initial_value)
1259     State(AsyncGenerator<T> source, T initial_value)
1264     AsyncGenerator<T> source;
1288 AsyncGenerator<Enumerated<T>> MakeEnumeratedGenerator(AsyncGenerator<T> source) {
1290       source().Then([source](const T& initial_value) -> AsyncGenerator<Enumerated<T>> {
1299   explicit TransferringGenerator(AsyncGenerator<T> source, internal::Executor* executor)
1305   AsyncGenerator<T> source_;
1325 AsyncGenerator<T> MakeTransferredGenerator(AsyncGenerator<T> source,
1582 static Result<AsyncGenerator<T>> MakeBackgroundGenerator(
1595   explicit GeneratorIterator(AsyncGenerator<T> source) : source_(std::move(source)) {}
1600   AsyncGenerator<T> source_;
1606 Iterator<T> MakeGeneratorIterator(AsyncGenerator<T> source) {
1626   AsyncGenerator<T> owned_bg_generator = [io_executor, background_generator]() {
1661 AsyncGenerator<T> MakeFailingGenerator(Status st) {
1678 AsyncGenerator<T> MakeFailingGenerator(const Result<T>& result) {
1687 AsyncGenerator<T> MakeGeneratorStartsWith(std::vector<T> initial_values,
1688                                           AsyncGenerator<T> following_values) {
1690   auto gen_gen = MakeVectorGenerator<AsyncGenerator<T>>(
1704   AsyncGenerator<T> source;
1712 AsyncGenerator<T> MakeCancellable(AsyncGenerator<T> source, StopToken stop_token) {
1719   PauseableGenerator(AsyncGenerator<T> source, std::shared_ptr<util::AsyncToggle> toggle)
1728     PauseableGeneratorState(AsyncGenerator<T> source,
1740     AsyncGenerator<T> source_;
1755 AsyncGenerator<T> MakePauseable(AsyncGenerator<T> source,
1763   DefaultIfEmptyGenerator(AsyncGenerator<T> source, T or_value)
1787     AsyncGenerator<T> source;
1790     State(AsyncGenerator<T> source_, T or_value_)
1801 AsyncGenerator<T> MakeDefaultIfEmptyGenerator(AsyncGenerator<T> source, T or_value) {