cpp-lazy-stream/streams/join.hpp

#pragma once

template <typename T, typename C, bool B>
stream::detail::stream_base<typename C::value_type, true> operator|(stream::detail::stream_base<T, B> const&s, stream::flatmap_t<T, C>&& f) {
  return s.flatmap(f.func);
}
template <typename T, typename C, bool B>
stream::detail::stream_base<typename C::value_type, true> operator|(stream::detail::stream_base<T, B> &&s, stream::flatmap_t<T, C>&& f) {
  return std::move(s).flatmap(f.func);
}

namespace stream {
  template <typename T, typename R>
  struct flatmap_t {
    template <typename F>
    explicit flatmap_t(F&& f) : func(f) {}
    std::function<R(const T&)> func;
  };
}

namespace stream {
  namespace detail {
    template <typename C, bool B>
    class join_stream : public stream_impl<typename C::value_type> {
    public:
      using T = typename C::value_type;
      class iterator : public iterator_impl<T> {
      public:
        typedef iterator_impl<T> super;
        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() {}
        T operator*() override { return *curr_; }
        
        super& operator++() override {
          ++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::iterator curr_, end_;
      };
      
      explicit join_stream(stream_base<C, B> const& sb) : source_(sb) {}
      explicit join_stream(stream_base<C, B> && sb) : source_(std::forward<stream_base<C, B>>(sb)) { }
      ~join_stream() override {}
      
      ::stream::iterator<T> begin() override {
        return ::stream::iterator<T>{new iterator{source_.begin(), source_.end()}};
      }
      ::stream::iterator<T> end() override {
        return ::stream::iterator<T>{new iterator{source_.end(), source_.end()}};
      }
    private:
      stream_base<C, B> source_;
    };
    
    template <typename C, bool Own>
    auto join(stream_base<C, Own> const& sb) -> stream_base<typename C::value_type, true>  {
      using T = typename C::value_type;
      using impl_t = join_stream<C, false>;
      return stream_base<T, true>{new impl_t{falsify(sb)}};
    }
    
    template <typename C, bool Own>
    auto join(stream_base<C, Own> && sb) -> stream_base<typename C::value_type, true> {
      using T = typename C::value_type;
      using impl_t = join_stream<C, Own>;
      return stream_base<T, true>{new impl_t(std::forward<stream_base<C, Own>>(sb))};
    }
    
    template <typename T, bool Own>
    template <typename F>
    auto stream_base<T, Own>::flatmap(F&& func) const& -> stream_base<flatmap_f<F>, true>  {
      return join((*this)).map(func);
    }
    
    template <typename T, bool Own>
    template <typename F>
    auto stream_base<T, Own>::flatmap(F&& func) && -> stream_base<flatmap_f<F>, true> {
      return join(std::forward<self>(*this)).map(func);
    }
  }
}