cpp-lazy-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));
  }
} }