Commit c90fe5e4 authored by Jakob Otto's avatar Jakob Otto

Add timeout management

parent e53ff78b
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <memory> #include <memory>
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/actor_clock.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp" #include "caf/intrusive/drr_queue.hpp"
...@@ -137,6 +138,9 @@ public: ...@@ -137,6 +138,9 @@ public:
/// Enqueues a message to the endpoint. /// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, std::vector<byte> payload); void enqueue(mailbox_element_ptr msg, std::vector<byte> payload);
/// Enqueues a timeout to the endpoint.
void enqueue(timeout_msg msg);
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
/// Initializes the manager before adding it to the multiplexer's event loop. /// Initializes the manager before adding it to the multiplexer's event loop.
...@@ -154,6 +158,9 @@ protected: ...@@ -154,6 +158,9 @@ protected:
/// Stores outbound messages. /// Stores outbound messages.
message_queue_type messages_; message_queue_type messages_;
/// Stores a proxy for interacting with the actor clock.
actor timeout_proxy_;
}; };
using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>; using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>;
......
...@@ -52,6 +52,17 @@ public: ...@@ -52,6 +52,17 @@ public:
return transport_; return transport_;
} }
// -- timeout management -----------------------------------------------------
template <class... Ts>
uint64_t set_timeout(actor_clock::time_point tp, atom_value type,
Ts&&... xs) {
auto act = actor_cast<abstract_actor*>(timeout_proxy_);
CAF_ASSERT(act != nullptr);
sys_.clock().set_multi_timeout(tp, act, type, next_timeout_id_);
transport_.set_timeout(next_timeout_id_, std::forward<Ts>(xs)...);
return next_timeout_id_++;
}
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
error init() override { error init() override {
...@@ -94,6 +105,9 @@ public: ...@@ -94,6 +105,9 @@ public:
private: private:
transport_type transport_; transport_type transport_;
/// Stores the id for the next timeout.
uint64_t next_timeout_id_;
}; };
} // namespace net } // namespace net
......
...@@ -33,7 +33,7 @@ ...@@ -33,7 +33,7 @@
namespace caf { namespace caf {
namespace net { namespace net {
/// Implements a scribe policy that manages a stream socket. /// Implements a stream_transport that manages a stream socket.
template <class Application> template <class Application>
class stream_transport { class stream_transport {
public: public:
...@@ -65,8 +65,7 @@ public: ...@@ -65,8 +65,7 @@ public:
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
auto decorator = make_write_packet_decorator(*this, parent); worker_.init(parent);
worker_.init(decorator);
parent.mask_add(operation::read); parent.mask_add(operation::read);
return none; return none;
} }
...@@ -115,10 +114,14 @@ public: ...@@ -115,10 +114,14 @@ public:
worker_.resolve(parent, path, listener); worker_.resolve(parent, path, listener);
} }
template <class... Ts>
uint64_t set_timeout(uint64_t, Ts&&...) {
// nop
}
template <class Parent> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void timeout(Parent& parent, atom_value value, uint64_t id) {
auto decorator = make_write_packet_decorator(*this, parent); worker_.timeout(parent, value, id);
worker_.timeout(decorator, value, id);
} }
void handle_error(sec code) { void handle_error(sec code) {
...@@ -167,6 +170,8 @@ public: ...@@ -167,6 +170,8 @@ public:
} }
private: private:
// -- private member functions -----------------------------------------------
bool write_some() { bool write_some() {
if (write_buf_.empty()) if (write_buf_.empty())
return false; return false;
...@@ -193,7 +198,6 @@ private: ...@@ -193,7 +198,6 @@ private:
} }
transport_worker<application_type> worker_; transport_worker<application_type> worker_;
stream_socket handle_; stream_socket handle_;
std::vector<byte> read_buf_; std::vector<byte> read_buf_;
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment