stream.h

Go to the documentation of this file.

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
 
#pragma once
 
 
#include <cstddef>
#include <deque>
#include <memory>
#include <optional>
#include <type_traits>
#include <utility>
#include <vector>
 
#include "iceberg/result.h"
 
namespace iceberg {
 
template <typename T>
class Stream {
 public:
  virtual ~Stream() = default;
 
  Stream() = default;
 
  Stream(const Stream&) = delete;
 
  Stream& operator=(const Stream&) = delete;
 
  Stream(Stream&&) noexcept = default;
 
  Stream& operator=(Stream&&) noexcept = default;
 
  static std::unique_ptr<Stream<T>> FromVector(std::vector<T> values);
 
  Result<std::optional<T>> Next() {
    if (error_.has_value()) {
      return unexpected(*error_);
    }
    if (finished_) {
      return std::nullopt;
    }
 
    auto result = NextImpl();
    if (!result.has_value()) {
      error_ = result.error();
    } else if (!result.value().has_value()) {
      finished_ = true;
    }
    return result;
  }
 
  Result<std::vector<T>> ToVector() {
    auto collect = [this](auto& values, auto append,
                          auto finish) -> Result<std::vector<T>> {
      while (true) {
        auto result = Next();
        if (!result.has_value()) {
          return unexpected(std::move(result.error()));
        }
        auto& value = result.value();
        if (!value.has_value()) {
          return finish(values);
        }
        append(values, value.value());
      }
    };
 
    if constexpr (!std::is_move_constructible_v<T>) {
      static_assert(std::is_copy_constructible_v<T>,
                    "Stream::ToVector requires T to be move- or copy-constructible");
 
      // For strictly copy-only T, collecting directly into a vector can repeatedly copy
      // previously collected elements during vector growth. Stage values in a deque,
      // then copy once into an exactly sized vector.
      std::deque<T> values;
      return collect(
          values, [](auto& destination, const T& value) { destination.push_back(value); },
          [](const auto& source) {
            return std::vector<T>(source.cbegin(), source.cend());
          });
    } else {
      std::vector<T> values;
      return collect(
          values,
          [](auto& destination, T& value) {
            destination.push_back(std::move_if_noexcept(value));
          },
          [](auto& source) { return std::move(source); });
    }
  }
 
 protected:
  virtual Result<std::optional<T>> NextImpl() = 0;
 
 private:
  bool finished_ = false;
  std::optional<Error> error_;
};
 
template <typename T>
std::unique_ptr<Stream<T>> Stream<T>::FromVector(std::vector<T> values) {
  class VectorStream final : public Stream<T> {
   public:
    explicit VectorStream(std::vector<T> values) : values_(std::move(values)) {}
 
   protected:
    Result<std::optional<T>> NextImpl() override {
      if (index_ == values_.size()) {
        return std::nullopt;
      }
      auto&& value = values_[index_++];
      return Result<std::optional<T>>(std::in_place, std::in_place,
                                      std::move_if_noexcept(value));
    }
 
   private:
    std::vector<T> values_;
    size_t index_ = 0;
  };
  return std::make_unique<VectorStream>(std::move(values));
}
 
}  // namespace iceberg