Commit c59d29be authored by Joseph Noir's avatar Joseph Noir

Clean whitespaces

parent f9f8efc1
...@@ -37,31 +37,31 @@ class acceptor : public event_handler { ...@@ -37,31 +37,31 @@ class acceptor : public event_handler {
public: public:
/// A manager providing the `accept` member function. /// A manager providing the `accept` member function.
using manager_type = acceptor_manager; using manager_type = acceptor_manager;
/// A smart pointer to an acceptor manager. /// A smart pointer to an acceptor manager.
using manager_ptr = intrusive_ptr<manager_type>; using manager_ptr = intrusive_ptr<manager_type>;
acceptor(default_multiplexer& backend_ref, native_socket sockfd); acceptor(default_multiplexer& backend_ref, native_socket sockfd);
/// Returns the accepted socket. This member function should /// Returns the accepted socket. This member function should
/// be called only from the `new_connection` callback. /// be called only from the `new_connection` callback.
inline native_socket& accepted_socket() { inline native_socket& accepted_socket() {
return sock_; return sock_;
} }
/// Starts this acceptor, forwarding all incoming connections to /// Starts this acceptor, forwarding all incoming connections to
/// `manager`. The intrusive pointer will be released after the /// `manager`. The intrusive pointer will be released after the
/// acceptor has been closed or an IO error occured. /// acceptor has been closed or an IO error occured.
void start(acceptor_manager* mgr); void start(acceptor_manager* mgr);
/// Activates the acceptor. /// Activates the acceptor.
void activate(acceptor_manager* mgr); void activate(acceptor_manager* mgr);
/// Closes the network connection and removes this handler from its parent. /// Closes the network connection and removes this handler from its parent.
void stop_reading(); void stop_reading();
void removed_from_loop(operation op) override; void removed_from_loop(operation op) override;
protected: protected:
template <class Policy> template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) { void handle_event_impl(io::network::operation op, Policy& policy) {
...@@ -76,7 +76,7 @@ protected: ...@@ -76,7 +76,7 @@ protected:
} }
} }
} }
private: private:
manager_ptr mgr_; manager_ptr mgr_;
native_socket sock_; native_socket sock_;
......
...@@ -38,11 +38,11 @@ public: ...@@ -38,11 +38,11 @@ public:
policy_(std::forward<Ts>(xs)...) { policy_(std::forward<Ts>(xs)...) {
// nop // nop
} }
void handle_event(io::network::operation op) override { void handle_event(io::network::operation op) override {
this->handle_event_impl(op, policy_); this->handle_event_impl(op, policy_);
} }
private: private:
ProtocolPolicy policy_; ProtocolPolicy policy_;
}; };
......
...@@ -40,28 +40,28 @@ class datagram_handler : public event_handler { ...@@ -40,28 +40,28 @@ class datagram_handler : public event_handler {
public: public:
/// A smart pointer to a datagram manager. /// A smart pointer to a datagram manager.
using manager_ptr = intrusive_ptr<datagram_manager>; using manager_ptr = intrusive_ptr<datagram_manager>;
/// A buffer class providing a compatible interface to `std::vector`. /// A buffer class providing a compatible interface to `std::vector`.
using write_buffer_type = std::vector<char>; using write_buffer_type = std::vector<char>;
using read_buffer_type = network::receive_buffer; using read_buffer_type = network::receive_buffer;
/// A job for sending a datagram consisting of the sender and a buffer. /// A job for sending a datagram consisting of the sender and a buffer.
using job_type = std::pair<datagram_handle, write_buffer_type>; using job_type = std::pair<datagram_handle, write_buffer_type>;
datagram_handler(default_multiplexer& backend_ref, native_socket sockfd); datagram_handler(default_multiplexer& backend_ref, native_socket sockfd);
/// Starts reading data from the socket, forwarding incoming data to `mgr`. /// Starts reading data from the socket, forwarding incoming data to `mgr`.
void start(datagram_manager* mgr); void start(datagram_manager* mgr);
/// Activates the datagram handler. /// Activates the datagram handler.
void activate(datagram_manager* mgr); void activate(datagram_manager* mgr);
void ack_writes(bool x); void ack_writes(bool x);
/// Copies data to the write buffer. /// Copies data to the write buffer.
/// @warning Not thread safe. /// @warning Not thread safe.
void write(datagram_handle hdl, const void* buf, size_t num_bytes); void write(datagram_handle hdl, const void* buf, size_t num_bytes);
/// Returns the write buffer of this enpoint. /// Returns the write buffer of this enpoint.
/// @warning Must not be modified outside the IO multiplexers event loop /// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
...@@ -70,45 +70,45 @@ public: ...@@ -70,45 +70,45 @@ public:
wr_offline_buf_.back().first = hdl; wr_offline_buf_.back().first = hdl;
return wr_offline_buf_.back().second; return wr_offline_buf_.back().second;
} }
/// Enqueues a buffer to be sent as a datagram. /// Enqueues a buffer to be sent as a datagram.
/// @warning Must not be modified outside the IO multiplexers event loop /// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
inline void enqueue_datagram(datagram_handle hdl, std::vector<char> buf) { inline void enqueue_datagram(datagram_handle hdl, std::vector<char> buf) {
wr_offline_buf_.emplace_back(hdl, move(buf)); wr_offline_buf_.emplace_back(hdl, move(buf));
} }
/// Returns the read buffer of this stream. /// Returns the read buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop /// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
inline read_buffer_type& rd_buf() { inline read_buffer_type& rd_buf() {
return rd_buf_; return rd_buf_;
} }
/// Sends the content of the write buffer, calling the `io_failure` /// Sends the content of the write buffer, calling the `io_failure`
/// member function of `mgr` in case of an error. /// member function of `mgr` in case of an error.
/// @warning Must not be called outside the IO multiplexers event loop /// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
void flush(const manager_ptr& mgr); void flush(const manager_ptr& mgr);
/// Closes the read channel of the underlying socket and removes /// Closes the read channel of the underlying socket and removes
/// this handler from its parent. /// this handler from its parent.
void stop_reading(); void stop_reading();
void removed_from_loop(operation op) override; void removed_from_loop(operation op) override;
void add_endpoint(datagram_handle hdl, const ip_endpoint& ep, void add_endpoint(datagram_handle hdl, const ip_endpoint& ep,
const manager_ptr mgr); const manager_ptr mgr);
std::unordered_map<datagram_handle, ip_endpoint>& endpoints(); std::unordered_map<datagram_handle, ip_endpoint>& endpoints();
const std::unordered_map<datagram_handle, ip_endpoint>& endpoints() const; const std::unordered_map<datagram_handle, ip_endpoint>& endpoints() const;
void remove_endpoint(datagram_handle hdl); void remove_endpoint(datagram_handle hdl);
inline ip_endpoint& sending_endpoint() { inline ip_endpoint& sending_endpoint() {
return sender_; return sender_;
} }
protected: protected:
template <class Policy> template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) { void handle_event_impl(io::network::operation op, Policy& policy) {
...@@ -150,32 +150,32 @@ protected: ...@@ -150,32 +150,32 @@ protected:
break; break;
} }
} }
private: private:
size_t max_consecutive_reads_; size_t max_consecutive_reads_;
void prepare_next_read(); void prepare_next_read();
void prepare_next_write(); void prepare_next_write();
bool handle_read_result(bool read_result); bool handle_read_result(bool read_result);
void handle_write_result(bool write_result, datagram_handle id, void handle_write_result(bool write_result, datagram_handle id,
std::vector<char>& buf, size_t wb); std::vector<char>& buf, size_t wb);
void handle_error(); void handle_error();
// known endpoints and broker servants // known endpoints and broker servants
std::unordered_map<ip_endpoint, datagram_handle> hdl_by_ep_; std::unordered_map<ip_endpoint, datagram_handle> hdl_by_ep_;
std::unordered_map<datagram_handle, ip_endpoint> ep_by_hdl_; std::unordered_map<datagram_handle, ip_endpoint> ep_by_hdl_;
// state for reading // state for reading
const size_t max_datagram_size_; const size_t max_datagram_size_;
size_t num_bytes_; size_t num_bytes_;
read_buffer_type rd_buf_; read_buffer_type rd_buf_;
manager_ptr reader_; manager_ptr reader_;
ip_endpoint sender_; ip_endpoint sender_;
// state for writing // state for writing
int send_buffer_size_; int send_buffer_size_;
bool ack_writes_; bool ack_writes_;
......
...@@ -39,11 +39,11 @@ public: ...@@ -39,11 +39,11 @@ public:
policy_(std::forward<Ts>(xs)...) { policy_(std::forward<Ts>(xs)...) {
// nop // nop
} }
void handle_event(io::network::operation op) override { void handle_event(io::network::operation op) override {
this->handle_event_impl(op, policy_); this->handle_event_impl(op, policy_);
} }
private: private:
ProtocolPolicy policy_; ProtocolPolicy policy_;
}; };
......
...@@ -33,50 +33,50 @@ namespace network { ...@@ -33,50 +33,50 @@ namespace network {
/// Default datagram servant implementation. /// Default datagram servant implementation.
class datagram_servant_impl : public datagram_servant { class datagram_servant_impl : public datagram_servant {
using id_type = int64_t; using id_type = int64_t;
public: public:
datagram_servant_impl(default_multiplexer& mx, native_socket sockfd, datagram_servant_impl(default_multiplexer& mx, native_socket sockfd,
int64_t id); int64_t id);
bool new_endpoint(network::receive_buffer& buf) override; bool new_endpoint(network::receive_buffer& buf) override;
void ack_writes(bool enable) override; void ack_writes(bool enable) override;
std::vector<char>& wr_buf(datagram_handle hdl) override; std::vector<char>& wr_buf(datagram_handle hdl) override;
void enqueue_datagram(datagram_handle hdl, std::vector<char> buf) override; void enqueue_datagram(datagram_handle hdl, std::vector<char> buf) override;
network::receive_buffer& rd_buf() override; network::receive_buffer& rd_buf() override;
void stop_reading() override; void stop_reading() override;
void flush() override; void flush() override;
std::string addr() const override; std::string addr() const override;
uint16_t port(datagram_handle hdl) const override; uint16_t port(datagram_handle hdl) const override;
uint16_t local_port() const override; uint16_t local_port() const override;
std::vector<datagram_handle> hdls() const override; std::vector<datagram_handle> hdls() const override;
void add_endpoint(const ip_endpoint& ep, datagram_handle hdl) override; void add_endpoint(const ip_endpoint& ep, datagram_handle hdl) override;
void remove_endpoint(datagram_handle hdl) override; void remove_endpoint(datagram_handle hdl) override;
void launch() override; void launch() override;
void add_to_loop() override; void add_to_loop() override;
void remove_from_loop() override; void remove_from_loop() override;
void detach_handles() override; void detach_handles() override;
private: private:
bool launched_; bool launched_;
datagram_handler_impl<policy::udp> handler_; datagram_handler_impl<policy::udp> handler_;
}; };
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
...@@ -34,25 +34,25 @@ namespace network { ...@@ -34,25 +34,25 @@ namespace network {
class doorman_impl : public doorman { class doorman_impl : public doorman {
public: public:
doorman_impl(default_multiplexer& mx, native_socket sockfd); doorman_impl(default_multiplexer& mx, native_socket sockfd);
bool new_connection() override; bool new_connection() override;
void stop_reading() override; void stop_reading() override;
void launch() override; void launch() override;
std::string addr() const override; std::string addr() const override;
uint16_t port() const override; uint16_t port() const override;
void add_to_loop() override; void add_to_loop() override;
void remove_from_loop() override; void remove_from_loop() override;
protected: protected:
acceptor_impl<policy::tcp> acceptor_; acceptor_impl<policy::tcp> acceptor_;
}; };
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -31,61 +31,61 @@ namespace network { ...@@ -31,61 +31,61 @@ namespace network {
class event_handler { class event_handler {
public: public:
event_handler(default_multiplexer& dm, native_socket sockfd); event_handler(default_multiplexer& dm, native_socket sockfd);
virtual ~event_handler(); virtual ~event_handler();
/// Returns true once the requested operation is done, i.e., /// Returns true once the requested operation is done, i.e.,
/// to signalize the multiplexer to remove this handler. /// to signalize the multiplexer to remove this handler.
/// The handler remains in the event loop as long as it returns false. /// The handler remains in the event loop as long as it returns false.
virtual void handle_event(operation op) = 0; virtual void handle_event(operation op) = 0;
/// Callback to signalize that this handler has been removed /// Callback to signalize that this handler has been removed
/// from the event loop for operations of type `op`. /// from the event loop for operations of type `op`.
virtual void removed_from_loop(operation op) = 0; virtual void removed_from_loop(operation op) = 0;
/// Returns the native socket handle for this handler. /// Returns the native socket handle for this handler.
inline native_socket fd() const { inline native_socket fd() const {
return fd_; return fd_;
} }
/// Returns the `multiplexer` this acceptor belongs to. /// Returns the `multiplexer` this acceptor belongs to.
inline default_multiplexer& backend() { inline default_multiplexer& backend() {
return backend_; return backend_;
} }
/// Returns the bit field storing the subscribed events. /// Returns the bit field storing the subscribed events.
inline int eventbf() const { inline int eventbf() const {
return eventbf_; return eventbf_;
} }
/// Sets the bit field storing the subscribed events. /// Sets the bit field storing the subscribed events.
inline void eventbf(int value) { inline void eventbf(int value) {
eventbf_ = value; eventbf_ = value;
} }
/// Checks whether `close_read` has been called. /// Checks whether `close_read` has been called.
inline bool read_channel_closed() const { inline bool read_channel_closed() const {
return read_channel_closed_; return read_channel_closed_;
} }
/// Closes the read channel of the underlying socket. /// Closes the read channel of the underlying socket.
void close_read_channel(); void close_read_channel();
/// Removes the file descriptor from the event loop of the parent. /// Removes the file descriptor from the event loop of the parent.
void passivate(); void passivate();
protected: protected:
/// Adds the file descriptor to the event loop of the parent. /// Adds the file descriptor to the event loop of the parent.
void activate(); void activate();
void set_fd_flags(); void set_fd_flags();
int eventbf_; int eventbf_;
native_socket fd_; native_socket fd_;
bool read_channel_closed_; bool read_channel_closed_;
default_multiplexer& backend_; default_multiplexer& backend_;
}; };
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -34,34 +34,34 @@ namespace network { ...@@ -34,34 +34,34 @@ namespace network {
class scribe_impl : public scribe { class scribe_impl : public scribe {
public: public:
scribe_impl(default_multiplexer& mx, native_socket sockfd); scribe_impl(default_multiplexer& mx, native_socket sockfd);
void configure_read(receive_policy::config config) override; void configure_read(receive_policy::config config) override;
void ack_writes(bool enable) override; void ack_writes(bool enable) override;
std::vector<char>& wr_buf() override; std::vector<char>& wr_buf() override;
std::vector<char>& rd_buf() override; std::vector<char>& rd_buf() override;
void stop_reading() override; void stop_reading() override;
void flush() override; void flush() override;
std::string addr() const override; std::string addr() const override;
uint16_t port() const override; uint16_t port() const override;
void launch(); void launch();
void add_to_loop() override; void add_to_loop() override;
void remove_from_loop() override; void remove_from_loop() override;
protected: protected:
bool launched_; bool launched_;
stream_impl<policy::tcp> stream_; stream_impl<policy::tcp> stream_;
}; };
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
...@@ -41,60 +41,60 @@ class stream : public event_handler { ...@@ -41,60 +41,60 @@ class stream : public event_handler {
public: public:
/// A smart pointer to a stream manager. /// A smart pointer to a stream manager.
using manager_ptr = intrusive_ptr<stream_manager>; using manager_ptr = intrusive_ptr<stream_manager>;
/// A buffer class providing a compatible /// A buffer class providing a compatible
/// interface to `std::vector`. /// interface to `std::vector`.
using buffer_type = std::vector<char>; using buffer_type = std::vector<char>;
stream(default_multiplexer& backend_ref, native_socket sockfd); stream(default_multiplexer& backend_ref, native_socket sockfd);
/// Starts reading data from the socket, forwarding incoming data to `mgr`. /// Starts reading data from the socket, forwarding incoming data to `mgr`.
void start(stream_manager* mgr); void start(stream_manager* mgr);
/// Activates the stream. /// Activates the stream.
void activate(stream_manager* mgr); void activate(stream_manager* mgr);
/// Configures how much data will be provided for the next `consume` callback. /// Configures how much data will be provided for the next `consume` callback.
/// @warning Must not be called outside the IO multiplexers event loop /// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
void configure_read(receive_policy::config config); void configure_read(receive_policy::config config);
void ack_writes(bool x); void ack_writes(bool x);
/// Copies data to the write buffer. /// Copies data to the write buffer.
/// @warning Not thread safe. /// @warning Not thread safe.
void write(const void* buf, size_t num_bytes); void write(const void* buf, size_t num_bytes);
/// Returns the write buffer of this stream. /// Returns the write buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop /// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
inline buffer_type& wr_buf() { inline buffer_type& wr_buf() {
return wr_offline_buf_; return wr_offline_buf_;
} }
/// Returns the read buffer of this stream. /// Returns the read buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop /// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
inline buffer_type& rd_buf() { inline buffer_type& rd_buf() {
return rd_buf_; return rd_buf_;
} }
/// Sends the content of the write buffer, calling the `io_failure` /// Sends the content of the write buffer, calling the `io_failure`
/// member function of `mgr` in case of an error. /// member function of `mgr` in case of an error.
/// @warning Must not be called outside the IO multiplexers event loop /// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started. /// once the stream has been started.
void flush(const manager_ptr& mgr); void flush(const manager_ptr& mgr);
/// Closes the read channel of the underlying socket and removes /// Closes the read channel of the underlying socket and removes
/// this handler from its parent. /// this handler from its parent.
void stop_reading(); void stop_reading();
void removed_from_loop(operation op) override; void removed_from_loop(operation op) override;
/// Forces this stream to subscribe to write events if no data is in the /// Forces this stream to subscribe to write events if no data is in the
/// write buffer. /// write buffer.
void force_empty_write(const manager_ptr& mgr); void force_empty_write(const manager_ptr& mgr);
protected: protected:
template <class Policy> template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) { void handle_event_impl(io::network::operation op, Policy& policy) {
...@@ -125,20 +125,20 @@ protected: ...@@ -125,20 +125,20 @@ protected:
break; break;
} }
} }
private: private:
void prepare_next_read(); void prepare_next_read();
void prepare_next_write(); void prepare_next_write();
bool handle_read_result(rw_state read_result, size_t rb); bool handle_read_result(rw_state read_result, size_t rb);
void handle_write_result(rw_state write_result, size_t wb); void handle_write_result(rw_state write_result, size_t wb);
void handle_error_propagation(); void handle_error_propagation();
size_t max_consecutive_reads_; size_t max_consecutive_reads_;
// State for reading. // State for reading.
manager_ptr reader_; manager_ptr reader_;
size_t read_threshold_; size_t read_threshold_;
...@@ -146,7 +146,7 @@ private: ...@@ -146,7 +146,7 @@ private:
size_t max_; size_t max_;
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
buffer_type rd_buf_; buffer_type rd_buf_;
// State for writing. // State for writing.
manager_ptr writer_; manager_ptr writer_;
bool ack_writes_; bool ack_writes_;
......
...@@ -35,11 +35,11 @@ public: ...@@ -35,11 +35,11 @@ public:
policy_(std::forward<Ts>(xs)...) { policy_(std::forward<Ts>(xs)...) {
// nop // nop
} }
void handle_event(io::network::operation op) override { void handle_event(io::network::operation op) override {
this->handle_event_impl(op, policy_); this->handle_event_impl(op, policy_);
} }
private: private:
ProtocolPolicy policy_; ProtocolPolicy policy_;
}; };
......
...@@ -29,7 +29,7 @@ ...@@ -29,7 +29,7 @@
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/default_multiplexer.hpp"
namespace { namespace {
constexpr size_t receive_buffer_size = std::numeric_limits<uint16_t>::max(); constexpr size_t receive_buffer_size = std::numeric_limits<uint16_t>::max();
} // namespace anonymous } // namespace anonymous
...@@ -162,7 +162,7 @@ void datagram_handler::prepare_next_write() { ...@@ -162,7 +162,7 @@ void datagram_handler::prepare_next_write() {
wr_offline_buf_.pop_front(); wr_offline_buf_.pop_front();
} }
} }
bool datagram_handler::handle_read_result(bool read_result) { bool datagram_handler::handle_read_result(bool read_result) {
if (!read_result) { if (!read_result) {
reader_->io_failure(&backend(), operation::read); reader_->io_failure(&backend(), operation::read);
......
...@@ -26,7 +26,7 @@ ...@@ -26,7 +26,7 @@
namespace caf { namespace caf {
namespace io { namespace io {
namespace network { namespace network {
event_handler::event_handler(default_multiplexer& dm, native_socket sockfd) event_handler::event_handler(default_multiplexer& dm, native_socket sockfd)
: eventbf_(0), : eventbf_(0),
fd_(sockfd), fd_(sockfd),
......
...@@ -99,7 +99,7 @@ void stream::removed_from_loop(operation op) { ...@@ -99,7 +99,7 @@ void stream::removed_from_loop(operation op) {
case operation::propagate_error: break; case operation::propagate_error: break;
} }
} }
void stream::force_empty_write(const manager_ptr& mgr) { void stream::force_empty_write(const manager_ptr& mgr) {
if (!writing_) { if (!writing_) {
backend().add(operation::write, fd(), this); backend().add(operation::write, fd(), this);
...@@ -201,7 +201,7 @@ void stream::handle_error_propagation() { ...@@ -201,7 +201,7 @@ void stream::handle_error_propagation() {
// backend will delete this handler anyway, // backend will delete this handler anyway,
// no need to call backend().del() here // no need to call backend().del() here
} }
} // namespace network } // namespace network
} // 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