From 3bc7376f3a77457679506eb2dbb503cf099be497 Mon Sep 17 00:00:00 2001
From: bergmann <info@bergmann89.de>
Date: Fri, 15 Nov 2019 15:27:10 +0100
Subject: [PATCH] * Implemented basic stream class

---
 include/asyncpp.h             |  2 ++
 include/asyncpp/stream.h      | 36 +++++++++++++++++++
 include/asyncpp/stream.inl    | 39 ++++++++++++++++++++
 include/asyncpp/stream.pre.h  | 27 ++++++++++++++
 test/asyncpp/stream_tests.cpp | 68 +++++++++++++++++++++++++++++++++++
 5 files changed, 172 insertions(+)
 create mode 100644 include/asyncpp/stream.h
 create mode 100644 include/asyncpp/stream.inl
 create mode 100644 include/asyncpp/stream.pre.h
 create mode 100644 test/asyncpp/stream_tests.cpp

diff --git a/include/asyncpp.h b/include/asyncpp.h
index f266853..86d0e88 100644
--- a/include/asyncpp.h
+++ b/include/asyncpp.h
@@ -2,6 +2,8 @@
 
 #include <asyncpp/future.h>
 #include <asyncpp/result.h>
+#include <asyncpp/stream.h>
 
 #include <asyncpp/future.inl>
 #include <asyncpp/result.inl>
+#include <asyncpp/stream.inl>
diff --git a/include/asyncpp/stream.h b/include/asyncpp/stream.h
new file mode 100644
index 0000000..31ff054
--- /dev/null
+++ b/include/asyncpp/stream.h
@@ -0,0 +1,36 @@
+#pragma once
+
+#include <type_traits>
+
+#include "result.h"
+#include "stream.pre.h"
+
+namespace asyncpp
+{
+
+    template<
+        typename T_object,
+        typename T_impl>
+    struct stream
+    {
+        using object_type       = T_object;
+        using clean_object_type = std::decay_t<object_type>;
+        using trait_type        = stream_trait<clean_object_type>;
+        using value_type        = typename trait_type::value_type;
+        using result_type       = stream_result<value_type>;
+
+        object_type ref;
+
+        /**
+         * @brief Value constructor.
+         */
+        template<typename X_object>
+        inline stream(X_object&& p_ref);
+
+        /**
+         * @brief Function that will be called repeatedly to check if the stream has values.
+         */
+        inline result_type poll();
+    };
+
+}
diff --git a/include/asyncpp/stream.inl b/include/asyncpp/stream.inl
new file mode 100644
index 0000000..434286c
--- /dev/null
+++ b/include/asyncpp/stream.inl
@@ -0,0 +1,39 @@
+#pragma once
+
+#include "stream.h"
+
+namespace asyncpp
+{
+
+    /* stream */
+
+    template<
+        typename T_value,
+        typename T_impl>
+    template<
+        typename X_object>
+    stream<T_value, T_impl>
+        ::stream(X_object&& p_ref)
+            : ref(std::forward<X_object>(p_ref))
+        { }
+
+    template<
+        typename T_value,
+        typename T_impl>
+    typename stream<T_value, T_impl>::result_type
+    stream<T_value, T_impl>
+        ::poll()
+        { return trait_type::poll(*this); }
+
+    /* misc */
+
+    template<typename X_value>
+    constexpr decltype(auto) as_stream(X_value&& value)
+    {
+        using value_type  = X_value;
+        using stream_type = stream<value_type>;
+
+        return stream_type(std::forward<X_value>(value));
+    }
+
+}
diff --git a/include/asyncpp/stream.pre.h b/include/asyncpp/stream.pre.h
new file mode 100644
index 0000000..95ad1c9
--- /dev/null
+++ b/include/asyncpp/stream.pre.h
@@ -0,0 +1,27 @@
+#pragma once
+
+namespace asyncpp
+{
+
+    template<typename T_impl>
+    struct stream_base
+    {
+        template<typename T_stream>
+        static inline auto poll(T_stream& self) = delete;
+    };
+
+    template<typename T, typename = void>
+    struct stream_trait;
+
+    template<
+        typename T_object,
+        typename T_impl = stream_trait<std::decay_t<T_object>>>
+    struct stream;
+
+    /**
+     * @brief Construct a stream from the given value.
+     */
+    template<typename X_value>
+    constexpr decltype(auto) as_stream(X_value&& value);
+
+}
diff --git a/test/asyncpp/stream_tests.cpp b/test/asyncpp/stream_tests.cpp
new file mode 100644
index 0000000..271266e
--- /dev/null
+++ b/test/asyncpp/stream_tests.cpp
@@ -0,0 +1,68 @@
+#include <gtest/gtest.h>
+
+#include <asyncpp.h>
+
+using namespace ::testing;
+using namespace ::asyncpp;
+
+struct delay
+{
+    int const delay     { 5 };
+    int const threshold { 2 };
+    int       count     { 0 };
+};
+
+namespace asyncpp
+{
+
+    template<>
+    struct stream_trait<delay, void>
+        : public future_base<future<delay, void>>
+    {
+        using value_type = int;
+
+        template<typename T_future>
+        static inline auto poll(T_future& self)
+        {
+            using result_type = stream_result<value_type>;
+
+            if (self.ref.count >= self.ref.delay)
+                return result_type::done();
+
+            ++self.ref.count;
+
+            return self.ref.count <= self.ref.threshold
+                ? result_type::not_ready()
+                : result_type::ready(self.ref.count);
+        }
+    };
+
+}
+
+TEST(stream_tests, poll)
+{
+    delay d { 5, 2, 0 };
+    auto s = as_stream(d);
+
+    auto r0 = s.poll();
+    ASSERT_FALSE(r0);
+
+    auto r1 = s.poll();
+    ASSERT_FALSE(r1);
+
+    auto r2 = s.poll();
+    ASSERT_TRUE (r2);
+    ASSERT_EQ   (3, *r2);
+
+    auto r3 = s.poll();
+    ASSERT_TRUE (r3);
+    ASSERT_EQ   (4, *r3);
+
+    auto r4 = s.poll();
+    ASSERT_TRUE (r4);
+    ASSERT_EQ   (5, *r4);
+
+    auto r = s.poll();
+    ASSERT_FALSE(r);
+    ASSERT_TRUE (r.is_done());
+}