* Implemented test for 'interval' with 'current_thread' executor

* Refactored 'timer' class
* Fixed bugs in 'interval'

include/asyncpp/core/future/and_then.inl

@@ -48,7 +48,7 @@ namespace asyncpp
                 if (!r)
                     return result_type::not_ready();
 
-                self->second = std::make_unique<second_future_type>(as_future(self->lambda(*r)));
+                self->second = std::make_unique<second_future_type>(as_future(r.call(self->lambda)));
             }
         }
     }

include/asyncpp/core/future/map.inl

@@ -38,7 +38,7 @@ namespace asyncpp
     {
         auto r = self->future.poll();
         return r
-            ? result_type::ready(self->lambda(*r))
+            ? result_type::ready(r.call(self->lambda))
             : result_type::not_ready();
     }
 

include/asyncpp/core/result.h

@@ -111,6 +111,18 @@ namespace asyncpp
              */
             inline const_reference_type value() const;
 
+            /**
+             * @brief Call the passed closure with the value stored in the result.
+             */
+            template<typename X_lambda>
+            inline auto call(const X_lambda& p_lambda);
+
+            /**
+             * @brief Call the passed closure with the value stored in the result.
+             */
+            template<typename X_lambda>
+            inline auto call(const X_lambda& p_lambda) const;
+
         public:
             inline operator bool() const;
             inline pointer_type operator-> ();

include/asyncpp/core/result.inl

@@ -91,6 +91,18 @@ namespace asyncpp
             ::value() const
             { return std::get<ready_type>(_storage).value; }
 
+        template<bool for_stream, typename T_value>
+        template<typename X_lambda>
+        auto result<for_stream, T_value>
+            ::call(const X_lambda& p_lambda)
+            { return p_lambda(value()); }
+
+        template<bool for_stream, typename T_value>
+        template<typename X_lambda>
+        auto result<for_stream, T_value>
+            ::call(const X_lambda& p_lambda) const
+            { return p_lambda(value()); }
+
         template<bool for_stream, typename T_value>
         result<for_stream, T_value>
             ::operator bool() const
@@ -181,6 +193,10 @@ namespace asyncpp
             inline void value()
                 { throw std::runtime_error("'void' result does not store any value!"); }
 
+            template<typename X_lambda>
+            inline auto call(const X_lambda& p_lambda)
+                { return p_lambda(); }
+
             inline operator bool() const
                 { return _status == result_status::ready; }
 

include/asyncpp/core/stream/for_each.h

@@ -45,8 +45,9 @@ namespace asyncpp
     {
         using stream_type       = T_stream;
         using inner_value_type  = typename stream_type::value_type;
+        using inner_result_type = future_result<inner_value_type>;
         using lambda_type       = T_lambda;
-        using value_type        = decltype(std::declval<lambda_type>()(std::declval<inner_value_type>()));
+        using value_type        = decltype(std::declval<inner_result_type>().call(std::declval<lambda_type>()));
         using result_type       = future_result<value_type>;
 
         template<typename X_future>

include/asyncpp/core/stream/for_each.inl

@@ -45,7 +45,7 @@ namespace asyncpp
             if (r.is_not_ready())
                 return result_type::not_ready();
 
-            self->lambda(*r);
+            r.call(self->lambda);
         }
     }
 

include/asyncpp/executor/current_thread.inl

