cpp-lazy-stream/include/stream/streams/join.hpp

#pragma once

#include <iterator>

namespace stream { namespace detail {
#define JOIN_CTOR(mod)                                                         \
  start_ = mod(other.start_);                                                  \
  finish_ = mod(other.finish_);                                                \
  std::ptrdiff_t n = std::distance(other.mem_.cbegin(), other.curr_);          \
  mem_ = mod(other.mem_);                                                      \
  curr_ = mem_.begin();                                                        \
  std::advance(curr_, n);                                                      \
  end_ = mem_.end();

  template <typename C> class join_iterator {
  public:
    using reference = typename C::const_iterator::reference;

  public:
    join_iterator(iterator<C> && f, iterator<C> && l)
        : start_(std::move(f)), finish_(std::move(l)) {
      if (start_ != finish_) {
        mem_ = *(start_);
        curr_ = mem_.begin();
        end_ = mem_.end();
        advance();
      }
    }

    join_iterator(join_iterator const & other) {
      JOIN_CTOR()
    }
    
    join_iterator(join_iterator && other){
      JOIN_CTOR(std::move)
    }

    join_iterator & operator=(join_iterator const & other) {
      JOIN_CTOR()
      return *this;
    }

    join_iterator & operator=(join_iterator && other) {
      JOIN_CTOR(std::move)
      return *this;
    }

    reference operator*() { return *curr_; }

    join_iterator & operator++() {
      ++curr_;
      advance();
      return *this;
    }

    bool operator==(join_iterator const & rhs) const { return start_ == rhs.start_; }
  private:
    void advance() {
      while (curr_ == end_ && start_ != finish_) {
        if (++start_ == finish_) { break; }
        mem_ = *start_;
        curr_ = mem_.begin();
        end_ = mem_.end();
      }
    }
    iterator<C> start_;
    iterator<C> finish_;
    C mem_;
    typename C::const_iterator curr_;
    typename C::const_iterator end_;
  };

  template <typename C> class join_stream {
  public:
    using T = typename C::value_type;
    explicit join_stream(stream_base<C> const & sb) : source_(sb) {}

    iterator<T> begin() {
      return {join_iterator<C>{source_.begin(), source_.end()}};
    }
    iterator<T> end() {
      return {join_iterator<C>{source_.end(), source_.end()}};
    }

  private:
    stream_base<C> source_;
  };

  template <typename C>
  stream_base<typename C::value_type> make_join(stream_base<C> const & sb) {
    return std::make_shared<join_stream<C>>(sb);
  }

  template <typename T>
  template <typename F>
  stream_base<traits::fmapped_t<T, F>> stream_base<T>::flatmap(F && func) && {
    return make_join(std::move(*this).map(func));
  }

  template <typename T>
  template <typename F>
  stream_base<traits::fmapped_t<T, F>>
  stream_base<T>::flatmap(F && func) const & {
    return make_join(map(func));
  }
}}