added pipeline reduction by value

include/pipes/fork.hpp

@@ -5,6 +5,7 @@
 
 #include "pipes/helpers/meta.hpp"
 #include "pipes/base.hpp"
+#include "pipes/impl/pipelines_reduction.hpp"
 
 namespace pipes
 {
@@ -19,6 +20,16 @@ class fork_pipeline : public pipeline_base<fork_pipeline<TailPipelines...>>
         detail::for_each(tailPipelines_, [&value](auto&& tailPipeline){ send(FWD(value), tailPipeline); });
     }
 
+	template<size_t... indices>
+	auto move_reduced_value_from(std::index_sequence<indices...>)
+	{
+		return std::make_tuple(detail::move_reduced_value_from(std::get<indices>(tailPipelines_))...);
+	}
+	auto move_reduced_value_from()
+	{
+		return move_reduced_value_from(std::make_index_sequence<sizeof...(TailPipelines)>());
+	}
+
     explicit fork_pipeline(TailPipelines const&... tailPipelines) : tailPipelines_(tailPipelines...) {}
 
 private:

include/pipes/impl/concepts.hpp

@@ -12,6 +12,11 @@ namespace pipes
     {
         namespace impl
         {
+			// definition of void_t from C++17
+
+			template< class... >
+			using void_t = void;
+
             // definition of range
 
             namespace adl
@@ -50,6 +55,15 @@ namespace pipes
         template<typename Pipeline>
         using IsAPipeline = impl::IsAPipeline<std::remove_reference_t<Pipeline>>;
 
+		template< class, class = void >
+		struct has_reduced_value : std::false_type {};
+		template< class T >
+		struct has_reduced_value<T, impl::void_t<decltype( std::declval<T&>().move_reduced_value_from() )> > : std::true_type {};
+
+        template<typename Pipeline, detail::IsAPipeline<Pipeline> = true>
+        using IsANonReturningPipeline = std::enable_if_t<!detail::has_reduced_value<Pipeline>::value, bool>;
+        template<typename Pipeline, detail::IsAPipeline<Pipeline> = true>
+        using IsAReturningPipeline = std::enable_if_t<detail::has_reduced_value<Pipeline>::value, bool>;
     } // namespace detail
 } // namespace pipes
 

include/pipes/impl/pipelines_reduction.hpp

@@ -0,0 +1,24 @@
+#ifndef PIPELINES_REDUCTION_HPP
+#define PIPELINES_REDUCTION_HPP 
+
+#include "pipes/impl/concepts.hpp"
+
+#include <tuple>
+
+namespace pipes
+{
+	namespace detail{
+		struct no_reduced_value_t{};
+
+		template<typename Pipeline, detail::IsANonReturningPipeline<Pipeline> = true>
+		auto move_reduced_value_from(Pipeline&&){return no_reduced_value_t{};}
+
+		template<typename Pipeline, detail::IsAReturningPipeline<Pipeline> = true>
+		auto move_reduced_value_from(Pipeline&& pipeline)
+		{
+			return pipeline.move_reduced_value_from();
+		}
+	}
+}
+
+#endif /* PIPELINES_REDUCTION_HPP */

include/pipes/impl/pipes_assembly.hpp

@@ -1,6 +1,7 @@
 #ifndef PIPES_PIPES_ASSEMBLY
 #define PIPES_PIPES_ASSEMBLY
 
+#include "pipes/impl/pipelines_reduction.hpp"
 
 
 namespace pipes
@@ -18,6 +19,11 @@ namespace pipes
                 headPipe_.template onReceive<Ts...>(FWD(inputs)..., tailPipeline_);
             }
 
+			auto move_reduced_value_from()
+			{
+				return detail::move_reduced_value_from(tailPipeline_);
+			}
+
             generic_pipeline(HeadPipe headPipe, TailPipeline tailPipeline) : headPipe_(headPipe), tailPipeline_(tailPipeline) {}
 
         private:

include/pipes/operator.hpp

@@ -3,6 +3,7 @@
 
 #include "pipes/impl/concepts.hpp"
 #include "pipes/impl/pipes_assembly.hpp"
+#include "pipes/impl/pipelines_reduction.hpp"
 
 #include <type_traits>
 
@@ -11,22 +12,22 @@ namespace pipes
 
 // range >>= pipeline (rvalue ranges)
 
-    template<typename Range, typename Pipeline, detail::IsARange<Range> = true, detail::IsAPipeline<Pipeline> = true>
-    std::enable_if_t<!std::is_lvalue_reference<Range>::value> operator>>=(Range&& range, Pipeline&& pipeline)
+    template<typename Range, typename Pipeline, detail::IsARange<Range> = true, detail::IsAPipeline<Pipeline> = true, std::enable_if_t<!std::is_lvalue_reference<Range>::value, int> = 0>
+    auto operator>>=(Range&& range, Pipeline&& pipeline)
     {
         using std::begin;
         using std::end;
-        std::copy(std::make_move_iterator(begin(range)), std::make_move_iterator(end(range)), pipeline);
+		return detail::move_reduced_value_from(std::copy(std::make_move_iterator(begin(range)), std::make_move_iterator(end(range)), pipeline));
     }
 
 // range >>= pipeline (lvalue ranges)
 