@@ -51,7 +51,10 @@ namespace executor {
     void current_thread<T_runtime>
         ::run(X_future&& p_future)
     {
-        __impl::current_executor_lock l(executor::local_storage(), *this);
+        auto executor_lock  = __impl::current_executor_lock(executor::local_storage(), *this);
+        auto runtime_lock   = _runtime.init_thread();
+
+        (void)runtime_lock;
 
         spawn(std::forward<X_future>(p_future));
 

include/asyncpp/timing/impl/timer_base.h

@@ -73,7 +73,7 @@ namespace __impl {
          */
         registration_ptr_w create_registration(const time_point& p_deadline);
 
-    private:
+    protected:
         struct storage
         {
             timer_base* current { nullptr };

include/asyncpp/timing/impl/timer_impl.h

@@ -0,0 +1,80 @@
+#pragma once
+
+namespace asyncpp {
+namespace timing {
+
+    template<typename T_inner = void>
+    struct timer;
+
+    namespace __impl
+    {
+
+        /* timer_impl - default */
+
+        template<typename T_inner>
+        struct timer_impl
+        {
+        public:
+            using inner_type                = T_inner;
+            using owner_type                = timer<inner_type>;
+            using inner_thread_lock_type    = decltype(std::declval<inner_type>().init_thread());
+
+            struct thread_lock
+            {
+                inner_thread_lock_type inner_lock;
+
+                template<typename... X_args>
+                inline thread_lock(X_args&&... p_args);
+
+                inline ~thread_lock();
+            };
+
+            using thread_lock_ptr_u = std::unique_ptr<thread_lock>;
+
+        public:
+            owner_type& owner;
+            inner_type  inner;
+
+        public:
+            template<typename... X_args>
+            inline timer_impl(
+                owner_type& p_owner,
+                X_args&&... p_args);
+
+            inline void idle(
+                const asyncpp::time_point * p_deadline);
+
+            inline thread_lock_ptr_u init_thread();
+        };
+
+        /* timer_impl<void> */
+
+        template<>
+        struct timer_impl<void>
+        {
+        public:
+            using owner_type = timer<void>;
+
+            struct thread_lock
+            {
+                inline ~thread_lock();
+            };
+
+            using thread_lock_ptr_u = std::unique_ptr<thread_lock>;
+
+        public:
+            owner_type& owner;
+
+        public:
+            inline timer_impl(
+                owner_type& p_owner);
+
+            inline void idle(
+                const asyncpp::time_point * p_deadline);
+
+            inline thread_lock_ptr_u init_thread();
+        };
+
+    }
+
+} }

include/asyncpp/timing/impl/timer_impl.inl

@@ -0,0 +1,73 @@
+#pragma once
+
+#include "timer_impl.h"
+
+namespace asyncpp {
+namespace timing {
+namespace __impl {
+
+    /* timer_impl::thread_lock - default */
+
+    template<typename T_inner>
+    template<typename... X_args>
+    timer_impl<T_inner>::thread_lock::thread_lock(X_args&&... p_args)
+        : inner_lock(std::forward<X_args>(p_args)...)
+        { }
+
+    template<typename T_inner>
+    timer_impl<T_inner>::thread_lock::~thread_lock()
+        { owner_type::local_storage().current = nullptr; }
+
+    /* timer_impl - default */
+
+    template<typename T_inner>
+    template<typename... X_args>
+    timer_impl<T_inner>::timer_impl(
+        owner_type& p_owner,
+        X_args&&... p_args)
+        : owner(p_owner)
+        , inner(std::forward<X_args>(p_args)...)
+        { }
+
+    template<typename T_inner>
+    void timer_impl<T_inner>::idle(const asyncpp::time_point * p_deadline)
+    {
+        {
+        auto r = owner._registrations.lock();
+        if (!r->empty())
+        {
+            p_deadline = merge_deadlines(p_deadline, &(*r->begin())->deadline);
+        }
+        }
+
+        inner.idle(p_deadline);
+    }
+
+    template<typename T_inner>
+    typename timer_impl<T_inner>::thread_lock_ptr_u
+    timer_impl<T_inner>::init_thread()
+        { return std::make_unique<thread_lock>(inner.init_thread()); }
+
+
+
+    /* timer_impl<void>::thread_lock - default */
+
+    timer_impl<void>::thread_lock::~thread_lock()
+        { owner_type::local_storage().current = nullptr; }
+
+    /* timer_impl<void> */
+
+    timer_impl<void>::timer_impl(
+        owner_type& p_owner)
+        : owner(p_owner)
+        { }
+
+    void timer_impl<void>::idle(
+        const asyncpp::time_point * p_deadline)
+        { }
+
+    timer_impl<void>::thread_lock_ptr_u
+    timer_impl<void>::init_thread()
+        { return std::unique_ptr<thread_lock>(); }
+
+} } }

include/asyncpp/timing/interval.inl

@@ -42,16 +42,17 @@ namespace asyncpp
     stream_trait<timing::interval, void>
         ::poll(X_stream& self)
     {
+        if (    self->_deadline.time_since_epoch().count()
+            &&  self->_delay->deadline() >= self->_deadline
+            &&  asyncpp::now()           >= self->_delay->deadline())
+            return result_type::done();
+
         auto ret = self->_delay.poll();
         if (ret.is_not_ready())
             return result_type::not_ready();
 
         auto now = self->_delay->deadline();
         auto new_deadline = now + self->_duration;
-        if (    self->_deadline.time_since_epoch().count()
-            &&  new_deadline >= self->_deadline)
-            return result_type::done();
-
         self->_delay->reset(new_deadline);
 
         return result_type::ready();

include/asyncpp/timing/timer.h

@@ -6,37 +6,25 @@
 #include <cppcore/synchronization/locked.h>
 
 #include "impl/timer_base.h"
+#include "impl/timer_impl.h"
 #include "impl/registration.h"
 
 namespace asyncpp {
 namespace timing {
 
-    namespace __impl
-    {
-
-        template<typename T_inner>
-        struct storage
-        {
-            using inner_type = T_inner;
-
-            inner_type inner;
-
-            template<typename... X_args>
-            inline storage(X_args&&... p_args);
-        };
-
-    }
-
-    template<typename T_inner = void>
+    template<typename T_inner>
     struct timer
         : public __impl::timer_base
     {
     public:
-        using inner_type    = T_inner;
-        using storage_type  = __impl::storage<inner_type>;
+        template<typename X>
+        friend struct __impl::timer_impl;
+
+        using inner_type        = T_inner;
+        using timer_impl_type   = __impl::timer_impl<inner_type>;
 
     private:
-        storage_type _storage;
+        timer_impl_type _impl;
 
     public:
         /**
@@ -53,20 +41,10 @@ namespace timing {
          */
         inline void idle(const asyncpp::time_point * p_deadline);
 
-    private:
-        /**
-         * @brief Call the idle method of the inner runtime.
-         */
-        template<typename X = inner_type>
-        inline auto inner_idle(const asyncpp::time_point * p_deadline)
-            -> std::enable_if_t<std::is_same_v<X, void>>;
-
         /**
-         * @brief Call the idle method of the inner runtime.
+         * @brief This method is calld by the executor when a new thread needs to be initialized.
          */
-        template<typename X = inner_type>
-        inline auto inner_idle(const asyncpp::time_point * p_deadline)
-            -> std::enable_if_t<!std::is_same_v<X, void>>;
+        inline decltype(auto) init_thread();
     };
 
 } }

include/asyncpp/timing/timer.inl

@@ -6,36 +6,18 @@
 #include "delay.inl"
 
 #include "impl/timer_base.inl"
+#include "impl/timer_impl.inl"
 #include "impl/registration.inl"
 
 namespace asyncpp {
 namespace timing {
 
-    namespace __impl
-    {
-
-        template<typename T_inner>
-        template<typename... X_args>
-        storage<T_inner>::storage(X_args&&... p_args)
-            : inner(std::forward<X_args>(p_args)...)
-            { }
-
-        template<>
-        struct storage<void>
-        {
-            using inner_type = void;
-
-            inline storage() = default;
-        };
-
-    }
-
     /* timer */
 
     template<typename T_inner>
     template<typename... X_args>
     timer<T_inner>::timer(X_args&&... p_args)
-        : _storage(std::forward<X_args>(p_args)...)
+        : _impl(*this, std::forward<X_args>(p_args)...)
         { }
 
     template<typename T_inner>
@@ -56,31 +38,14 @@ namespace timing {
         }
         }
 
-        inner_idle(p_deadline);
-    }
-
-    template<typename T_inner>
-    template<typename X>
-    auto timer<T_inner>::inner_idle(const asyncpp::time_point * p_deadline)
-        -> std::enable_if_t<std::is_same_v<X, void>>
-    {
-        /* no-op */
+        _impl.idle(p_deadline);
     }
 
     template<typename T_inner>
-    template<typename X>
-    auto timer<T_inner>::inner_idle(const asyncpp::time_point * p_deadline)
-        -> std::enable_if_t<!std::is_same_v<X, void>>
+    decltype(auto) timer<T_inner>::init_thread()
     {
-        {
-        auto r = _registrations.lock();
-        if (!r->empty())
-        {
-            p_deadline = merge_deadlines(p_deadline, &(*r->begin())->deadline);
-        }
-        }
-
-        _storage.inner.idle(p_deadline);
+        local_storage().current = this;
+        return _impl.init_thread();
     }
 
 } }

test/asyncpp/executor/current_thread_tests.cpp

@@ -1,9 +1,11 @@
 #include <gtest/gtest.h>
 
+#include "../../helper/now_mock.h"
 #include "../../helper/runtime_mock.h"
 
-#include <asyncpp.h>
+#include <asyncpp/timing.h>
 #include <asyncpp/executor/current_thread.h>
+#include <asyncpp.h>
 
 using namespace ::testing;
 using namespace ::asyncpp;
@@ -14,6 +16,11 @@ struct test
     task_ptr_w  task;
 };
 
+struct mock
+{
+    MOCK_METHOD0(call, void());
+};
+
 namespace asyncpp
 {
 
@@ -37,11 +44,13 @@ namespace asyncpp
 }
 
 TEST(current_thread_tests, run)
-{
+{/*
     Sequence s;
     StrictMock<runtime_mock> m;
     executor::current_thread<StrictMock<runtime_mock>&> e(m);
 
+    EXPECT_CALL(m, init_thread());
+
     test t;
     auto f = as_future(t);
 
@@ -59,5 +68,90 @@ TEST(current_thread_tests, run)
             s->notify();
         }));;
 
-    e.run(f);
+    e.run(f); */
+}
+
+TEST(current_thread_tests, interval)
+{
+    using mock_type         = StrictMock<mock>;
+    using now_mock_type     = StrictMock<now_mock>;
+    using runtime_mock_type = StrictMock<runtime_mock>;
+    using runtime_type      = timing::timer<runtime_mock_type&>;
+    using executor_type     = executor::current_thread<runtime_type&>;
+
+    InSequence          s;
+    mock_type           m;
+    now_mock_type       nm;
+    runtime_mock_type   rm;
+    runtime_type        r(rm);
+    executor_type       e(r);
+
+    EXPECT_CALL(rm, init_thread());
+    EXPECT_CALL(nm, now()) // timing::delay::poll
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+    EXPECT_CALL(m, call());
+
+    EXPECT_CALL(nm, now()) // timing::delay::poll
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+
+    EXPECT_CALL(nm, now()) // timer::idle
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+    EXPECT_CALL(rm, idle(Pointee(time_point(std::chrono::seconds(10)))));
+
+    EXPECT_CALL(nm, now()) // timer::idle
+        .WillOnce(Return(time_point(std::chrono::seconds(10))));
+    EXPECT_CALL(rm, idle(Pointee(time_point(std::chrono::seconds(10)))));
+
+    EXPECT_CALL(nm, now()) // timing::delay::poll
+        .WillOnce(Return(time_point(std::chrono::seconds(11))));
+    EXPECT_CALL(m, call());
+
+    EXPECT_CALL(nm, now()) // timing::delay::poll
+        .WillOnce(Return(time_point(std::chrono::seconds(11))));
+
+    EXPECT_CALL(nm, now()) // timer::idle
+        .WillOnce(Return(time_point(std::chrono::seconds(15))));
+    EXPECT_CALL(rm, idle(Pointee(time_point(std::chrono::seconds(20)))));
+
+    EXPECT_CALL(nm, now()) // timer::idle
+        .WillOnce(Return(time_point(std::chrono::seconds(49))));
+    EXPECT_CALL(rm, idle(Pointee(time_point(std::chrono::seconds(20)))));
+
+    EXPECT_CALL(nm, now()) // timing::delay::poll
+        .WillOnce(Return(time_point(std::chrono::seconds(49))));
+    EXPECT_CALL(m, call());
+
+    EXPECT_CALL(nm, now()) // timing::delay::poll
+        .WillOnce(Return(time_point(std::chrono::seconds(49))));
+    EXPECT_CALL(m, call());
+
+    EXPECT_CALL(nm, now()) // timing::delay::poll
+        .WillOnce(Return(time_point(std::chrono::seconds(49))));
+    EXPECT_CALL(m, call());
+    EXPECT_CALL(nm, now()) // timing::interval::poll
+        .WillOnce(Return(time_point(std::chrono::seconds(49))));
+
+    EXPECT_CALL(nm, now()) // timing::delay::poll
+        .WillOnce(Return(time_point(std::chrono::seconds(49))));
+
+    EXPECT_CALL(nm, now()) // timer::idle
+        .WillOnce(Return(time_point(std::chrono::seconds(49))));
+    EXPECT_CALL(rm, idle(Pointee(time_point(std::chrono::seconds(50)))));
+
+    EXPECT_CALL(nm, now()) // timer::idle
+        .WillOnce(Return(time_point(std::chrono::seconds(50))));
+    EXPECT_CALL(rm, idle(Pointee(time_point(std::chrono::seconds(50)))));
+
+    EXPECT_CALL(nm, now()) // timing::interval::poll
+        .WillOnce(Return(time_point(std::chrono::seconds(50))));
+
+    e.run(
+        as_stream(
+            timing::interval(
+                time_point(),
+                std::chrono::seconds(10),
+                time_point() + std::chrono::seconds(50)))
+            .for_each([&m]{
+                m.call();
+            }));
 }

test/asyncpp/timing/interval_tests.cpp

@@ -21,8 +21,7 @@ TEST(interval_tests, poll)
 
     // 30 - 10 = 20
     // 20 / 5  = 4
-    // 4  - 1  = 3
-    // 3 x Ready
+    // 4 x Ready
 
     t.make_current();
 
@@ -43,7 +42,7 @@ TEST(interval_tests, poll)
     EXPECT_CALL(m, now)
         .WillOnce(Return(time_point(std::chrono::seconds(10))));
 
-    auto r2 = s.poll();
+    auto r2 = s.poll(); // ready @ 10sec
     ASSERT_EQ(result_status::ready, r2.status());
 
     EXPECT_CALL(m, now)
@@ -55,18 +54,24 @@ TEST(interval_tests, poll)
     EXPECT_CALL(m, now)
         .WillOnce(Return(time_point(std::chrono::seconds(30))));
 
-    auto r4 = s.poll();
+    auto r4 = s.poll(); // ready @ 15sec
     ASSERT_EQ(result_status::ready, r4.status());
 
     EXPECT_CALL(m, now)
         .WillOnce(Return(time_point(std::chrono::seconds(30))));
 
-    auto r5 = s.poll();
+    auto r5 = s.poll(); // ready @ 20sec
     ASSERT_EQ(result_status::ready, r5.status());
 
     EXPECT_CALL(m, now)
         .WillOnce(Return(time_point(std::chrono::seconds(30))));
 
-    auto r6 = s.poll();
-    ASSERT_EQ(result_status::done, r6.status());
+    auto r6 = s.poll(); // ready @ 25sec
+    ASSERT_EQ(result_status::ready, r6.status());
+
+    EXPECT_CALL(m, now)
+        .WillOnce(Return(time_point(std::chrono::seconds(30))));
+
+    auto r7 = s.poll();
+    ASSERT_EQ(result_status::done, r7.status());
 }

test/helper/runtime_mock.h

@@ -5,8 +5,12 @@
 
 #include <asyncpp/core/misc.h>
 
+struct runtime_thread_lock
+    { };
+
 struct runtime_mock
 {
 public:
-    MOCK_METHOD1(idle, void (const asyncpp::time_point * p_deadline));
+    MOCK_METHOD1(idle,          void                (const asyncpp::time_point * p_deadline));
+    MOCK_METHOD0(init_thread,   runtime_thread_lock (void));
 };