Merge pull request #28 from andreiavrammsd/stream-order

Stream order

Author: andreiavrammsd

CMakeLists.txt

@@ -1,6 +1,6 @@
 cmake_minimum_required(VERSION 3.12)
 project(cpp_channel)
-set(PROJECT_VERSION 0.7.3)
+set(PROJECT_VERSION 0.8.0)
 
 set(CMAKE_CXX_STANDARD 11)
 set(CMAKE_CXX_STANDARD_REQUIRED YES)

README.md

@@ -5,12 +5,12 @@
 ### Thread-safe container for sharing data between threads. Header-only.
 
 * Thread-safe push and fetch.
-* Use stream operators to push (>>) and fetch (<<) items.
+* Use stream operators to push (<<) and fetch (>>) items.
 * Blocking (forever waiting to fetch).
 * Range-based for loop supported.
 * Close to prevent pushing and stop waiting to fetch.
 * Integrates well with STL algorithms. Eg: std::move(ch.begin(), ch.end(), ...).
-* Tested with GCC and Clang.
+* Tested with GCC, Clang, and MSVC.
 
 ## Requirements
 
@@ -36,10 +36,10 @@ int main() {
     int out = 0;
 
     // Send to channel
-    in >> chan;
+    chan << in;
 
     // Read from channel
-    out << chan;
+    chan >> out;
 
     assert(out == 1);
 }
