From e8d33bdc521cfbac164c356c935ddb4432447769 Mon Sep 17 00:00:00 2001
From: bergmann <info@bergmann89.de>
Date: Sat, 23 Nov 2019 01:09:58 +0100
Subject: [PATCH] * Implemented 'timeout' for futures and streams

---
 include/asyncpp/core.h                        |   2 +
 include/asyncpp/core/future.h                 |  10 +
 include/asyncpp/core/future.inl               |  38 +++-
 include/asyncpp/core/future.pre.h             |  17 +-
 include/asyncpp/core/misc.inl                 |   6 +-
 include/asyncpp/core/result.h                 |   3 +
 include/asyncpp/core/result.inl               |   2 +
 include/asyncpp/core/stream.h                 |  10 +
 include/asyncpp/core/stream.inl               |  38 +++-
 include/asyncpp/core/stream.pre.h             |  15 +-
 include/asyncpp/timer.h                       |   9 -
 include/asyncpp/timing.h                      |  13 ++
 include/asyncpp/{timer => timing}/delay.h     |   9 +-
 include/asyncpp/{timer => timing}/delay.inl   |  15 +-
 include/asyncpp/{timer => timing}/delay.pre.h |   2 +-
 .../{timer => timing}/impl/registration.h     |   2 +-
 .../{timer => timing}/impl/registration.inl   |   2 +-
 .../{timer => timing}/impl/timer_base.h       |   2 +-
 .../{timer => timing}/impl/timer_base.inl     |   2 +-
 .../{timer => timing}/impl/timer_base.pre.h   |   2 +-
 include/asyncpp/{timer => timing}/interval.h  |   8 +-
 .../asyncpp/{timer => timing}/interval.inl    |  10 +-
 include/asyncpp/timing/timeout.h              |  94 ++++++++
 include/asyncpp/timing/timeout.inl            |  85 ++++++++
 include/asyncpp/{timer => timing}/timer.h     |   2 +-
 include/asyncpp/{timer => timing}/timer.inl   |   4 +-
 .../asyncpp/{timer => timing}/delay_tests.cpp |   6 +-
 .../{timer => timing}/interval_tests.cpp      |   6 +-
 test/asyncpp/timing/timeout_tests.cpp         | 202 ++++++++++++++++++
 .../asyncpp/{timer => timing}/timer_tests.cpp |  38 ++--
 test/helper/now_mock.h                        |   6 +-
 31 files changed, 589 insertions(+), 71 deletions(-)
 delete mode 100644 include/asyncpp/timer.h
 create mode 100644 include/asyncpp/timing.h
 rename include/asyncpp/{timer => timing}/delay.h (88%)
 rename include/asyncpp/{timer => timing}/delay.inl (76%)
 rename include/asyncpp/{timer => timing}/delay.pre.h (75%)
 rename include/asyncpp/{timer => timing}/impl/registration.h (97%)
 rename include/asyncpp/{timer => timing}/impl/registration.inl (95%)
 rename include/asyncpp/{timer => timing}/impl/timer_base.h (99%)
 rename include/asyncpp/{timer => timing}/impl/timer_base.inl (99%)
 rename include/asyncpp/{timer => timing}/impl/timer_base.pre.h (81%)
 rename include/asyncpp/{timer => timing}/interval.h (89%)
 rename include/asyncpp/{timer => timing}/interval.inl (85%)
 create mode 100644 include/asyncpp/timing/timeout.h
 create mode 100644 include/asyncpp/timing/timeout.inl
 rename include/asyncpp/{timer => timing}/timer.h (98%)
 rename include/asyncpp/{timer => timing}/timer.inl (97%)
 rename test/asyncpp/{timer => timing}/delay_tests.cpp (89%)
 rename test/asyncpp/{timer => timing}/interval_tests.cpp (94%)
 create mode 100644 test/asyncpp/timing/timeout_tests.cpp
 rename test/asyncpp/{timer => timing}/timer_tests.cpp (63%)

diff --git a/include/asyncpp/core.h b/include/asyncpp/core.h
index 78aa6ce..401c176 100644
--- a/include/asyncpp/core.h
+++ b/include/asyncpp/core.h
@@ -1,10 +1,12 @@
 #pragma once
 
 #include "core/future.h"
+#include "core/misc.h"
 #include "core/result.h"
 #include "core/stream.h"
 #include "core/task.h"
 
 #include "core/future.inl"
+#include "core/misc.inl"
 #include "core/result.inl"
 #include "core/stream.inl"
diff --git a/include/asyncpp/core/future.h b/include/asyncpp/core/future.h
index 32465ce..ca5d0ef 100644
--- a/include/asyncpp/core/future.h
+++ b/include/asyncpp/core/future.h
@@ -52,6 +52,16 @@ namespace asyncpp
         template<typename X_lambda>
         inline auto and_then(X_lambda&& p_lambda);
 
+        #ifdef asyncpp_timing
+        /**
+         * @brief Throw an execption if the timeout has passed.
+         *
+         * This method is only enabled if timer.h is included before.
+         */
+        template<typename X_base, typename X_ratio>
+        inline auto timeout(const duration<X_base, X_ratio>& p_timeout);
+        #endif
+
     public:
         inline pointer operator->();
         inline reference operator*();
diff --git a/include/asyncpp/core/future.inl b/include/asyncpp/core/future.inl
index c4b0f6f..1659aa5 100644
--- a/include/asyncpp/core/future.inl
+++ b/include/asyncpp/core/future.inl
@@ -5,6 +5,10 @@
 #include "future/map.inl"
 #include "future/and_then.inl"
 