-    template<typename Range, typename Pipeline, detail::IsARange<Range> = true, detail::IsAPipeline<Pipeline> = true>
-    std::enable_if_t<std::is_lvalue_reference<Range>::value> operator>>=(Range&& range, Pipeline&& pipeline)
+    template<typename Range, typename Pipeline, detail::IsARange<Range> = true, detail::IsAPipeline<Pipeline> = true, std::enable_if_t<std::is_lvalue_reference<Range>::value, int> = 0>
+    auto operator>>=(Range&& range, Pipeline&& pipeline)
     {
         using std::begin;
         using std::end;
-        std::copy(begin(range), end(range), pipeline);
+		return detail::move_reduced_value_from(std::copy(begin(range), end(range), pipeline));
     }
 
 // pipe >>= pipe

include/pipes/pipes.hpp

@@ -26,6 +26,7 @@
 #include "pipes/take.hpp"
 #include "pipes/take_while.hpp"
 #include "pipes/tee.hpp"
+#include "pipes/to_container.hpp"
 #include "pipes/to_out_stream.hpp"
 #include "pipes/transform.hpp"
 #include "pipes/unzip.hpp"

include/pipes/to_container.hpp

@@ -0,0 +1,32 @@
+#ifndef TO_CONTAINER_HPP
+#define TO_CONTAINER_HPP
+
+#include "pipes/base.hpp"
+#include "pipes/helpers/FWD.hpp"
+
+#include <functional>
+#include <iostream>
+
+namespace pipes
+{
+    template<typename Container>
+    class to_ : public pipeline_base<to_<Container>>
+    {
+    public:
+        template<typename T>
+        void onReceive(T&& value)
+        {
+            container_.push_back(FWD(value));
+        }
+
+		auto move_reduced_value_from()
+		{
+			return std::exchange(container_, {});
+		}
+    private:
+		Container container_;
+    };
+
+}
+
+#endif /* TO_CONTAINER_HPP */

tests/CMakeLists.txt

@@ -26,6 +26,7 @@ add_executable(pipes_test
     tap.cpp
     tee.cpp
     transform.cpp
+    to_container.cpp
     unzip.cpp
     integration_tests.cpp)
 add_test(NAME pipes_test COMMAND pipes_test)

tests/fork.cpp

@@ -4,6 +4,7 @@
 #include "pipes/override.hpp"
 #include "pipes/transform.hpp"
 #include "pipes/push_back.hpp"
+#include "pipes/to_container.hpp"
 
 #include <algorithm>
 #include <utility>
@@ -55,9 +56,11 @@ TEST_CASE("fork can send data to other pipes")
 
     std::vector<int> multiplesOf2, multiplesOf3, multiplesOf4;
 
-    std::copy(begin(numbers), end(numbers), pipes::fork(pipes::filter([](int i){return i%2 == 0;}) >>= pipes::push_back(multiplesOf2),
-                                                         pipes::filter([](int i){return i%3 == 0;}) >>= pipes::push_back(multiplesOf3),
-                                                         pipes::filter([](int i){return i%4 == 0;}) >>= pipes::push_back(multiplesOf4)));
+	std::tie(std::ignore, multiplesOf3, multiplesOf4) = (numbers>>=pipes::fork(
+		pipes::filter([](int i){return i%2 == 0;}) >>= pipes::push_back(multiplesOf2),
+        pipes::filter([](int i){return i%3 == 0;}) >>= pipes::to_<std::vector<int>>(),
+        pipes::filter([](int i){return i%4 == 0;}) >>= pipes::to_<std::vector<int>>()
+	));
 
     REQUIRE(multiplesOf2 == expectedMultiplesOf2);
     REQUIRE(multiplesOf3 == expectedMultiplesOf3);

tests/to_container.cpp

@@ -0,0 +1,32 @@
+#include "catch.hpp"
+#include "pipes/pipes.hpp"
+
+#include <algorithm>
+#include <vector>
+
+TEST_CASE("to_vector")
+{
+    std::vector<int> input = {1, 2, 3, 4, 5, 6, 7 ,8, 9, 10};
+    std::vector<int> expected = input;
+
+	REQUIRE(pipes::detail::IsAReturningPipeline< pipes::to_<std::vector<int>> >{true});
+
+    auto results=(input>>=pipes::to_<std::vector<int>>());
+
+    REQUIRE(results == expected);
+}
+
+TEST_CASE("filter_to_vector")
+{
+    std::vector<int> input = {1, 2, 3, 4, 5, 6, 7 ,8, 9, 10};
+    std::vector<int> expected = {4, 8};
+
+	REQUIRE(pipes::detail::IsAReturningPipeline< pipes::to_<std::vector<int>> >{true});
+
+    auto results=(input
+        >>=pipes::filter([](int i){return i%4 == 0;})
+		>>=pipes::to_<std::vector<int>>()
+	);
+
+    REQUIRE(results == expected);
+}