3 files changed, 114 insertions, 0 deletions
diff --git a/Rx/v2/examples/linesfrombytes/CMakeLists.txt b/Rx/v2/examples/linesfrombytes/CMakeLists.txt
new file mode 100644
index 0000000..3bf2979
--- /dev/null
+++ b/Rx/v2/examples/linesfrombytes/CMakeLists.txt
@@ -0,0 +1,36 @@
+cmake_minimum_required(VERSION 3.2 FATAL_ERROR)
+
+get_filename_component(SAMPLE_PROJECT "${CMAKE_CURRENT_SOURCE_DIR}" NAME)
+
+project(${SAMPLE_PROJECT} LANGUAGES C CXX)
+
+# define some folders
+get_filename_component(RXCPP_DIR "${CMAKE_CURRENT_SOURCE_DIR}" PATH)
+get_filename_component(RXCPP_DIR "${RXCPP_DIR}" PATH)
+get_filename_component(RXCPP_DIR "${RXCPP_DIR}" PATH)
+get_filename_component(RXCPP_DIR "${RXCPP_DIR}" PATH)
+
+MESSAGE( STATUS "RXCPP_DIR: " ${RXCPP_DIR} )
+
+include(${RXCPP_DIR}/projects/CMake/shared.cmake)
+
+# define the sources
+set(SAMPLE_SOURCES
+    ${CMAKE_CURRENT_SOURCE_DIR}/main.cpp
+)
+add_executable(${SAMPLE_PROJECT} ${SAMPLE_SOURCES})
+add_executable(rxcpp::examples::${SAMPLE_PROJECT} ALIAS ${SAMPLE_PROJECT})
+target_compile_options(${SAMPLE_PROJECT} PUBLIC ${RX_COMPILE_OPTIONS})
+target_compile_features(${SAMPLE_PROJECT} PUBLIC ${RX_COMPILE_FEATURES})
+target_include_directories(${SAMPLE_PROJECT} PUBLIC ${RX_SRC_DIR})
+target_link_libraries(${SAMPLE_PROJECT} ${CMAKE_THREAD_LIBS_INIT})
+
+# configure unit tests via CTest
+enable_testing()
+set(CTEST_CONFIGURATION_TYPE "${JOB_BUILD_CONFIGURATION}")
+
+set_target_properties(${SAMPLE_PROJECT} PROPERTIES FOLDER "Examples")
+
+add_test(NAME RunTests
+     WORKING_DIRECTORY "${CMAKE_CURRENT_BINARY_DIR}"
+     COMMAND ${SAMPLE_PROJECT} ${TEST_ARGS})
diff --git a/Rx/v2/examples/linesfrombytes/main.cpp b/Rx/v2/examples/linesfrombytes/main.cpp
new file mode 100644
index 0000000..30639a3
--- /dev/null
+++ b/Rx/v2/examples/linesfrombytes/main.cpp
@@ -0,0 +1,77 @@
+
+#include "rxcpp/rx.hpp"
+using namespace rxcpp;
+using namespace rxcpp::sources;
+using namespace rxcpp::util;
+
+#include <regex>
+#include <random>
+using namespace std;
+
+int main()
+{
+    random_device rd;   // non-deterministic generator
+    mt19937 gen(rd());
+    uniform_int_distribution<> dist(4, 18);
+
+    // produce byte stream that contains lines of text
+    auto bytes = range(1, 10).
+        map([&](int i){ 
+            return from((uint8_t)('A' + i)).
+                repeat(dist(gen)).
+                concat(from((uint8_t)'\r'));
+        }).
+        merge().
+        window(17).
+        map([](observable<uint8_t> w){ 
+            return w.
+                reduce(
+                    vector<uint8_t>(), 
+                    [](vector<uint8_t>& v, uint8_t b){
+                        v.push_back(b); 
+                        return move(v);
+                    }, 
+                    [](vector<uint8_t>& v){return move(v);}).
+                as_dynamic(); 
+        }).
+        merge().
+        filter([](vector<uint8_t>& v){
+            copy(v.begin(), v.end(), ostream_iterator<long>(cout, " "));
+            cout << endl; 
+            return true;
+        });
+
+    // create strings split on \r
+    auto strings = bytes.
+        map([](vector<uint8_t> v){
+            string s(v.begin(), v.end());
+            regex delim(R"/(\r)/");
+            sregex_token_iterator cursor(s.begin(), s.end(), delim, {-1, 0});
+            sregex_token_iterator end;
+            vector<string> splits(cursor, end);
+            return iterate(move(splits));
+        }).
+        concat();
+
+    // group strings by line
+    int group = 0;
+    auto linewindows = strings.
+        group_by(
+            [=](string& s) mutable {
+                return s.back() == '\r' ? group++ : group;
+            },
+            [](string& s) { return move(s);});
+
+    // reduce the strings for a line into one string
+    auto lines = linewindows.
+        map([](grouped_observable<int, string> w){ 
+            return w.sum(); 
+        }).
+        merge();
+
+    // print result
+    lines.
+        subscribe(println(cout));
+
+    return 0;
+}
diff --git a/projects/CMake/CMakeLists.txt b/projects/CMake/CMakeLists.txt
index db6dc94..b567ea7 100644
--- a/projects/CMake/CMakeLists.txt
+++ b/projects/CMake/CMakeLists.txt
@@ -19,6 +19,7 @@ set(EXAMPLES_DIR ${RXCPP_DIR}/Rx/v2/examples)
 
 add_subdirectory(${EXAMPLES_DIR}/cep ${CMAKE_CURRENT_BINARY_DIR}/examples/cep)
 add_subdirectory(${EXAMPLES_DIR}/stop ${CMAKE_CURRENT_BINARY_DIR}/examples/stop)
+add_subdirectory(${EXAMPLES_DIR}/linesfrombytes ${CMAKE_CURRENT_BINARY_DIR}/examples/linesfrombytes)
 add_subdirectory(${EXAMPLES_DIR}/println ${CMAKE_CURRENT_BINARY_DIR}/examples/println)
 add_subdirectory(${EXAMPLES_DIR}/pythagorian ${CMAKE_CURRENT_BINARY_DIR}/examples/pythagorian)
 add_subdirectory(${EXAMPLES_DIR}/tests ${CMAKE_CURRENT_BINARY_DIR}/examples/tests)