+#ifdef asyncpp_timing
+#include <asyncpp/timing/timeout.inl>
+#endif
+
 namespace asyncpp
 {
 
@@ -38,6 +42,21 @@ namespace asyncpp
             std::forward<X_lambda>(p_lambda)));
     }
 
+    #ifdef asyncpp_timing
+    template<typename T_impl>
+    template<typename X_future, typename X_base, typename X_ratio>
+    auto future_base<T_impl>
+        ::timeout(X_future&& self, const duration<X_base, X_ratio>& p_timeout)
+    {
+        using future_type   = X_future;
+        using timeout_type  = timing::timeout<future_type>;
+
+        return as_future(timeout_type(
+            std::forward<X_future>(self),
+            p_timeout));
+    }
+    #endif
+
     /* future */
 
     template<
@@ -76,6 +95,18 @@ namespace asyncpp
         ::and_then(X_lambda&& p_lambda)
         { return trait_type::and_then(std::move(*this), std::forward<X_lambda>(p_lambda)); }
 
+    #ifdef asyncpp_timing
+    template<
+        typename T_value,
+        typename T_impl>
+    template<
+        typename X_base,
+        typename X_ratio>
+    auto future<T_value, T_impl>
+        ::timeout(const duration<X_base, X_ratio>& p_timeout)
+        { return trait_type::timeout(std::move(*this), p_timeout); }
+    #endif
+
     template<
         typename T_value,
         typename T_impl>
@@ -107,7 +138,7 @@ namespace asyncpp
     /* misc */
 
     template<typename X_value>
-    constexpr decltype(auto) as_future(X_value&& value)
+    constexpr future<X_value> as_future(X_value&& value)
     {
         using value_type  = X_value;
         using future_type = future<value_type>;
@@ -115,4 +146,9 @@ namespace asyncpp
         return future_type(std::forward<X_value>(value));
     }
 
+    template<typename T>
+    struct is_future<future<T>, void>
+        : public std::true_type
+        { };
+
 }
diff --git a/include/asyncpp/core/future.pre.h b/include/asyncpp/core/future.pre.h
index b8c7e77..c61db97 100644
--- a/include/asyncpp/core/future.pre.h
+++ b/include/asyncpp/core/future.pre.h
@@ -1,5 +1,7 @@
 #pragma once
 
+#include "misc.h"
+
 namespace asyncpp
 {
 
@@ -18,6 +20,11 @@ namespace asyncpp
 
         template<typename X_future, typename X_lambda>
         static inline auto and_then(X_future&& self, X_lambda&& p_lambda);
+
+        #ifdef asyncpp_timing
+        template<typename X_future, typename X_base, typename X_ratio>
+        static inline auto timeout(X_future&& self, const duration<X_base, X_ratio>& p_timeout);
+        #endif
     };
 
     template<typename T, typename = void>
@@ -28,10 +35,18 @@ namespace asyncpp
         typename T_impl = future_trait<std::decay_t<T_object>>>
     struct future;
 
+    template<typename T, typename = void>
+    struct is_future
+        : public std::false_type
+        { };
+
+    template<typename T>
+    constexpr decltype(auto) is_future_v = is_future<T>::value;
+
     /**
      * @brief Construct a future from the given value.
      */
     template<typename X_value>
-    constexpr decltype(auto) as_future(X_value&& value);
+    constexpr future<X_value> as_future(X_value&& value);
 
 }
diff --git a/include/asyncpp/core/misc.inl b/include/asyncpp/core/misc.inl
index f969fd7..853448a 100644
--- a/include/asyncpp/core/misc.inl
+++ b/include/asyncpp/core/misc.inl
@@ -2,8 +2,8 @@
 
 #include "misc.h"
 
-namespace asyncpp {
-namespace timer {
+namespace asyncpp
+{
 
     #ifndef __asyncpp_has_impl_timer_now
     time_point now()
@@ -34,4 +34,4 @@ namespace timer {
         }
     }
 
-} }
+}
diff --git a/include/asyncpp/core/result.h b/include/asyncpp/core/result.h
index e205fbf..6c2ce10 100644
--- a/include/asyncpp/core/result.h
+++ b/include/asyncpp/core/result.h
@@ -119,6 +119,9 @@ namespace asyncpp
             inline const_reference_type operator* () const;
         };
 
+        template<bool for_stream>
+        struct result<for_stream, void>;
+
     }
 
 }
diff --git a/include/asyncpp/core/result.inl b/include/asyncpp/core/result.inl
index 074da4c..cd7855e 100644
--- a/include/asyncpp/core/result.inl
+++ b/include/asyncpp/core/result.inl
@@ -120,6 +120,8 @@ namespace asyncpp
             ::operator* () const
             { return value(); }
 
+        /* result<void> */
+
         template<bool for_stream>
         struct result<for_stream, void>
         {
diff --git a/include/asyncpp/core/stream.h b/include/asyncpp/core/stream.h
index a37d3b5..1752769 100644
--- a/include/asyncpp/core/stream.h
+++ b/include/asyncpp/core/stream.h
@@ -46,6 +46,16 @@ namespace asyncpp
         template<typename X_lambda>
         inline auto for_each(X_lambda&& p_lambda);
 
+        #ifdef asyncpp_timing
+        /**
+         * @brief Throw an execption if the timeout has passed.
+         *
+         * This method is only enabled if timer.h is included before.
+         */
+        template<typename X_base, typename X_ratio>
+        inline auto timeout(const duration<X_base, X_ratio>& p_timeout);
+        #endif
+
     public:
         inline pointer operator->();
         inline reference operator*();
diff --git a/include/asyncpp/core/stream.inl b/include/asyncpp/core/stream.inl
index 51550e7..ed1c88f 100644
--- a/include/asyncpp/core/stream.inl
+++ b/include/asyncpp/core/stream.inl
@@ -4,6 +4,10 @@
 
 #include "stream/for_each.inl"
 
+#ifdef asyncpp_timing
+#include <asyncpp/timing/timeout.inl>
+#endif
+
 namespace asyncpp
 {
 
@@ -23,6 +27,21 @@ namespace asyncpp
             std::forward<X_lambda>(p_lambda)));
     }
 
+    #ifdef asyncpp_timing
+    template<typename T_impl>
+    template<typename X_stream, typename X_base, typename X_ratio>
+    auto stream_base<T_impl>
+        ::timeout(X_stream&& self, const duration<X_base, X_ratio>& p_timeout)
+    {
+        using stream_type   = X_stream;
+        using timeout_type  = timing::timeout<stream_type>;
+
+        return as_stream(timeout_type(
+            std::forward<X_stream>(self),
+            p_timeout));
+    }
+    #endif
+
     /* stream */
 
     template<
@@ -52,6 +71,18 @@ namespace asyncpp
         ::for_each(X_lambda&& p_lambda)
         { return trait_type::for_each(std::move(*this), std::forward<X_lambda>(p_lambda)); }
 
+    #ifdef asyncpp_timing
+    template<
+        typename T_value,
+        typename T_impl>
+    template<
+        typename X_base,
+        typename X_ratio>
+    auto stream<T_value, T_impl>
+        ::timeout(const duration<X_base, X_ratio>& p_timeout)
+        { return trait_type::timeout(std::move(*this), p_timeout); }
+    #endif
+
     template<
         typename T_value,
         typename T_impl>
@@ -83,7 +114,7 @@ namespace asyncpp
     /* misc */
 
     template<typename X_value>
-    constexpr decltype(auto) as_stream(X_value&& value)
+    constexpr stream<X_value> as_stream(X_value&& value)
     {
         using value_type  = X_value;
         using stream_type = stream<value_type>;
@@ -91,4 +122,9 @@ namespace asyncpp
         return stream_type(std::forward<X_value>(value));
     }
 
+    template<typename T>
+    struct is_stream<stream<T>, void>
+        : public std::true_type
+        { };
+
 }
diff --git a/include/asyncpp/core/stream.pre.h b/include/asyncpp/core/stream.pre.h
index 40a8738..2a4910e 100644
--- a/include/asyncpp/core/stream.pre.h
+++ b/include/asyncpp/core/stream.pre.h
@@ -15,6 +15,11 @@ namespace asyncpp
 
         template<typename X_stream, typename X_lambda>
         static inline auto for_each(X_stream&& self, X_lambda&& p_lambda);
+
+        #ifdef asyncpp_timing
+        template<typename X_stream, typename X_base, typename X_ratio>
+        static inline auto timeout(X_stream&& self, const duration<X_base, X_ratio>& p_timeout);
+        #endif
     };
 
     template<typename T, typename = void>
@@ -25,10 +30,18 @@ namespace asyncpp
         typename T_impl = stream_trait<std::decay_t<T_object>>>
     struct stream;
 
+    template<typename T, typename = void>
+    struct is_stream
+        : public std::false_type
+        { };
+
+    template<typename T>
+    constexpr decltype(auto) is_stream_v = is_stream<T>::value;
+
     /**
      * @brief Construct a stream from the given value.
      */
     template<typename X_value>
-    constexpr decltype(auto) as_stream(X_value&& value);
+    constexpr stream<X_value> as_stream(X_value&& value);
 
 }
diff --git a/include/asyncpp/timer.h b/include/asyncpp/timer.h
deleted file mode 100644
index f8898b7..0000000
--- a/include/asyncpp/timer.h
+++ /dev/null
@@ -1,9 +0,0 @@
-#pragma once
-
-#include "timer/delay.h"
-#include "timer/interval.h"
-#include "timer/timer.h"
-
-#include "timer/delay.inl"
-#include "timer/interval.inl"
-#include "timer/timer.inl"
diff --git a/include/asyncpp/timing.h b/include/asyncpp/timing.h
new file mode 100644
index 0000000..47708b7
--- /dev/null
+++ b/include/asyncpp/timing.h
@@ -0,0 +1,13 @@
+#pragma once
+
+#define asyncpp_timing
+
+#include "timing/delay.h"
+#include "timing/interval.h"
+#include "timing/timeout.h"
+#include "timing/timer.h"
+
+#include "timing/delay.inl"
+#include "timing/interval.inl"
+#include "timing/timeout.inl"
+#include "timing/timer.inl"
diff --git a/include/asyncpp/timer/delay.h b/include/asyncpp/timing/delay.h
similarity index 88%
rename from include/asyncpp/timer/delay.h
rename to include/asyncpp/timing/delay.h
index 5b04065..116f560 100644
--- a/include/asyncpp/timer/delay.h
+++ b/include/asyncpp/timing/delay.h
@@ -1,12 +1,13 @@
 #pragma once
 
 #include <asyncpp/core/misc.h>