@@ -51,12 +51,10 @@ int main() {
 int main() {
     msd::channel<int> chan{2}; // buffered
 
-    int in = 1;
-
     // Send to channel
-    in >> chan;
-    in >> chan;
-    in >> chan; // blocking because capacity is 2 (and no one reads from channel)
+    chan << 1;
+    chan << 2;
+    chan << 3; // blocking because capacity is 2 (and no one reads from channel)
 }
 ```
 
@@ -70,13 +68,13 @@ int main() {
     int out = 0;
 
     // Send to channel
-    in >> chan;
-    in >> chan;
+    chan << in;
+    chan << in;
 
     // Read from channel
-    out << chan;
-    out << chan;
-    out << chan; // blocking because channel is empty (and no one writes on it)
+    chan >> out;
+    chan >> out;
+    chan >> out; // blocking because channel is empty (and no one writes on it)
 }
 ```
 
@@ -89,12 +87,11 @@ int main() {
     msd::channel<int> chan;
 
     int in1 = 1;
-    in1 >> chan;
-
     int in2 = 2;
-    in2 >> chan;
 
-    for (const auto out : chan) {  // blocking: forever waiting for channel items
+    chan << in1 << in2;
+
+    for (const auto out : chan) { // blocking: forever waiting for channel items
         std::cout << out << '\n';
     }
 }

examples/basic.cpp

@@ -9,13 +9,13 @@ int main()
     int in{};
 
     in = 1;
-    in >> ch;
+    ch << in;
 
     in = 2;
-    in >> ch;
+    ch << in;
 
     in = 3;
-    in >> ch;
+    ch << in;
 
     for (auto out : ch) {
         std::cout << out << '\n';

examples/close.cpp

@@ -17,7 +17,7 @@ int main()
                 break;
             }
 
-            ++i >> ch;
+            ch << ++i;
             std::cout << "in: " << i << "\n";
 
             std::this_thread::sleep_for(std::chrono::milliseconds{ms});

examples/cmake-project/CMakeLists.txt

@@ -14,7 +14,7 @@ target_link_libraries(cmake_project)
 
 include(FetchContent)
 if (NOT channel_POPULATED)
-    FetchContent_Declare(channel URL https://github.com/andreiavrammsd/cpp-channel/archive/v0.7.3.zip)
+    FetchContent_Declare(channel URL https://github.com/andreiavrammsd/cpp-channel/archive/v0.8.0.zip)
     FetchContent_Populate(channel)
     include_directories(${channel_SOURCE_DIR}/include)
     # OR

examples/cmake-project/Makefile

@@ -4,10 +4,10 @@ CMAKE=cmake
 
 build:
 	$(CMAKE) -B $(BUILD_PATH) -DCMAKE_BUILD_TYPE=$(BUILD_TYPE)
-	$(CMAKE) --build $(BUILD_PATH) --target cmake-project -- -j
+	$(CMAKE) --build $(BUILD_PATH) --target cmake_project -- -j
 
 run:
-	$(BUILD_PATH)/cmake-project
+	$(BUILD_PATH)/cmake_project
 
 release:
 	make BUILD_TYPE=release

examples/cmake-project/src/main.cpp

@@ -11,7 +11,7 @@ using chan = msd::channel<int>;
 {
     while (true) {
         static int i = 0;
-        ++i >> incoming;
+        incoming << ++i;
     }
 }
 
@@ -27,7 +27,7 @@ void Transform(chan& incoming, chan& outgoing)
 {
     for (auto in : incoming) {
         auto result = Add(in, 2);
-        result >> outgoing;
+        outgoing << result;
     }
 }
 

examples/move.cpp

@@ -41,12 +41,12 @@ int main()
     msd::channel<Data> ch{10};
 
     auto in1 = Data{1};
-    in1 >> ch;
+    ch << in1;
 
-    Data{2} >> ch;
+    ch << Data{2};
 
     auto in3 = Data{3};
-    std::move(in3) >> ch;
+    ch << std::move(in3);
 
     for (auto out : ch) {
         std::cout << out.getI() << '\n';

examples/multithreading.cpp

@@ -29,7 +29,7 @@ int main()
     const auto in = [](msd::channel<std::int64_t>& ch) {
         while (true) {
             static std::int64_t i = 0;
-            ++i >> ch;
+            ch << ++i;
         }
     };
 

examples/streaming.cpp

@@ -23,7 +23,7 @@ int main()
             }
 
             ++i;
-            std::string{std::to_string(i) + " from: " + std::to_string(thread)} >> ch;
+            ch << std::string{std::to_string(i) + " from: " + std::to_string(thread)};
 
             std::this_thread::sleep_for(pause);
         }

include/msd/blocking_iterator.hpp

@@ -33,8 +33,8 @@ class blocking_iterator {
      */
     value_type operator*() const
     {
-        value_type value;
-        value << ch_;
+        value_type value{};
+        ch_ >> value;
 
         return value;
     }

include/msd/channel.hpp

@@ -62,15 +62,15 @@ class channel {
      * @throws closed_channel if channel is closed.
      */
     template <typename Type>
-    friend void operator>>(Type&&, channel<typename std::decay<Type>::type>&);
+    friend channel<typename std::decay<Type>::type>& operator<<(channel<typename std::decay<Type>::type>&, Type&&);
 
     /**
      * Pops an element from the channel.
      *
      * @tparam Type The type of the elements
      */
     template <typename Type>
-    friend void operator<<(Type&, channel<Type>&);
+    friend channel<Type>& operator>>(channel<Type>&, Type&);
 
     /**
      * Returns the number of elements in the channel.
@@ -120,8 +120,8 @@ class channel {
     friend class blocking_iterator<channel>;
 };
 
-#include "channel_impl.hpp"
-
 }  // namespace msd
 
+#include "channel.inl"
+
 #endif  // MSD_CHANNEL_HPP_

include/msd/channel_impl.hpp include/msd/channel.inl

@@ -1,12 +1,14 @@
 // Copyright (C) 2023 Andrei Avram
 
+namespace msd {
+
 template <typename T>
 constexpr channel<T>::channel(const size_type capacity) : cap_{capacity}
 {
 }
 
 template <typename T>
-void operator>>(T&& in, channel<typename std::decay<T>::type>& ch)
+channel<typename std::decay<T>::type>& operator<<(channel<typename std::decay<T>::type>& ch, T&& in)
 {
     if (ch.closed()) {
         throw closed_channel{"cannot write on closed channel"};
@@ -21,13 +23,15 @@ void operator>>(T&& in, channel<typename std::decay<T>::type>& ch)
     }
 
     ch.cnd_.notify_one();
+
+    return ch;
 }
 
 template <typename T>
-void operator<<(T& out, channel<T>& ch)
+channel<T>& operator>>(channel<T>& ch, T& out)
 {
     if (ch.closed() && ch.empty()) {
-        return;
+        return ch;
     }
 
     {
@@ -42,6 +46,8 @@ void operator<<(T& out, channel<T>& ch)
     }
 
     ch.cnd_.notify_one();
+
+    return ch;
 }
 
 template <typename T>
@@ -94,3 +100,5 @@ void channel<T>::waitBeforeWrite(std::unique_lock<std::mutex>& lock)
         cnd_.wait(lock, [this]() { return size_ < cap_; });
     }
 }
+
+}  // namespace msd

tests/blocking_iterator_test.cpp

@@ -21,9 +21,9 @@ TEST(ChannelIteratorTest, Dereference)
     msd::blocking_iterator<msd::channel<int>> it{channel};
 
     int in = 1;
-    in >> channel;
+    channel << in;
     in = 2;
-    in >> channel;
+    channel << in;
 
     EXPECT_EQ(1, *it);
     EXPECT_EQ(2, *it);
@@ -44,7 +44,7 @@ TEST(ChannelIteratorTest, NotEqualContinue)
     msd::channel<int> channel;
     msd::blocking_iterator<msd::channel<int>> it{channel};
 
-    1 >> channel;
+    channel << 1;
 
     EXPECT_TRUE(it != it);
 }

tests/channel_benchmark.cpp

@@ -33,8 +33,8 @@ static void BM_ChannelWithInt(benchmark::State& state)
     int in = 1;
     int out = 0;
     for (auto _ : state) {
-        in >> channel;
-        out << channel;
+        channel << in;
+        channel >> out;
     }
 }
 
@@ -46,8 +46,8 @@ static void BM_ChannelWithString(benchmark::State& state)
     std::string in = "input";
     std::string out;
     for (auto _ : state) {
-        in >> channel;
-        out << channel;
+        channel << in;
+        channel >> out;
     }
 }
 
@@ -59,8 +59,8 @@ static void BM_ChannelWithStruct(benchmark::State& state)
     Entry in{};
     Entry out{};
     for (auto _ : state) {
-        in >> channel;
-        out << channel;
+        channel << in;
+        channel >> out;
     }
 }