Lazy emaluation based on stream

stream_test.cpp

#pragma warning(disable : 4251 4275)
#include "gtest/gtest.h"

#include <functional>
#include <initializer_list>
#include <memory>
#include <optional>
#include <vector>

/**
 * Эскиз организации ленивых вычислений для реляционной алгебры
 */
namespace {
using T = int;
/**
 * @brief The iterator struct
 */
struct iterator {
  virtual ~iterator() = default;
  virtual bool has_next() const = 0;
  virtual T next() = 0;
};
/**
 * @brief The stream struct
 * @warning поток ленивый. только терминальные операторы запускают реальный
 * процесс вычисления
 */
struct stream : std::enable_shared_from_this<stream> {
public:
  stream() = default;
  virtual ~stream() = default;

public: // геттеры
  virtual std::unique_ptr<iterator> get_iterator() = 0;

public: // операторы
  /**
   * @brief of порождает поток из элементов списка инициализации
   * @param lst список инициализации
   * @warning ленивый оператор (никаких вычислений не запускает)
   * @return новый поток
   */
  static std::shared_ptr<stream> of(std::initializer_list<T> lst) {

    // реализация потока
    struct of_stream : stream {
      std::vector<T> sequence;

      explicit of_stream(std::vector<T> v) : sequence{std::move(v)} {}

      virtual std::unique_ptr<iterator> get_iterator() override {
        // реализация итератора
        struct of_stream_iterator : iterator {
          std::shared_ptr<of_stream> owner;
          typename std::vector<T>::iterator it;
          explicit of_stream_iterator(std::shared_ptr<of_stream> owner)
              : owner{owner}, it{owner->sequence.begin()} {}
          virtual bool has_next() const override {
            return it != owner->sequence.end();
          }
          virtual T next() override { return *it++; }
        };
        return std::make_unique<of_stream_iterator>(
            std::static_pointer_cast<of_stream>(shared_from_this()));
      }
    };

    return std::make_shared<of_stream>(lst);
  }

  /**
   * @brief map возвращает новый поток где каждый элемент получен применением
   * функции f к элементу исходного потока
   * @param f функция
   * @warning ленивый оператор (никаких вычислений не запускает)
   * @return новый поток
   */
  std::shared_ptr<stream> map(std::function<T(T)> f) {

    // реализация потока
    struct map_stream : stream {
      std::shared_ptr<stream> base;
      std::function<T(T)> f;

      explicit map_stream(std::shared_ptr<stream> base, std::function<T(T)> f)
          : base(std::move(base)), f(std::move(f)) {}

      virtual std::unique_ptr<iterator> get_iterator() override {

        // реализация итератора
        struct map_stream_iterator : iterator {
          std::shared_ptr<map_stream> owner;
          std::unique_ptr<iterator> it;
          explicit map_stream_iterator(std::shared_ptr<map_stream> owner,
                                       std::unique_ptr<iterator> it)
              : owner(std::move(owner)), it(std::move(it)) {}
          virtual bool has_next() const override { return it->has_next(); }
          virtual T next() override { return owner->f(it->next()); }
        };

        return std::make_unique<map_stream_iterator>(
            std::static_pointer_cast<map_stream>(shared_from_this()),
            base->get_iterator());
      }
    };

    return std::make_shared<map_stream>(shared_from_this(), f);
  }

  /**
   * @brief filter возращает новый поток с отфильтрованными предикатом p
   * элементами исходного потока
   * @param p предикат
   * @warning ленивый оператор (никаких вычислений не запускает)
   * @return новый отфильтрованный предикатом p поток
   */
  std::shared_ptr<stream> filter(std::function<bool(T)> p) {

    // реализация потока
    struct filter_stream : stream {
      std::shared_ptr<stream> base;
      std::function<bool(T)> p;

      explicit filter_stream(std::shared_ptr<stream> base,
                             std::function<bool(T)> p)
          : base(std::move(base)), p(std::move(p)) {}

      virtual std::unique_ptr<iterator> get_iterator() override {

        // реализация итератора
        struct filter_stream_iterator : iterator {
          std::shared_ptr<filter_stream> owner;
          std::unique_ptr<iterator> it;
          mutable T value;
          filter_stream_iterator(std::shared_ptr<filter_stream> owner,
                                 std::unique_ptr<iterator> it)
              : owner(std::move(owner)), it(std::move(it)) {}

          virtual bool has_next() const override {
            while (it->has_next()) {
              if (owner->p(value = it->next())) {
                return true;
              }
            }
            return false;
          }
          virtual T next() override { return value; }
        };

        return std::make_unique<filter_stream_iterator>(
            std::static_pointer_cast<filter_stream>(shared_from_this()),
            base->get_iterator());
      }
    };

    return std::make_shared<filter_stream>(shared_from_this(), p);
  }

  /**
   * @brief count возращает число элементов потока
   * @warning count терминальный оператор (запускает процесс вычисления)
   * @return число элементов потока
   */
  std::optional<size_t> count() {
    auto it = get_iterator();
    auto has_next = it->has_next();
    if (has_next) {
      size_t c = 0;
      do {
        it->next();
        ++c;
      } while (has_next = it->has_next());
      return c;
    }
    return std::optional<size_t>{};
  }

  /**
   * @brief collect терминальный оператор собирающий элементы исходного потока в
   * коллекцию
   * @return коллекция элементов исходного потока
   */
  std::vector<T> collect() {
    std::vector<T> v;
    for (auto it = get_iterator(); it->has_next();) {
      v.emplace_back(it->next());
    }
    return v;
  }
};
} // namespace

TEST(stream, iterator) {
  std::unique_ptr<iterator> si;
  {
    auto s = stream::of({1, 2, 3});
    si = s->get_iterator();
  }
  // итератор продлевает время жизни потоку, поэтому следующие операции
  // безопасны
  ASSERT_EQ(1, si->next());
  ASSERT_EQ(2, si->next());
  ASSERT_EQ(3, si->next());
}

TEST(stream, count) {
  auto s = stream::of({1, 2, 3});
  ASSERT_EQ(true, static_cast<bool>(s->count()));
  ASSERT_EQ(true, static_cast<bool>(s->count()));
  ASSERT_EQ(3, *s->count());
  ASSERT_EQ(3, *s->count());
}

TEST(stream, map) {
  auto s1 = stream::of({1, 2, 3});
  auto s2 = s1->map([](auto e) { return e + e; });
  ASSERT_EQ(3, *s1->count());
  ASSERT_EQ(3, *s2->count());
  auto s2i = s2->get_iterator();
  ASSERT_EQ(2, s2i->next());
  ASSERT_EQ(4, s2i->next());
  ASSERT_EQ(6, s2i->next());
}

TEST(stream, filter) {
  auto s1 = stream::of({1, 2, 3});
  auto s2 = s1->filter([](auto e) { return e > 2; });
  ASSERT_EQ(3, *s1->count());
  ASSERT_EQ(1, *s2->count());
  auto s2i = s2->get_iterator();
  ASSERT_EQ(true, s2i->has_next());
  ASSERT_EQ(3, s2i->next());
}

TEST(stream, collect) {
  auto s1 = stream::of({1, 2, 3});
  auto s2 = s1->filter([](auto e) { return e > 1; });
  auto v = s2->collect();
  ASSERT_EQ(std::vector<int>({2, 3}), v);
}