Commit 914e17d6 authored by Joseph Noir's avatar Joseph Noir

Rename dgram_communicator to dgram_stream

parent 9c128526
...@@ -25,8 +25,8 @@ set (LIBCAF_IO_SRCS ...@@ -25,8 +25,8 @@ set (LIBCAF_IO_SRCS
src/uri.cpp src/uri.cpp
src/dgram_scribe.cpp src/dgram_scribe.cpp
src/dgram_doorman.cpp src/dgram_doorman.cpp
src/dgram_stream_manager.cpp
src/dgram_acceptor_manager.cpp src/dgram_acceptor_manager.cpp
src/dgram_communicator_manager.cpp
src/visitors.cpp src/visitors.cpp
# BASP files # BASP files
src/header.cpp src/header.cpp
......
...@@ -38,7 +38,7 @@ ...@@ -38,7 +38,7 @@
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/io/network/stream_manager.hpp" #include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp" #include "caf/io/network/acceptor_manager.hpp"
#include "caf/io/network/dgram_communicator_manager.hpp" #include "caf/io/network/dgram_stream_manager.hpp"
#include "caf/io/network/dgram_acceptor_manager.hpp" #include "caf/io/network/dgram_acceptor_manager.hpp"
namespace caf { namespace caf {
......
...@@ -27,7 +27,7 @@ ...@@ -27,7 +27,7 @@
#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/dgram_doorman_handle.hpp" #include "caf/io/dgram_doorman_handle.hpp"
#include "caf/io/network/dgram_communicator_manager.hpp" #include "caf/io/network/dgram_acceptor_manager.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
......
...@@ -27,12 +27,12 @@ ...@@ -27,12 +27,12 @@
#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/dgram_scribe_handle.hpp" #include "caf/io/dgram_scribe_handle.hpp"
#include "caf/io/network/dgram_acceptor_manager.hpp" #include "caf/io/network/dgram_stream_manager.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
using dgram_scribe_base = broker_servant<network::dgram_communicator_manager, using dgram_scribe_base = broker_servant<network::dgram_stream_manager,
dgram_scribe_handle, dgram_scribe_handle,
new_datagram_msg>; new_datagram_msg>;
......
...@@ -38,7 +38,7 @@ ...@@ -38,7 +38,7 @@
#include "caf/io/network/multiplexer.hpp" #include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/stream_manager.hpp" #include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp" #include "caf/io/network/acceptor_manager.hpp"
#include "caf/io/network/dgram_communicator_manager.hpp" #include "caf/io/network/dgram_stream_manager.hpp"
#include "caf/io/network/dgram_acceptor_manager.hpp" #include "caf/io/network/dgram_acceptor_manager.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
...@@ -560,20 +560,20 @@ private: ...@@ -560,20 +560,20 @@ private:
native_socket sock_; native_socket sock_;
}; };
/// A dgram_communicator is responsible for sending datagrams to an endpoint. /// A dgram_stream is responsible for sending datagrams to an endpoint.
class dgram_communicator: public event_handler { class dgram_stream: public event_handler {
public: public:
/// A manger providing the TODO /// A manger providing the TODO
using manager_type = dgram_communicator_manager; using manager_type = dgram_stream_manager;
/// A smart pointer to a datagram sink manger /// A smart pointer to a datagram sink manger
using manager_ptr = intrusive_ptr<dgram_communicator_manager>; using manager_ptr = intrusive_ptr<dgram_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>;
dgram_communicator(default_multiplexer& backend_ref, native_socket sockfd); dgram_stream(default_multiplexer& backend_ref, native_socket sockfd);
/// Configures how much buffer will be provided for the next datagram. /// Configures how much buffer will be provided for the next datagram.
/// @warning Must not be called outside the IO multiplexers event loop /// @warning Must not be called outside the IO multiplexers event loop
......
...@@ -17,8 +17,8 @@ ...@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_IO_NETWORK_DGRAM_COMMUNICATOR_MANGER_HPP #ifndef CAF_IO_NETWORK_DGRAM_STREAM_MANGER_HPP
#define CAF_IO_NETWORK_DGRAM_COMMUNICATOR_MANGER_HPP #define CAF_IO_NETWORK_DGRAM_STREAM_MANGER_HPP
#include "caf/io/network/manager.hpp" #include "caf/io/network/manager.hpp"
...@@ -28,11 +28,11 @@ namespace network { ...@@ -28,11 +28,11 @@ namespace network {
/// A datagram manager provides callbacks for outgoing /// A datagram manager provides callbacks for outgoing
/// datagrams as well as for error handling. /// datagrams as well as for error handling.
class dgram_communicator_manager : public manager { class dgram_stream_manager : public manager {
public: public:
dgram_communicator_manager(abstract_broker* ptr); dgram_stream_manager(abstract_broker* ptr);
~dgram_communicator_manager(); ~dgram_stream_manager();
/// Called by the underlying I/O device whenever it received data. /// Called by the underlying I/O device whenever it received data.
/// @returns `true` if the manager accepts further reads, otherwise `false`. /// @returns `true` if the manager accepts further reads, otherwise `false`.
...@@ -46,4 +46,4 @@ public: ...@@ -46,4 +46,4 @@ public:
} // namespace io } // namespace io
} // namespace caf } // namespace caf
#endif // CAF_IO_NETWORK_DGRAM_COMMUNICATOR_MANGER_HPP #endif // CAF_IO_NETWORK_DGRAM_STREAM_MANGER_HPP
...@@ -909,48 +909,48 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd, ...@@ -909,48 +909,48 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
impl(abstract_broker* ptr, default_multiplexer& mx, native_socket sockfd) impl(abstract_broker* ptr, default_multiplexer& mx, native_socket sockfd)
: dgram_scribe(ptr, network::dg_sink_hdl_from_socket(sockfd)), : dgram_scribe(ptr, network::dg_sink_hdl_from_socket(sockfd)),
launched_(false), launched_(false),
communicator_(mx, sockfd) { stream_(mx, sockfd) {
// nop // nop
} }
void configure_datagram_size(size_t buf_size) override { void configure_datagram_size(size_t buf_size) override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
communicator_.configure_datagram_size(buf_size); stream_.configure_datagram_size(buf_size);
// TODO: Do we need this? // TODO: Do we need this?
if (!launched_) if (!launched_)
launch(); launch();
} }
void ack_writes(bool enable) override { void ack_writes(bool enable) override {
communicator_.ack_writes(enable); stream_.ack_writes(enable);
} }
std::vector<char>& wr_buf() override { std::vector<char>& wr_buf() override {
return communicator_.wr_buf(); return stream_.wr_buf();
} }
std::vector<char>& rd_buf() override { std::vector<char>& rd_buf() override {
return communicator_.rd_buf(); return stream_.rd_buf();
} }
void stop_reading() override { void stop_reading() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
communicator_.stop_reading(); stream_.stop_reading();
detach(&communicator_.backend(), false); detach(&stream_.backend(), false);
} }
void flush() override { void flush() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
communicator_.flush(this); stream_.flush(this);
} }
std::string addr() const override { std::string addr() const override {
auto x = remote_addr_of_fd(communicator_.fd()); auto x = remote_addr_of_fd(stream_.fd());
if (!x) if (!x)
return ""; return "";
return *x; return *x;
} }
uint16_t local_port() const override { uint16_t local_port() const override {
auto x = local_port_of_fd(communicator_.fd()); auto x = local_port_of_fd(stream_.fd());
if (!x) if (!x)
return 0; return 0;
return *x; return *x;
} }
uint16_t port() const override { uint16_t port() const override {
auto x = remote_port_of_fd(communicator_.fd()); auto x = remote_port_of_fd(stream_.fd());
if (!x) if (!x)
return 0; return 0;
return *x; return *x;
...@@ -959,21 +959,21 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd, ...@@ -959,21 +959,21 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
CAF_ASSERT(!launched_); CAF_ASSERT(!launched_);
launched_ = true; launched_ = true;
communicator_.start(this); stream_.start(this);
} }
void add_to_loop() override { void add_to_loop() override {
communicator_.activate(this); stream_.activate(this);
} }
void remove_from_loop() override { void remove_from_loop() override {
communicator_.passivate(); stream_.passivate();
} }
void set_endpoint_addr(const sockaddr_storage& addr, size_t len, void set_endpoint_addr(const sockaddr_storage& addr, size_t len,
bool is_tmp_endpoint) { bool is_tmp_endpoint) {
communicator_.set_endpoint_addr(addr, len, is_tmp_endpoint); stream_.set_endpoint_addr(addr, len, is_tmp_endpoint);
} }
private: private:
bool launched_; bool launched_;
network::dgram_communicator communicator_; network::dgram_stream stream_;
}; };
auto ptr = make_counted<impl>(self, *this, fd); auto ptr = make_counted<impl>(self, *this, fd);
if (addr) if (addr)
...@@ -1576,7 +1576,7 @@ void acceptor::removed_from_loop(operation op) { ...@@ -1576,7 +1576,7 @@ void acceptor::removed_from_loop(operation op) {
mgr_.reset(); mgr_.reset();
} }
dgram_communicator::dgram_communicator(default_multiplexer& backend_ref, dgram_stream::dgram_stream(default_multiplexer& backend_ref,
native_socket sockfd) native_socket sockfd)
: event_handler(backend_ref, sockfd), : event_handler(backend_ref, sockfd),
dgram_size_(0), dgram_size_(0),
...@@ -1587,12 +1587,12 @@ dgram_communicator::dgram_communicator(default_multiplexer& backend_ref, ...@@ -1587,12 +1587,12 @@ dgram_communicator::dgram_communicator(default_multiplexer& backend_ref,
configure_datagram_size(1500); configure_datagram_size(1500);
} }
void dgram_communicator::start(manager_type* mgr) { void dgram_stream::start(manager_type* mgr) {
CAF_ASSERT(mgr != nullptr); CAF_ASSERT(mgr != nullptr);
activate(mgr); activate(mgr);
} }
void dgram_communicator::activate(manager_type* mgr) { void dgram_stream::activate(manager_type* mgr) {
if (!reader_) { if (!reader_) {
reader_.reset(mgr); reader_.reset(mgr);
event_handler::activate(); event_handler::activate();
...@@ -1600,22 +1600,22 @@ void dgram_communicator::activate(manager_type* mgr) { ...@@ -1600,22 +1600,22 @@ void dgram_communicator::activate(manager_type* mgr) {
} }
} }
void dgram_communicator::configure_datagram_size(size_t size) { void dgram_stream::configure_datagram_size(size_t size) {
dgram_size_ = size; dgram_size_ = size;
} }
void dgram_communicator::ack_writes(bool x) { void dgram_stream::ack_writes(bool x) {
ack_writes_ = x; ack_writes_ = x;
} }
void dgram_communicator::write(const void* buf, size_t num_bytes) { void dgram_stream::write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes)); CAF_LOG_TRACE(CAF_ARG(num_bytes));
auto first = reinterpret_cast<const char*>(buf); auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes; auto last = first + num_bytes;
wr_offline_buf_.insert(wr_offline_buf_.end(), std::vector<char>{first, last}); wr_offline_buf_.insert(wr_offline_buf_.end(), std::vector<char>{first, last});
} }
void dgram_communicator::flush(const manager_ptr& mgr) { void dgram_stream::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr); CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size())); CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (!wr_offline_buf_.empty() && !writing_) { if (!wr_offline_buf_.empty() && !writing_) {
...@@ -1626,13 +1626,13 @@ void dgram_communicator::flush(const manager_ptr& mgr) { ...@@ -1626,13 +1626,13 @@ void dgram_communicator::flush(const manager_ptr& mgr) {
} }
} }
void dgram_communicator::stop_reading() { void dgram_stream::stop_reading() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
close_read_channel(); close_read_channel();
passivate(); passivate();
} }
void dgram_communicator::removed_from_loop(operation op) { void dgram_stream::removed_from_loop(operation op) {
switch (op) { switch (op) {
case operation::read: reader_.reset(); break; case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break; case operation::write: writer_.reset(); break;
...@@ -1640,11 +1640,11 @@ void dgram_communicator::removed_from_loop(operation op) { ...@@ -1640,11 +1640,11 @@ void dgram_communicator::removed_from_loop(operation op) {
}; };
} }
void dgram_communicator::handle_event(operation op) { void dgram_stream::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op)); CAF_LOG_TRACE(CAF_ARG(op));
switch (op) { switch (op) {
case operation::read: { case operation::read: {
// incoming message should signify a new remote enpoint // incoming message should signify a new remote endpoint
size_t rb; size_t rb;
if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(), if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(),
sockaddr_, sockaddr_len_)) { sockaddr_, sockaddr_len_)) {
...@@ -1653,9 +1653,9 @@ void dgram_communicator::handle_event(operation op) { ...@@ -1653,9 +1653,9 @@ void dgram_communicator::handle_event(operation op) {
return; return;
} }
// currently handles the change from acceptor // currently handles the change from acceptor
// to communicator, TODO: Find a better solution // to stream, TODO: Find a better solution
// Either keep sending to the same endpoint, // Either keep sending to the same endpoint,
// Which would require some logic to determin if // Which would require some logic to determine if
// the sender is a new endpoint ... // the sender is a new endpoint ...
if (waiting_for_remote_endpoint) { if (waiting_for_remote_endpoint) {
std::tie(host_, port_) = sender_from_sockaddr(sockaddr_, sockaddr_len_); std::tie(host_, port_) = sender_from_sockaddr(sockaddr_, sockaddr_len_);
...@@ -1700,12 +1700,12 @@ void dgram_communicator::handle_event(operation op) { ...@@ -1700,12 +1700,12 @@ void dgram_communicator::handle_event(operation op) {
} }
} }
void dgram_communicator::prepare_next_write() { void dgram_stream::prepare_next_write() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size())); CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size()));
wr_buf_.clear(); wr_buf_.clear();
/* /*
while (wr_offline_buf_.size() > 0 && wr_offline_buf_.front().size() == 0) { while (wr_offline_buf_.size() > 0 && wr_offline_buf_.front().size() == 0) {
std::cerr << "Removing empy element from buffer ... " << std::endl; std::cerr << "Removing empty element from buffer ... " << std::endl;
wr_offline_buf_.pop_front(); wr_offline_buf_.pop_front();
} }
*/ */
...@@ -1718,12 +1718,12 @@ void dgram_communicator::prepare_next_write() { ...@@ -1718,12 +1718,12 @@ void dgram_communicator::prepare_next_write() {
} }
} }
void dgram_communicator::prepare_next_read() { void dgram_stream::prepare_next_read() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size())); CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size()));
rd_buf_.resize(dgram_size_); rd_buf_.resize(dgram_size_);
} }
void dgram_communicator::set_endpoint_addr(const sockaddr_storage& sa, void dgram_stream::set_endpoint_addr(const sockaddr_storage& sa,
size_t len, size_t len,
bool is_tmp_endpoint) { bool is_tmp_endpoint) {
waiting_for_remote_endpoint = is_tmp_endpoint; waiting_for_remote_endpoint = is_tmp_endpoint;
......
...@@ -17,18 +17,18 @@ ...@@ -17,18 +17,18 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/io/network/dgram_communicator_manager.hpp" #include "caf/io/network/dgram_stream_manager.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
namespace network { namespace network {
dgram_communicator_manager::dgram_communicator_manager(abstract_broker* ptr) dgram_stream_manager::dgram_stream_manager(abstract_broker* ptr)
: manager(ptr) { : manager(ptr) {
// nop // nop
} }
dgram_communicator_manager::~dgram_communicator_manager() { dgram_stream_manager::~dgram_stream_manager() {
// nop // nop
} }
......
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