Commit cb6e7d4d authored by Joseph Noir's avatar Joseph Noir

Fix circular include and add missing message types

parent ae46af0e
...@@ -29,9 +29,7 @@ ...@@ -29,9 +29,7 @@
#include "caf/io/fwd.hpp" #include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp" #include "caf/io/accept_handle.hpp"
#include "caf/io/datagram_sink.hpp"
#include "caf/io/receive_policy.hpp" #include "caf/io/receive_policy.hpp"
#include "caf/io/datagram_source.hpp"
#include "caf/io/system_messages.hpp" #include "caf/io/system_messages.hpp"
#include "caf/io/connection_handle.hpp" #include "caf/io/connection_handle.hpp"
#include "caf/io/datagram_sink_handle.hpp" #include "caf/io/datagram_sink_handle.hpp"
...@@ -88,6 +86,8 @@ public: ...@@ -88,6 +86,8 @@ public:
// even brokers need friends // even brokers need friends
friend class scribe; friend class scribe;
friend class doorman; friend class doorman;
friend class datagram_sink;
friend class datagram_source;
// -- overridden modifiers of abstract_actor --------------------------------- // -- overridden modifiers of abstract_actor ---------------------------------
...@@ -170,7 +170,8 @@ public: ...@@ -170,7 +170,8 @@ public:
/// Tries to connect to `host` on given `port` and creates /// Tries to connect to `host` on given `port` and creates
/// a new scribe describing the connection afterwards. /// a new scribe describing the connection afterwards.
/// @returns The handle of the new `scribe` on success. /// @returns The handle of the new `scribe` on success.
expected<connection_handle> add_tcp_scribe(const std::string& host, uint16_t port); expected<connection_handle> add_tcp_scribe(const std::string& host,
uint16_t port);
/// Assigns a detached `scribe` instance identified by `hdl` /// Assigns a detached `scribe` instance identified by `hdl`
/// from the `multiplexer` to this broker. /// from the `multiplexer` to this broker.
...@@ -291,10 +292,12 @@ protected: ...@@ -291,10 +292,12 @@ protected:
return scribes_; return scribes_;
} }
// meta programming utility
inline datagram_sink_map& get_map(datagram_sink_handle) { inline datagram_sink_map& get_map(datagram_sink_handle) {
return datagram_sinks_; return datagram_sinks_;
} }
// meta programming utility
inline datagram_source_map& get_map(datagram_source_handle) { inline datagram_source_map& get_map(datagram_source_handle) {
return datagram_sources_; return datagram_sources_;
} }
...@@ -315,7 +318,8 @@ protected: ...@@ -315,7 +318,8 @@ protected:
/// Returns the `multiplexer` running this broker. /// Returns the `multiplexer` running this broker.
network::multiplexer& backend(); network::multiplexer& backend();
/// Returns a `scribe` or `doorman` identified by `hdl`. /// Returns a `scribe`, `doorman`, `datagram_sink` or `datagram_source`
/// identified by `hdl`.
template <class Handle> template <class Handle>
auto by_id(Handle hdl) -> optional<decltype(*ptr_of(hdl))> { auto by_id(Handle hdl) -> optional<decltype(*ptr_of(hdl))> {
auto& elements = get_map(hdl); auto& elements = get_map(hdl);
...@@ -325,8 +329,8 @@ protected: ...@@ -325,8 +329,8 @@ protected:
return *(i->second); return *(i->second);
} }
/// Returns an intrusive pointer to a `scribe` or `doorman` /// Returns an intrusive pointer to a `scribe`, `doorman`, `datagram_sink`
/// identified by `hdl` and remove it from this broker. /// or `datagram_source` identified by `hdl` and remove it from this broker.
template <class Handle> template <class Handle>
auto take(Handle hdl) -> decltype(ptr_of(hdl)) { auto take(Handle hdl) -> decltype(ptr_of(hdl)) {
using std::swap; using std::swap;
......
...@@ -30,7 +30,9 @@ ...@@ -30,7 +30,9 @@
#include "caf/io/scribe.hpp" #include "caf/io/scribe.hpp"
#include "caf/io/doorman.hpp" #include "caf/io/doorman.hpp"
#include "caf/io/datagram_sink.hpp"
#include "caf/io/abstract_broker.hpp" #include "caf/io/abstract_broker.hpp"
#include "caf/io/datagram_source.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
......
...@@ -102,13 +102,21 @@ protected: ...@@ -102,13 +102,21 @@ protected:
typename std::conditional< typename std::conditional<
std::is_same<Handle, connection_handle>::value, std::is_same<Handle, connection_handle>::value,
connection_passivated_msg, connection_passivated_msg,
acceptor_passivated_msg typename std::conditional<
std::is_same<Handle, accept_handle>::value,
acceptor_passivated_msg,
typename std::conditional<
std::is_same<Handle, datagram_sink_handle>::value,
datagram_sink_passivated_msg,
datagram_source_passivated_msg
>::type
>::type
>::type; >::type;
using tmp_t = mailbox_element_vals<passiv_t>; using tmp_t = mailbox_element_vals<passiv_t>;
tmp_t tmp{strong_actor_ptr{}, message_id::make(), tmp_t tmp{strong_actor_ptr{}, message_id::make(),
mailbox_element::forwarding_stack{}, passiv_t{hdl()}}; mailbox_element::forwarding_stack{}, passiv_t{hdl()}};
invoke_mailbox_element_impl(ctx, tmp); invoke_mailbox_element_impl(ctx, tmp);
return activity_tokens_ != size_t{0}; return activity_tokens_ != size_t{0};
} }
return true; return true;
} }
......
...@@ -20,6 +20,10 @@ ...@@ -20,6 +20,10 @@
#ifndef CAF_IO_DATAGRAM_SINK_HPP #ifndef CAF_IO_DATAGRAM_SINK_HPP
#define CAF_IO_DATAGRAM_SINK_HPP #define CAF_IO_DATAGRAM_SINK_HPP
#include <vector>
#include "caf/message.hpp"
#include "caf/io/broker_servant.hpp" #include "caf/io/broker_servant.hpp"
#include "caf/io/system_messages.hpp" #include "caf/io/system_messages.hpp"
#include "caf/io/datagram_sink_handle.hpp" #include "caf/io/datagram_sink_handle.hpp"
...@@ -32,7 +36,7 @@ using datagram_sink_base = broker_servant<network::datagram_sink_manager, ...@@ -32,7 +36,7 @@ using datagram_sink_base = broker_servant<network::datagram_sink_manager,
datagram_sink_handle, datagram_sink_handle,
datagram_sent_msg>; datagram_sent_msg>;
/// Manages writing to a datagram sink /// Manages writing to a datagram sink.
/// @ingroup Broker /// @ingroup Broker
class datagram_sink : public datagram_sink_base { class datagram_sink : public datagram_sink_base {
public: public:
...@@ -40,6 +44,9 @@ public: ...@@ -40,6 +44,9 @@ public:
~datagram_sink(); ~datagram_sink();
/// Enables or disables write notifications.
virtual void ack_writes(bool enable) = 0;
/// Returns the current write buffer. /// Returns the current write buffer.
virtual std::vector<char>& wr_buf() = 0; virtual std::vector<char>& wr_buf() = 0;
......
...@@ -20,6 +20,10 @@ ...@@ -20,6 +20,10 @@
#ifndef CAF_IO_DATAGRAM_SOURCE_HPP #ifndef CAF_IO_DATAGRAM_SOURCE_HPP
#define CAF_IO_DATAGRAM_SOURCE_HPP #define CAF_IO_DATAGRAM_SOURCE_HPP
#include <vector>
#include "caf/message.hpp"
#include "caf/io/broker_servant.hpp" #include "caf/io/broker_servant.hpp"
#include "caf/io/system_messages.hpp" #include "caf/io/system_messages.hpp"
#include "caf/io/datagram_source_handle.hpp" #include "caf/io/datagram_source_handle.hpp"
......
...@@ -28,8 +28,10 @@ class broker; ...@@ -28,8 +28,10 @@ class broker;
class doorman; class doorman;
class middleman; class middleman;
class basp_broker; class basp_broker;
class datagram_sink;
class receive_policy; class receive_policy;
class abstract_broker; class abstract_broker;
class datagram_source;
template <class... Sigs> template <class... Sigs>
class typed_broker; class typed_broker;
......
...@@ -180,6 +180,30 @@ typename Inspector::result_type inspect(Inspector& f, new_datagram_msg& x) { ...@@ -180,6 +180,30 @@ typename Inspector::result_type inspect(Inspector& f, new_datagram_msg& x) {
return f(meta::type_name("new_datagram_msg"), x.handle, x.buf); return f(meta::type_name("new_datagram_msg"), x.handle, x.buf);
} }
/// Signalizes that a datagram sink has entered passive mode.
struct datagram_sink_passivated_msg {
datagram_sink_handle handle;
};
/// @relates datagram_sink_passivated_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, datagram_sink_passivated_msg& x) {
return f(meta::type_name("datagram_sink_passivated_msg"), x.handle);
}
/// Signalizes that a datagram source has entered passive mode.
struct datagram_source_passivated_msg {
datagram_source_handle handle;
};
/// @relates datagram_source_passivated_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, datagram_source_passivated_msg& x) {
return f(meta::type_name("datagram_source_passivated_msg"), x.handle);
}
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
...@@ -123,8 +123,8 @@ void abstract_broker::add_scribe(const intrusive_ptr<scribe>& ptr) { ...@@ -123,8 +123,8 @@ void abstract_broker::add_scribe(const intrusive_ptr<scribe>& ptr) {
scribes_.emplace(ptr->hdl(), ptr); scribes_.emplace(ptr->hdl(), ptr);
} }
expected<connection_handle> abstract_broker::add_tcp_scribe(const std::string& hostname, expected<connection_handle>
uint16_t port) { abstract_broker::add_tcp_scribe(const std::string& hostname, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hostname) << ", " << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(hostname) << ", " << CAF_ARG(port));
return backend().add_tcp_scribe(this, hostname, port); return backend().add_tcp_scribe(this, hostname, port);
} }
......
...@@ -34,7 +34,7 @@ datagram_sink::~datagram_sink() { ...@@ -34,7 +34,7 @@ datagram_sink::~datagram_sink() {
} }
message datagram_sink::detach_message() { message datagram_sink::detach_message() {
return make_message(endpoint_closed_msg{hdl()}); return make_message(datagram_sink_closed_msg{hdl()});
} }
void datagram_sink::datagram_sent(execution_unit* ctx, size_t written) { void datagram_sink::datagram_sent(execution_unit* ctx, size_t written) {
......
...@@ -29,13 +29,13 @@ datagram_source::datagram_source(abstract_broker* parent, ...@@ -29,13 +29,13 @@ datagram_source::datagram_source(abstract_broker* parent,
: datagram_source_base(parent, hdl) { : datagram_source_base(parent, hdl) {
// nop // nop
} }
/*
datagram_source::~datagram_source() { datagram_source::~datagram_source() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
} }
message datagram_source::detach_message() { message datagram_source::detach_message() {
return make_message(endpoint_closed_msg{hdl()}); return make_message(datagram_source_closed_msg{hdl()});
} }
bool datagram_source::consume(execution_unit* ctx, const void*, bool datagram_source::consume(execution_unit* ctx, const void*,
...@@ -70,6 +70,6 @@ void datagram_source::io_failure(execution_unit* ctx, network::operation op) { ...@@ -70,6 +70,6 @@ void datagram_source::io_failure(execution_unit* ctx, network::operation op) {
static_cast<void>(op); static_cast<void>(op);
detach(ctx, true); detach(ctx, true);
} }
*/
} // namespace io } // namespace io
} // namespace caf } // namespace caf
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