+#include <asyncpp/core/future.h>
 
 #include "timer.h"
 #include "delay.pre.h"
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 
     struct delay
     {
@@ -56,11 +57,11 @@ namespace timer {
 namespace asyncpp
 {
 
-    /* future_impl for timer::delay */
+    /* future_impl for timing::delay */
 
     template<>
-    struct future_trait<timer::delay, void>
-        : public future_base<future<timer::delay, void>>
+    struct future_trait<timing::delay, void>
+        : public future_base<future<timing::delay, void>>
     {
         using value_type    = void;
         using result_type   = future_result<value_type>;
diff --git a/include/asyncpp/timer/delay.inl b/include/asyncpp/timing/delay.inl
similarity index 76%
rename from include/asyncpp/timer/delay.inl
rename to include/asyncpp/timing/delay.inl
index b731218..d3b7f4f 100644
--- a/include/asyncpp/timer/delay.inl
+++ b/include/asyncpp/timing/delay.inl
@@ -2,13 +2,14 @@
 
 #include <asyncpp/core/future.pre.h>
 #include <asyncpp/core/result.pre.h>
+#include <asyncpp/core/result.inl>
 
 #include "delay.h"
 
 #include "timer.inl"
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 
     /* delay */
 
@@ -19,7 +20,7 @@ namespace timer {
 
     template<typename T_base, typename T_ratio>
     delay::delay(const duration<T_base, T_ratio>& p_duration)
-        : _deadline     (clock::now() + p_duration)
+        : _deadline     (asyncpp::now() + p_duration)
         , _registration ()
         { }
 
@@ -39,7 +40,7 @@ namespace timer {
     void delay::reset(const duration<T_base, T_ratio>& p_duration)
     {
         __impl::timer_base::unregister_resource(_registration);
-        _deadline = now() + p_duration;
+        _deadline = asyncpp::now() + p_duration;
     }
 
 } }
@@ -47,19 +48,19 @@ namespace timer {
 namespace asyncpp
 {
 
-    /* future_impl for timer::delay */
+    /* future_impl for timing::delay */
 
     template<typename X_future>
-    auto future_trait<timer::delay, void>
+    auto future_trait<timing::delay, void>
         ::poll(X_future& self)
     {
-        auto now = timer::now();
+        auto now = asyncpp::now();
 
         if (self->_deadline <= now)
             return result_type::ready();
 
         if (self->_registration.expired())
-            self->_registration = timer::__impl::timer_base::register_resource(self->_deadline);
+            self->_registration = timing::__impl::timer_base::register_resource(self->_deadline);
 
         return result_type::not_ready();
     }
diff --git a/include/asyncpp/timer/delay.pre.h b/include/asyncpp/timing/delay.pre.h
similarity index 75%
rename from include/asyncpp/timer/delay.pre.h
rename to include/asyncpp/timing/delay.pre.h
index 52daa6a..b34a62b 100644
--- a/include/asyncpp/timer/delay.pre.h
+++ b/include/asyncpp/timing/delay.pre.h
@@ -1,7 +1,7 @@
 #pragma once
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 
     struct delay;
 
diff --git a/include/asyncpp/timer/impl/registration.h b/include/asyncpp/timing/impl/registration.h
similarity index 97%
rename from include/asyncpp/timer/impl/registration.h
rename to include/asyncpp/timing/impl/registration.h
index add261f..8acbffb 100644
--- a/include/asyncpp/timer/impl/registration.h
+++ b/include/asyncpp/timing/impl/registration.h
@@ -8,7 +8,7 @@
 #include "timer_base.pre.h"
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 namespace __impl {
 
     struct registration
diff --git a/include/asyncpp/timer/impl/registration.inl b/include/asyncpp/timing/impl/registration.inl
similarity index 95%
rename from include/asyncpp/timer/impl/registration.inl
rename to include/asyncpp/timing/impl/registration.inl
index 477bdc9..0963866 100644
--- a/include/asyncpp/timer/impl/registration.inl
+++ b/include/asyncpp/timing/impl/registration.inl
@@ -3,7 +3,7 @@
 #include "registration.h"
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 namespace __impl {
 
     /* registration */
diff --git a/include/asyncpp/timer/impl/timer_base.h b/include/asyncpp/timing/impl/timer_base.h
similarity index 99%
rename from include/asyncpp/timer/impl/timer_base.h
rename to include/asyncpp/timing/impl/timer_base.h
index dc364ed..cf81212 100644
--- a/include/asyncpp/timer/impl/timer_base.h
+++ b/include/asyncpp/timing/impl/timer_base.h
@@ -11,7 +11,7 @@
 #include "registration.h"
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 namespace __impl {
 
     struct timer_base
diff --git a/include/asyncpp/timer/impl/timer_base.inl b/include/asyncpp/timing/impl/timer_base.inl
similarity index 99%
rename from include/asyncpp/timer/impl/timer_base.inl
rename to include/asyncpp/timing/impl/timer_base.inl
index af31e96..7d31710 100644
--- a/include/asyncpp/timer/impl/timer_base.inl
+++ b/include/asyncpp/timing/impl/timer_base.inl
@@ -3,7 +3,7 @@
 #include "timer_base.h"
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 namespace __impl {
 
     /* timer_base::inner_less_compare */
diff --git a/include/asyncpp/timer/impl/timer_base.pre.h b/include/asyncpp/timing/impl/timer_base.pre.h
similarity index 81%
rename from include/asyncpp/timer/impl/timer_base.pre.h
rename to include/asyncpp/timing/impl/timer_base.pre.h
index 4f11170..2f7230f 100644
--- a/include/asyncpp/timer/impl/timer_base.pre.h
+++ b/include/asyncpp/timing/impl/timer_base.pre.h
@@ -1,7 +1,7 @@
 #pragma once
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 namespace __impl {
 
     struct timer_base;
diff --git a/include/asyncpp/timer/interval.h b/include/asyncpp/timing/interval.h
similarity index 89%
rename from include/asyncpp/timer/interval.h
rename to include/asyncpp/timing/interval.h
index 10fa2b2..5ba811a 100644
--- a/include/asyncpp/timer/interval.h
+++ b/include/asyncpp/timing/interval.h
@@ -7,7 +7,7 @@
 #include "delay.h"
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 
     struct interval
     {
@@ -51,11 +51,11 @@ namespace timer {
 namespace asyncpp
 {
 
-    /* stream_trait for timer::interval */
+    /* stream_trait for timing::interval */
 
     template<>
-    struct stream_trait<timer::interval, void>
-        : public stream_base<stream<timer::interval, void>>
+    struct stream_trait<timing::interval, void>
+        : public stream_base<stream<timing::interval, void>>
     {
         using value_type    = void;
         using result_type   = stream_result<value_type>;
diff --git a/include/asyncpp/timer/interval.inl b/include/asyncpp/timing/interval.inl
similarity index 85%
rename from include/asyncpp/timer/interval.inl
rename to include/asyncpp/timing/interval.inl
index 3ea1ee4..8e57125 100644
--- a/include/asyncpp/timer/interval.inl
+++ b/include/asyncpp/timing/interval.inl
@@ -6,7 +6,7 @@
 #include "interval.h"
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 
     /* interval */
 
@@ -14,7 +14,7 @@ namespace timer {
     interval::interval(
         const asyncpp::duration<T_base, T_ratio>&   p_duration,
         time_point                                  p_deadline)
-            : interval(clock::now() + p_duration, p_duration, p_deadline)
+            : interval(asyncpp::now() + p_duration, p_duration, p_deadline)
             { }
 
     template<typename T_base, typename T_ratio>
@@ -35,11 +35,11 @@ namespace timer {
 namespace asyncpp
 {
 
-    /* stream_trait for timer::interval */
+    /* stream_trait for timing::interval */
 
     template<typename X_stream>
-    typename stream_trait<timer::interval, void>::result_type
-    stream_trait<timer::interval, void>
+    typename stream_trait<timing::interval, void>::result_type
+    stream_trait<timing::interval, void>
         ::poll(X_stream& self)
     {
         auto ret = self->_delay.poll();
diff --git a/include/asyncpp/timing/timeout.h b/include/asyncpp/timing/timeout.h
new file mode 100644
index 0000000..18f9a94
--- /dev/null
+++ b/include/asyncpp/timing/timeout.h
@@ -0,0 +1,94 @@
+#pragma once
+
+#include <cppcore/misc/exception.h>
+
+#include <asyncpp/core/misc.h>
+#include <asyncpp/core/future.h>
+#include <asyncpp/core/stream.h>
+
+#include "delay.h"
+
+namespace asyncpp {
+namespace timing {
+
+    struct timeout_exception
+        : public cppcore::exception
+    {
+        using cppcore::exception::exception;
+    };
+
+    template<typename T_inner>
+    struct timeout
+    {
+    public:
+        using inner_type        = T_inner;
+        using delay_future_type = future<delay>;
+
+    public:
+        friend struct stream_trait<timeout, void>;
+        friend struct future_trait<timeout, void>;
+
+    private:
+        inner_type          _inner;     //!< Inner future / stream.
+        clock::duration     _timeout;   //!< Timeout.
+        delay_future_type   _delay;     //!< Delay future.
+
+    public:
+        /**
+         * @brief Constructor.
+         */
+        template<typename X_inner, typename X_base, typename X_ratio>
+        inline timeout(
+            X_inner&&                           p_inner,
+            const duration<X_base, X_ratio>&    p_timeout);
+
+        /**
+         * @brief Reset the deadline.
+         */
+        template<typename X_base, typename X_ratio>
+        inline void reset(
+            const duration<X_base, X_ratio>& p_timeout);
+    };
+
+} }
+
+namespace asyncpp
+{
+
+    /* future_trait for timing::timeout */
+
+    template<typename T_inner>
+    struct future_trait<
+        timing::timeout<T_inner>,
+        std::enable_if_t<is_future_v<T_inner>>
+    >
+        : public future_base<future<timing::timeout<T_inner>, void>>
+    {
+        using inner_type    = T_inner;
+        using timeout_type  = timing::timeout<inner_type>;
+        using value_type    = typename inner_type::value_type;
+        using result_type   = typename inner_type::result_type;
+
+        template<typename X_future>
+        static inline result_type poll(X_future& self);
+    };
+
+    /* stream_trait for timing::timeout */
+
+    template<typename T_inner>
+    struct stream_trait<
+        timing::timeout<T_inner>,
+        std::enable_if_t<is_stream_v<T_inner>>
+    >
+        : public stream_base<stream<timing::timeout<T_inner>, void>>
+    {
+        using inner_type    = T_inner;
+        using timeout_type  = timing::timeout<inner_type>;
+        using value_type    = typename inner_type::value_type;
+        using result_type   = typename inner_type::result_type;
+
+        template<typename X_stream>
+        static inline result_type poll(X_stream& self);
+    };
+
+}
diff --git a/include/asyncpp/timing/timeout.inl b/include/asyncpp/timing/timeout.inl
new file mode 100644
index 0000000..62ad510
--- /dev/null
+++ b/include/asyncpp/timing/timeout.inl
@@ -0,0 +1,85 @@
+#pragma once
+
+#include <asyncpp/core/misc.inl>
+
+#include "timeout.h"
+
+namespace asyncpp {
+namespace timing {
+
+    /* timer */
+
+    template<typename T_inner>
+    template<typename X_inner, typename X_base, typename X_ratio>
+    timeout<T_inner>
+        ::timeout(
+            X_inner&&                           p_inner,
+            const duration<X_base, X_ratio>&    p_duration)
+            : _inner    (std::forward<X_inner>(p_inner))
+            , _timeout  (p_duration)
+            , _delay    (as_future(delay(asyncpp::now() + p_duration)))
+        { }
+
+    template<typename T_inner>
+    template<typename X_base, typename X_ratio>
+    void timeout<T_inner>
+        ::reset(const duration<X_base, X_ratio>& p_duration)
+    {
+        _timeout = p_duration;
+        _delay->reset(asyncpp::now() + p_duration);
+    }
+
+} }
+
+namespace asyncpp
+{
+
+    /* future_trait for timing::timeout */
+
+    template<typename T_inner>
+    template<typename X_future>
+    typename T_inner::result_type
+    future_trait<
+        timing::timeout<T_inner>,
+        std::enable_if_t<is_future_v<T_inner>>
+    >
+        ::poll(X_future& self)
+    {
+        auto r = self->_inner.poll();
+
+        if (    r.is_not_ready()
+            &&  self->_delay.poll())
+        {
+            auto new_deadline = self->_delay->deadline() + self->_timeout;
+            self->_delay->reset(new_deadline);
+            throw timing::timeout_exception();
+        }
+
+        return r;
+    }
+
+    /* stream_trait for timing::timeout */
+
+    template<typename T_inner>
+    template<typename X_stream>
+    typename T_inner::result_type
+    stream_trait<
+        timing::timeout<T_inner>,
+        std::enable_if_t<is_stream_v<T_inner>>
+    >
+        ::poll(X_stream& self)
+    {
+        auto r = self->_inner.poll();
+
+        if (    r.is_not_ready()
+            &&  self->_delay.poll())
+        {
+            auto new_deadline = self->_delay->deadline() + self->_timeout;
+            self->_delay->reset(new_deadline);
+            throw timing::timeout_exception();
+        }
+
+        return r;
+    }
+
+}
diff --git a/include/asyncpp/timer/timer.h b/include/asyncpp/timing/timer.h
similarity index 98%
rename from include/asyncpp/timer/timer.h
rename to include/asyncpp/timing/timer.h
index 1aba160..c77c416 100644
--- a/include/asyncpp/timer/timer.h
+++ b/include/asyncpp/timing/timer.h
@@ -9,7 +9,7 @@
 #include "impl/registration.h"
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 
     namespace __impl
     {
diff --git a/include/asyncpp/timer/timer.inl b/include/asyncpp/timing/timer.inl
similarity index 97%
rename from include/asyncpp/timer/timer.inl
rename to include/asyncpp/timing/timer.inl
index 8f65b37..5ec0903 100644
--- a/include/asyncpp/timer/timer.inl
+++ b/include/asyncpp/timing/timer.inl
@@ -9,7 +9,7 @@
 #include "impl/registration.inl"
 
 namespace asyncpp {
-namespace timer {
+namespace timing {
 
     namespace __impl
     {
@@ -41,7 +41,7 @@ namespace timer {
     template<typename T_inner>
     void timer<T_inner>::idle(const asyncpp::time_point * p_deadline)
     {
-        auto now = clock::now();
+        auto now = asyncpp::now();
 
         {
         auto r = _registrations.lock();
diff --git a/test/asyncpp/timer/delay_tests.cpp b/test/asyncpp/timing/delay_tests.cpp
similarity index 89%
rename from test/asyncpp/timer/delay_tests.cpp
rename to test/asyncpp/timing/delay_tests.cpp
index 5aeaa09..4f4bc5d 100644
--- a/test/asyncpp/timer/delay_tests.cpp
+++ b/test/asyncpp/timing/delay_tests.cpp
@@ -3,11 +3,11 @@
 #include "../../helper/now_mock.h"
 
 #include <asyncpp.h>
-#include <asyncpp/timer.h>
+#include <asyncpp/timing.h>
 
 using namespace ::testing;
 using namespace ::asyncpp;
-using namespace ::asyncpp::timer;
+using namespace ::asyncpp::timing;
 
 TEST(delay_tests, poll)
 {
@@ -18,7 +18,7 @@ TEST(delay_tests, poll)
         .WillOnce(Return(time_point(std::chrono::seconds( 999))))
         .WillOnce(Return(time_point(std::chrono::seconds(1000))));
 
-    asyncpp::timer::timer t;
+    asyncpp::timing::timer t;
     t.make_current();
 
     {
diff --git a/test/asyncpp/timer/interval_tests.cpp b/test/asyncpp/timing/interval_tests.cpp
similarity index 94%
rename from test/asyncpp/timer/interval_tests.cpp
rename to test/asyncpp/timing/interval_tests.cpp
index 59de013..ba57aaa 100644
--- a/test/asyncpp/timer/interval_tests.cpp
+++ b/test/asyncpp/timing/interval_tests.cpp
@@ -3,17 +3,17 @@
 #include "../../helper/now_mock.h"
 
 #include <asyncpp.h>
-#include <asyncpp/timer.h>
+#include <asyncpp/timing.h>
 
 using namespace ::testing;
 using namespace ::asyncpp;
-using namespace ::asyncpp::timer;
+using namespace ::asyncpp::timing;
 
 TEST(interval_tests, poll)
 {
     InSequence seq;
     StrictMock<now_mock> m;
-    asyncpp::timer::timer t;
+    asyncpp::timing::timer t;
     interval i(
         time_point(std::chrono::seconds(10)),
         std::chrono::seconds(5),
diff --git a/test/asyncpp/timing/timeout_tests.cpp b/test/asyncpp/timing/timeout_tests.cpp
new file mode 100644
index 0000000..aa7342b
--- /dev/null
+++ b/test/asyncpp/timing/timeout_tests.cpp
@@ -0,0 +1,202 @@
+#include <gtest/gtest.h>
+
+#include "../../helper/now_mock.h"
+
+#include <asyncpp/timing.h>
+#include <asyncpp.h>
+
+using namespace ::testing;
+using namespace ::asyncpp;
+using namespace ::asyncpp::timing;
+
+struct test
+{
+    int i;
+};
+
+namespace asyncpp
+{
+
+    template<>
+    struct future_trait<test, void>
+        : public future_base<future<test, void>>
+    {
+        using value_type    = int;
+        using result_type   = future_result<value_type>;
+
+        template<typename X_future>
+        static inline result_type poll(X_future& self)
+        {
+            return self->i
+                ? result_type::ready(self->i)
+                : result_type::not_ready();
+        }
+    };
+
+    template<>
+    struct stream_trait<test, void>
+        : public stream_base<stream<test, void>>
+    {
+        using value_type    = int;
+        using result_type   = stream_result<value_type>;
+
+        template<typename X_future>
+        static inline result_type poll(X_future& self)
+        {
+            return self->i == 0 ? result_type::not_ready() :
+                   self->i <  0 ? result_type::done() :
+                                  result_type::ready(self->i);
+        }
+    };
+
+}
+
+TEST(timeout_tests, poll_future_no_timeout)
+{
+    InSequence seq;
+    StrictMock<now_mock> m;
+
+    asyncpp::timing::timer tmr;
+    tmr.make_current();
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+
+    auto t = test { 0 };
+    auto f = as_future(t)
+        .timeout(std::chrono::seconds(5));
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+
+    auto r0 = f.poll();
+    ASSERT_FALSE(r0);
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+
+    auto r1 = f.poll();
+    ASSERT_FALSE(r1);
+
+    ++t.i;
+
+    auto r2 = f.poll();
+    ASSERT_TRUE(r2);
+}
+
+TEST(timeout_tests, poll_future_timeout)
+{
+    InSequence seq;
+    StrictMock<now_mock> m;
+
+    asyncpp::timing::timer tmr;
+    tmr.make_current();
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+
+    auto t = test { 0 };
+    auto f = as_future(t)
+        .timeout(std::chrono::seconds(5));
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+
+    auto r0 = f.poll();
+    ASSERT_FALSE(r0);
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(4))));
+
+    auto r1 = f.poll();
+    ASSERT_FALSE(r1);
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(5))));
+
+    EXPECT_THROW(f.poll(), timing::timeout_exception);
+}
+
+TEST(timeout_tests, poll_stream_no_timeout)
+{
+    InSequence seq;
+    StrictMock<now_mock> m;
+
+    asyncpp::timing::timer tmr;
+    tmr.make_current();
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+
+    auto t = test { 0 };
+    auto f = as_stream(t)
+        .timeout(std::chrono::seconds(5));
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+
+    auto r0 = f.poll();
+    ASSERT_FALSE(r0);
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+
+    auto r1 = f.poll();
+    ASSERT_FALSE(r1);
+
+    ++t.i;
+
+    auto r2 = f.poll();
+    ASSERT_TRUE(r2);
+}
+
+TEST(timeout_tests, poll_stream_timeout)
+{
+    InSequence seq;
+    StrictMock<now_mock> m;
+
+    asyncpp::timing::timer tmr;
+    tmr.make_current();
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+
+    auto t = test { 0 };
+    auto f = as_stream(t)
+        .timeout(std::chrono::seconds(5));
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+
+    auto r0 = f.poll();
+    ASSERT_FALSE(r0);
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(4))));
+
+    auto r1 = f.poll();
+    ASSERT_FALSE(r1);
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(5))));
+
+    EXPECT_THROW(f.poll(), timing::timeout_exception);
+
+    ++t.i;
+
+    auto r2 = f.poll();
+    ASSERT_TRUE(r2);
+
+    t.i = 0;
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(9))));
+
+    auto r3 = f.poll();
+    ASSERT_FALSE(r3);
+
+    EXPECT_CALL(m, now())
+        .WillOnce(Return(time_point(std::chrono::seconds(10))));
+
+    EXPECT_THROW(f.poll(), timing::timeout_exception);
+}
diff --git a/test/asyncpp/timer/timer_tests.cpp b/test/asyncpp/timing/timer_tests.cpp
similarity index 63%
rename from test/asyncpp/timer/timer_tests.cpp
rename to test/asyncpp/timing/timer_tests.cpp
index 5c1321b..87c2a11 100644
--- a/test/asyncpp/timer/timer_tests.cpp
+++ b/test/asyncpp/timing/timer_tests.cpp
@@ -4,7 +4,7 @@
 #include "../../helper/runtime_mock.h"
 
 #include <asyncpp.h>
-#include <asyncpp/timer.h>
+#include <asyncpp/timing.h>
 
 using namespace ::testing;
 using namespace ::asyncpp;
@@ -12,24 +12,24 @@ using namespace ::asyncpp;
 TEST(timer_tests, current)
 {
     {
-    timer::timer t;
+    timing::timer t;
 
-    EXPECT_EQ(nullptr, timer::__impl::timer_base::current());
+    EXPECT_EQ(nullptr, timing::__impl::timer_base::current());
 
     t.make_current();
 
-    EXPECT_EQ(&t, timer::__impl::timer_base::current());
+    EXPECT_EQ(&t, timing::__impl::timer_base::current());
 
     t.clear_current();
 
-    EXPECT_EQ(nullptr, timer::__impl::timer_base::current());
+    EXPECT_EQ(nullptr, timing::__impl::timer_base::current());
 
     t.make_current();
 
-    EXPECT_EQ(&t, timer::__impl::timer_base::current());
+    EXPECT_EQ(&t, timing::__impl::timer_base::current());
     }
 
-    EXPECT_EQ(nullptr, timer::__impl::timer_base::current());
+    EXPECT_EQ(nullptr, timing::__impl::timer_base::current());
 }
 
 TEST(timer_tests, resource_registration)
@@ -38,14 +38,14 @@ TEST(timer_tests, resource_registration)
     EXPECT_CALL(m, now)
         .WillRepeatedly(Return(time_point(std::chrono::seconds(0))));
 
-    using delay_future_type = decltype(as_future(timer::delay(std::chrono::seconds(10))));
+    using delay_future_type = decltype(as_future(timing::delay(std::chrono::seconds(10))));
 
-    timer::timer t;
+    timing::timer t;
     t.make_current();
 
-    auto f1 = std::make_unique<delay_future_type>(timer::delay(std::chrono::seconds(10)));
-    auto f2 = std::make_unique<delay_future_type>(timer::delay(std::chrono::seconds(20)));
-    auto f3 = std::make_unique<delay_future_type>(timer::delay(std::chrono::seconds(30)));
+    auto f1 = std::make_unique<delay_future_type>(timing::delay(std::chrono::seconds(10)));
+    auto f2 = std::make_unique<delay_future_type>(timing::delay(std::chrono::seconds(20)));
+    auto f3 = std::make_unique<delay_future_type>(timing::delay(std::chrono::seconds(30)));
 
     EXPECT_EQ(0, t.resource_count());
 
@@ -80,39 +80,43 @@ TEST(timer_tests, resource_registration)
 
 TEST(timer_tests, idle)
 {
-    using delay_future_type = decltype(as_future(std::declval<timer::delay>()));
+    using delay_future_type = decltype(as_future(std::declval<timing::delay>()));
 
     time_point x;
     InSequence seq;
     StrictMock<now_mock> nm;
     StrictMock<runtime_mock> rm;
-    timer::timer<StrictMock<runtime_mock>&> t(rm);
+    timing::timer<StrictMock<runtime_mock>&> t(rm);
     t.make_current();
 
+    EXPECT_CALL(nm, now);
     EXPECT_CALL(rm, idle(nullptr));
 
     t.idle(nullptr);
 
+    EXPECT_CALL(nm, now);
     EXPECT_CALL(rm, idle(Pointee(time_point(std::chrono::seconds(5)))));
 
     x = time_point(std::chrono::seconds(5));
     t.idle(&x);
 
-    EXPECT_CALL(nm, now)
-        .WillOnce(Return(time_point(std::chrono::seconds(0))));
+    EXPECT_CALL(nm, now);
 
-    auto f = std::make_unique<delay_future_type>(timer::delay(time_point(std::chrono::seconds(10))));
+    auto f = std::make_unique<delay_future_type>(timing::delay(time_point(std::chrono::seconds(10))));
     f->poll();
 
+    EXPECT_CALL(nm, now);
     EXPECT_CALL(rm, idle(Pointee(time_point(std::chrono::seconds(10)))));
 
     t.idle(nullptr);
 
+    EXPECT_CALL(nm, now);
     EXPECT_CALL(rm, idle(Pointee(time_point(std::chrono::seconds(5)))));
 
     x = time_point(std::chrono::seconds(5));
     t.idle(&x);
 
+    EXPECT_CALL(nm, now);
     EXPECT_CALL(rm, idle(Pointee(time_point(std::chrono::seconds(10)))));
 
     x = time_point(std::chrono::seconds(15));
diff --git a/test/helper/now_mock.h b/test/helper/now_mock.h
index 7c8b42f..8d1eef5 100644
--- a/test/helper/now_mock.h
+++ b/test/helper/now_mock.h
@@ -27,8 +27,8 @@ public:
     MOCK_METHOD0(now, asyncpp::time_point());
 };
 
-namespace asyncpp {
-namespace timer {
+namespace asyncpp
+{
 
     time_point now()
     {
@@ -37,4 +37,4 @@ namespace timer {
             : std::chrono::steady_clock::now();
     }
 
-} }
+}