make operator| work better

converts linesfrombytes example to use operator|

1 files changed, 31 insertions, 29 deletions

diff --git a/README.md b/README.md
index ce0b5ab..ee47366 100644
--- a/README.md
+++ b/README.md
@@ -3,24 +3,23 @@ The Reactive Extensions for Native (__RxCpp__) is a library for composing asynch
 Windows: [![Windows Status](http://img.shields.io/appveyor/ci/kirkshoop/RxCpp-446.svg?style=flat-square)](https://ci.appveyor.com/project/kirkshoop/rxcpp-446)
 Linux & OSX: [![Linux & Osx Status](http://img.shields.io/travis/Reactive-Extensions/RxCpp.svg?style=flat-square)](https://travis-ci.org/Reactive-Extensions/RxCpp)
 
+[![GitHub license](https://img.shields.io/github/license/Reactive-Extensions/RxCpp.svg?style=flat-square)](https://github.com/Reactive-Extensions/RxCpp)
+
+[![Join in on gitter.im](https://img.shields.io/gitter/room/Reactive-Extensions/RxCpp.svg?style=flat-square)](https://gitter.im/Reactive-Extensions/RxCpp?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
+[![doxygen documentation](https://img.shields.io/badge/documentation-latest-brightgreen.svg?style=flat-square)](http://reactive-extensions.github.io/RxCpp)
+
 Github: [![GitHub release](https://img.shields.io/github/release/Reactive-Extensions/RxCpp.svg?style=flat-square)](https://github.com/Reactive-Extensions/RxCpp/releases)
 [![GitHub commits](https://img.shields.io/github/commits-since/Reactive-Extensions/RxCpp/v2.2.0.svg?style=flat-square)](https://github.com/Reactive-Extensions/RxCpp)
 
 NuGet: [![NuGet version](http://img.shields.io/nuget/v/RxCpp.svg?style=flat-square)](http://www.nuget.org/packages/RxCpp/)
 [![NuGet downloads](http://img.shields.io/nuget/dt/RxCpp.svg?style=flat-square)](http://www.nuget.org/packages/RxCpp/)
 
-[![GitHub license](https://img.shields.io/github/license/Reactive-Extensions/RxCpp.svg?style=flat-square)](https://github.com/Reactive-Extensions/RxCpp)
-
-[![Join in on gitter.im](https://img.shields.io/gitter/room/Reactive-Extensions/RxCpp.svg?style=flat-square)](https://gitter.im/Reactive-Extensions/RxCpp?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
-[![doxygen documentation](https://img.shields.io/badge/documentation-latest-brightgreen.svg?style=flat-square)](http://reactive-extensions.github.io/RxCpp)
-
 #Example
 Add ```Rx/v2/src``` to the include paths
 
 [![lines from bytes](https://img.shields.io/badge/blog%20post-lines%20from%20bytes-blue.svg?style=flat-square)](http://kirkshoop.github.io/async/rxcpp/c++/2015/07/07/rxcpp_-_parsing_bytes_to_lines_of_text.html)
 
 ```cpp
-
 #include "rxcpp/rx.hpp"
 using namespace rxcpp;
 using namespace rxcpp::sources;
@@ -31,41 +30,42 @@ using namespace rxcpp::util;
 #include <random>
 using namespace std;
 
-//using rxcpp::operators::sum;
-
 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
+    // for testing purposes, produce byte stream that from lines of text
     auto bytes = range(1, 10) |
-        flat_map([&](int i){ 
-            return from(
-                from((uint8_t)('A' + i)) |
-                    repeat(dist(gen)), 
-                from((uint8_t)'\r')) |
-                concat();
+        flat_map([&](int i){
+            auto body = from((uint8_t)('A' + i)) |
+                repeat(dist(gen)) |
+                as_dynamic();
+            auto delim = from((uint8_t)'\r');
+            return from(body, delim) | concat();
         }) |
         window(17) |
-        flat_map([](observable<uint8_t> w){ 
+        flat_map([](observable<uint8_t> w){
             return w |
                 reduce(
-                    vector<uint8_t>(), 
+                    vector<uint8_t>(),
                     [](vector<uint8_t>& v, uint8_t b){
-                        v.push_back(b); 
+                        v.push_back(b);
                         return move(v);
-                    }, 
-                    [](vector<uint8_t>& v){return move(v);}) |
-                as_dynamic(); 
+                    }) |
+                as_dynamic();
         }) |
-        filter([](vector<uint8_t>& v){
+        tap([](vector<uint8_t>& v){
+            // print input packet of bytes
             copy(v.begin(), v.end(), ostream_iterator<long>(cout, " "));
-            cout << endl; 
-            return true;
+            cout << endl;
         });
 
+    //
+    // recover lines of text from byte stream
+    //
+
     // create strings split on \r
     auto strings = bytes |
         concat_map([](vector<uint8_t> v){
@@ -75,6 +75,9 @@ int main()
             sregex_token_iterator end;
             vector<string> splits(cursor, end);
             return iterate(move(splits));
+        }) |
+        filter([](string& s){
+            return !s.empty();
         });
 
     // group strings by line
@@ -83,18 +86,17 @@ int main()
         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 |
-        flat_map([](grouped_observable<int, string> w){ 
-            return w | sum(); 
+        flat_map([](grouped_observable<int, string> w){
+            return w | sum();
         });
 
     // print result
     lines |
-        subscribe(println(cout));
+        subscribe<string>(println(cout));
 
     return 0;
 }