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();

  namespace join {
    template <typename C> class iterator {
    private:
      using impl_t = ::stream::iterator<C>;
      using local_iterator = typename C::const_iterator;

    public:
      using reference = typename local_iterator::reference;

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

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

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

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

      reference operator*() { return *curr_; }

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

      DELEGATE_ITERATOR_IMPL_BASE(start_)
    private:
      void advance() {
        while (curr_ == end_ && start_ != finish_) {
          if (++start_ == finish_) { break; }
          mem_ = *start_;
          curr_ = mem_.begin();
          end_ = mem_.end();
        }
      }
      impl_t start_, finish_;
      C mem_;
      local_iterator curr_, end_;
    };
  }

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

    ::stream::iterator<T> begin() {
      return {join::iterator<C>{source_.begin(), source_.end()}};
    }
    ::stream::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));
  }
}}