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 {
    public:
      iterator(::stream::iterator<C>&& f, ::stream::iterator<C>&& l)
      : start_(std::forward<::stream::iterator<C>>(f))
      , finish_(std::forward<::stream::iterator<C>>(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;
      }
      
      ~iterator() {}
      typename C::value_type 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();
        }
      }
      ::stream::iterator<C> start_, finish_;
      C mem_;
      typename C::const_iterator curr_, end_;
    };
  }
  
  template <typename C>
  class join_stream : public stream_impl<typename C::value_type> {
  public:
    using T = typename C::value_type;
    explicit join_stream(stream_base<C> const& sb) : source_(sb) {}
    ~join_stream() override {}
    
    ::stream::iterator<T> begin() override {
      return {join::iterator<C>{source_.begin(), source_.end()}};
    }
    ::stream::iterator<T> end() override {
      return {join::iterator<C>{source_.end(), source_.end()}};
    }
  private:
    stream_base<C> source_;
  };
  
  template <typename C>
  auto make_join(stream_base<C> const& sb) -> stream_base<typename C::value_type>  {
    using T = typename C::value_type;
    using impl_t = join_stream<C>;
    return {std::make_shared<impl_t>(sb)};
  }
  
  template <typename T>
  template <typename F>
  auto stream_base<T>::flatmap(F&& func) const -> stream_base<flatmap_f<F>>  {
    return make_join((*this).map(func));
  }
} }