From 651806c14ffd2678d626685499b3cbacc21c7e5f Mon Sep 17 00:00:00 2001 From: Mislav Bradac Date: Thu, 16 Nov 2017 13:53:19 +0100 Subject: [PATCH] Remove FindChannel function Reviewers: buda, mtomic Reviewed By: mtomic Subscribers: pullbot Differential Revision: https://phabricator.memgraph.io/D990 --- src/communication/raft/raft.cpp | 6 +- src/communication/raft/raft_network.hpp | 19 ++-- src/communication/reactor/common_messages.hpp | 4 - src/communication/reactor/protocol.cpp | 49 ++++------- src/communication/reactor/protocol.hpp | 14 +-- .../reactor/reactor_distributed.cpp | 14 +++ .../reactor/reactor_distributed.hpp | 54 +++++------- src/communication/reactor/reactor_local.cpp | 43 ++++----- src/communication/reactor/reactor_local.hpp | 87 +++++++----------- tests/unit/reactor_distributed.cpp | 32 ++++--- tests/unit/reactor_local.cpp | 88 +++++++++---------- 11 files changed, 172 insertions(+), 238 deletions(-) diff --git a/src/communication/raft/raft.cpp b/src/communication/raft/raft.cpp index 25940eaa2..19872ae50 100644 --- a/src/communication/raft/raft.cpp +++ b/src/communication/raft/raft.cpp @@ -25,10 +25,8 @@ RaftMember::RaftMember(System &system, const std::string &id, mode_(Mode::FOLLOWER), leader_watchdog_(config_.leader_timeout_min, config_.leader_timeout_max, [this]() { - auto channel = system_.FindChannel(id_, "main"); - if (channel) { - channel->Send(); - } + LocalChannelWriter channel(id_, "main", system_); + channel.Send(); }), heartbeat_watchdog_( config_.heartbeat_interval, config_.heartbeat_interval, diff --git a/src/communication/raft/raft_network.hpp b/src/communication/raft/raft_network.hpp index ec522fadb..a9b2315a5 100644 --- a/src/communication/raft/raft_network.hpp +++ b/src/communication/raft/raft_network.hpp @@ -1,5 +1,6 @@ #pragma once +#include #include #include "communication/reactor/reactor_local.hpp" @@ -84,11 +85,11 @@ class LocalReactorNetworkInterface : public RaftNetworkInterface { private: template bool SendMessage(const std::string &recipient, const TMessage &msg) { - auto channel = system_.FindChannel(recipient, "main"); - if (!channel) { - return false; - } - channel->Send(msg); + reactor::LocalChannelWriter channel(recipient, "main", system_); + channel.Send(msg); + // TODO: We always return true here even though we do not know if message + // was delievered or not. Maybe return value of functions in + // RaftNetworkInterface should be changed to be void, not bool. return true; } @@ -141,12 +142,8 @@ class FakeNetworkInterface : public RaftNetworkInterface { } if (ok) { - auto channel = system_.FindChannel(recipient, "main"); - if (!channel) { - ok = false; - } else { - channel->Send(msg); - } + reactor::LocalChannelWriter channel(recipient, "main", system_); + channel.Send(msg); } return ok; diff --git a/src/communication/reactor/common_messages.hpp b/src/communication/reactor/common_messages.hpp index ff32f1506..722ce0857 100644 --- a/src/communication/reactor/common_messages.hpp +++ b/src/communication/reactor/common_messages.hpp @@ -38,10 +38,6 @@ class ReturnAddressMessage : public Message { channel_); } - auto FindChannel(ChannelFinder &finder) const { - return finder.FindChannel(address_, port_, reactor_, channel_); - } - protected: friend class cereal::access; ReturnAddressMessage() {} // Cereal needs access to a default constructor. diff --git a/src/communication/reactor/protocol.cpp b/src/communication/reactor/protocol.cpp index 6e9d2f332..bfc092483 100644 --- a/src/communication/reactor/protocol.cpp +++ b/src/communication/reactor/protocol.cpp @@ -2,6 +2,7 @@ #include "communication/reactor/protocol.hpp" #include "communication/reactor/reactor_distributed.hpp" +#include "communication/reactor/reactor_local.hpp" #include "glog/logging.h" @@ -44,8 +45,8 @@ void Session::Execute() { DLOG(INFO) << "Reactor: " << reactor_ << "; Channel: " << channel_ << std::endl; - auto channel = system_.FindChannel(reactor_, channel_); - SendSuccess(channel != nullptr); + LocalChannelWriter channel(reactor_, channel_, system_); + SendSuccess(true); handshake_done_ = true; } @@ -65,14 +66,8 @@ void Session::Execute() { iarchive(message); buffer_.Shift(len_data); - auto channel = system_.FindChannel(reactor_, channel_); - if (channel == nullptr) { - SendSuccess(false); - return; - } - - channel->Send(std::move(message)); - SendSuccess(true); + LocalChannelWriter channel(reactor_, channel_, system_); + channel.Send(std::move(message)); } StreamBuffer Session::Allocate() { return buffer_.Allocate(); } @@ -100,15 +95,7 @@ bool SendLength(Socket &socket, SizeT length) { return socket.Write(reinterpret_cast(&length), sizeof(SizeT)); } -bool GetSuccess(Socket &socket) { - uint8_t val; - if (socket.Read(&val, 1) != 1) { - return false; - } - return val == 0x80; -} - -bool SendMessage(std::string address, uint16_t port, std::string reactor, +void SendMessage(std::string address, uint16_t port, std::string reactor, std::string channel, std::unique_ptr message) { // Initialize endpoint. Endpoint endpoint; @@ -116,7 +103,7 @@ bool SendMessage(std::string address, uint16_t port, std::string reactor, endpoint = Endpoint(address.c_str(), port); } catch (io::network::NetworkEndpointException &e) { LOG(INFO) << "Address is invalid!"; - return false; + return; } // Initialize socket. @@ -124,31 +111,28 @@ bool SendMessage(std::string address, uint16_t port, std::string reactor, if (!socket.Connect(endpoint)) { LOG(INFO) << "Couldn't connect to remote address: " << address << ":" << port; - return false; + return; } // Send data if (!SendLength(socket, reactor.size())) { LOG(INFO) << "Couldn't send reactor size!"; - return false; + return; } if (!SendLength(socket, channel.size())) { LOG(INFO) << "Couldn't send channel size!"; - return false; + return; } if (!socket.Write(reactor)) { LOG(INFO) << "Couldn't send reactor data!"; - return false; + return; } if (!socket.Write(channel)) { LOG(INFO) << "Couldn't send channel data!"; - return false; + return; } - bool success = GetSuccess(socket); - if (message == nullptr || !success) { - return success; - } + if (message == nullptr) return; // Serialize and send message std::ostringstream stream; @@ -158,14 +142,11 @@ bool SendMessage(std::string address, uint16_t port, std::string reactor, const std::string &buffer = stream.str(); if (!SendLength(socket, buffer.size())) { LOG(INFO) << "Couldn't send message size!"; - return false; + return; } if (!socket.Write(buffer)) { LOG(INFO) << "Couldn't send message data!"; - return false; + return; } - - // TODO: send message is blocking because of this. This is potential problem. - return GetSuccess(socket); } } diff --git a/src/communication/reactor/protocol.hpp b/src/communication/reactor/protocol.hpp index c333be291..77fa1faca 100644 --- a/src/communication/reactor/protocol.hpp +++ b/src/communication/reactor/protocol.hpp @@ -25,21 +25,16 @@ * Client sends: * len_reactor_name(SizeT) len_channel_name(SizeT) reactor_name channel_name * Server responds: - * 0x40 if the reactor/channel combo doesn't exist - * 0x80 if the reactor/channel combo exists + * 0x80 * * MESSAGES * * Client sends: * len_message(SizeT) cereal_encoded_binary_message - * Server responds: - * 0x40 if the reactor/channel combo doesn't exist or the message wasn't - * successfully decoded and delivered - * 0x80 if the reactor/channel combo exist and the message was successfully - * decoded and delivered * * Currently the server is implemented to handle more than one message after * the initial handshake, but the client can only send one message. + * TODO: no reason to do any sort of handshake at all. */ namespace communication::reactor { @@ -55,8 +50,6 @@ using SizeT = uint16_t; /** * Distributed Protocol Data - * - * This typically holds living data shared by all sessions. Currently empty. */ struct SessionData { System system; @@ -120,6 +113,7 @@ class Session { private: SizeT GetLength(int offset = 0); std::string GetStringAndShift(SizeT len); + // Should be renamed to SendHandshake. bool SendSuccess(bool success); bool alive_{true}; @@ -138,6 +132,6 @@ class Session { * If message is a nullptr then it only checks whether the remote reactor * and channel exist, else it returns the complete message send success. */ -bool SendMessage(std::string address, uint16_t port, std::string reactor, +void SendMessage(std::string address, uint16_t port, std::string reactor, std::string channel, std::unique_ptr message); } diff --git a/src/communication/reactor/reactor_distributed.cpp b/src/communication/reactor/reactor_distributed.cpp index 46e1e1d8f..609403104 100644 --- a/src/communication/reactor/reactor_distributed.cpp +++ b/src/communication/reactor/reactor_distributed.cpp @@ -3,3 +3,17 @@ // reactor adress can't be 0.0.0.0. DEFINE_string(reactor_address, "127.0.0.1", "Network server bind address"); DEFINE_int32(reactor_port, 10000, "Network server bind port"); + +namespace communication::reactor { + +RemoteChannelWriter::RemoteChannelWriter(const std::string &address, + uint16_t port, + const std::string &reactor, + const std::string &channel, + DistributedSystem &system) + : network_(&system.network_), + address_(address), + port_(port), + reactor_(reactor), + channel_(channel) {} +} diff --git a/src/communication/reactor/reactor_distributed.hpp b/src/communication/reactor/reactor_distributed.hpp index b4c592bd7..79b766bed 100644 --- a/src/communication/reactor/reactor_distributed.hpp +++ b/src/communication/reactor/reactor_distributed.hpp @@ -105,23 +105,25 @@ class Network { class RemoteChannelWriter : public ChannelWriter { public: - RemoteChannelWriter(Network *network, const std::string &address, - uint16_t port, const std::string &reactor, - const std::string &channel) - : network_(network), - address_(address), - port_(port), - reactor_(reactor), - channel_(channel) {} + RemoteChannelWriter(const std::string &address, uint16_t port, + const std::string &reactor, const std::string &channel, + DistributedSystem &system); + // TODO: This is wrong. We should probbly have base class Address that would + // contain everything needed to reference a channel. (address, port, + // reactor_name, channel_name) in remote reactors and (reactor_name, + // channel_name) in local reactors. virtual std::string Address() { return address_; } - virtual uint16_t Port() { return port_; } - std::string ReactorName() const override { return reactor_; } - std::string Name() const override { return channel_; } + template + void Send(Args &&... args) { + Send(std::unique_ptr( + std::make_unique(std::forward(args)...))); + } + void Send(std::unique_ptr message) override { std::lock_guard lock(network_->mutex_); network_->queue_.Emplace(address_, port_, reactor_, channel_, @@ -184,11 +186,13 @@ class Network { std::unique_ptr server_{nullptr}; }; +using RemoteChannelWriter = Network::RemoteChannelWriter; + /** * Placeholder for all functionality related to non-local communication. * E.g. resolve remote channels by memgraph node id, etc. */ -class DistributedSystem : public ChannelFinder { +class DistributedSystem { public: DistributedSystem() { network_.StartClient(4); @@ -198,7 +202,7 @@ class DistributedSystem : public ChannelFinder { // Thread safe. std::unique_ptr Spawn(const std::string &name, std::function setup) { - return system_.Spawn(name, setup, this); + return system_.Spawn(name, setup); } // Non-thread safe. @@ -209,31 +213,13 @@ class DistributedSystem : public ChannelFinder { network_.StopServer(); } - std::shared_ptr FindChannel( - const std::string &reactor_name, - const std::string &channel_name) override { - return system_.FindChannel(reactor_name, channel_name); - } - - /** - * Resolves remote channel synchronously. - * - * @return EventStream on which message will arrive once channel is resolved. - * @warning It can only be called from local Reactor. - */ - std::shared_ptr FindChannel( - const std::string &address, uint16_t port, - const std::string &reactor_name, - const std::string &channel_name) override { - return std::make_shared( - &network_, address, port, reactor_name, channel_name); - } - Network &network() { return network_; } const Network &network() const { return network_; } - private: + // Should be private Network network_; + + private: System &system_ = network_.protocol_data_.system; DistributedSystem(const DistributedSystem &) = delete; diff --git a/src/communication/reactor/reactor_local.cpp b/src/communication/reactor/reactor_local.cpp index db333271c..9bceac434 100644 --- a/src/communication/reactor/reactor_local.cpp +++ b/src/communication/reactor/reactor_local.cpp @@ -25,47 +25,39 @@ std::string Channel::LocalChannelWriter::ReactorName() const { void Channel::LocalChannelWriter::Send(std::unique_ptr m) { // Atomic, per the standard. We guarantee here that if channel exists it // will not be destroyed by the end of this function. - for (int i = 0; i < 2; ++i) { - std::shared_ptr queue = queue_.lock(); - // Check if cached queue exists and send message. - if (queue) { - queue->Push(std::move(m)); - break; - } - // If it doesn't exist. Check if there is a new channel with same name. - auto new_channel = system_.FindChannel(reactor_name_, channel_name_); - auto t = - std::dynamic_pointer_cast(new_channel); - CHECK(t) << "t is of unexpected type"; - queue_ = t->queue_; + std::shared_ptr queue = queue_.lock(); + // Check if cached queue exists and send message. + if (queue) { + queue->Push(std::move(m)); + return; + } + // If it doesn't exist. Check if there is a new channel with same name. + auto channel = system_.Resolve(reactor_name_, channel_name_); + if (channel) { + channel->Push(std::move(m)); + queue_ = channel; } } std::string Channel::LocalChannelWriter::Name() const { return channel_name_; } std::shared_ptr Channel::LockedOpenChannel() { - // TODO(zuza): fix this CHECK using this answer - // https://stackoverflow.com/questions/45507041/how-to-check-if-weak-ptr-is-empty-non-assigned - // TODO: figure out zuza's TODO. Does that mean this CHECK is kind of flaky - // or that it doesn't fail sometimes, when it should. - CHECK(!self_ptr_.expired()); return std::make_shared(reactor_name_, channel_name_, - self_ptr_, reactor_.system_); + reactor_.system_); } void Channel::Close() { reactor_.CloseChannel(channel_name_); } -Reactor::Reactor(ChannelFinder &system, const std::string &name, - const std::function &setup, System &system2) +Reactor::Reactor(System &system, const std::string &name, + const std::function &setup) : system_(system), - system2_(system2), name_(name), setup_(setup), main_(Open("main")), thread_([this] { setup_(*this); RunEventLoop(); - system2_.RemoveReactor(name_); + system_.RemoveReactor(name_); }) {} Reactor::~Reactor() { @@ -110,12 +102,11 @@ std::pair> Reactor::Open() { } while (true); } -std::shared_ptr Reactor::FindChannel( - const std::string &channel_name) { +std::shared_ptr Reactor::FindChannel(const std::string &channel_name) { std::unique_lock lock(*mutex_); auto it_channel = channels_.find(channel_name); if (it_channel == channels_.end()) return nullptr; - return it_channel->second->LockedOpenChannel(); + return it_channel->second; } void Reactor::CloseChannel(const std::string &s) { diff --git a/src/communication/reactor/reactor_local.hpp b/src/communication/reactor/reactor_local.hpp index 72455bfdd..5f509acb7 100644 --- a/src/communication/reactor/reactor_local.hpp +++ b/src/communication/reactor/reactor_local.hpp @@ -63,20 +63,6 @@ class ChannelWriter { virtual std::string Name() const = 0; }; -class ChannelFinder { - public: - virtual ~ChannelFinder() {} - - // Find local channel. - virtual std::shared_ptr FindChannel( - const std::string &reactor_name, const std::string &channel_name) = 0; - - // Find remote channel. - virtual std::shared_ptr FindChannel( - const std::string &address, uint16_t port, - const std::string &reactor_name, const std::string &channel_name) = 0; -}; - /** * Read-end of a Channel (between two reactors). */ @@ -272,14 +258,6 @@ class Channel { friend class Reactor; // to create a Params initialization object friend class EventStream::Subscription; - explicit Channel(const Params ¶ms) - : channel_name_(params.channel_name), - reactor_name_(params.reactor_name), - mutex_(params.mutex), - cvar_(params.cvar), - stream_(mutex_, this), - reactor_(params.reactor) {} - /** * LocalChannelWriter represents the channels to reactors living in the same * reactor system (write-end of the channels). @@ -290,21 +268,24 @@ class Channel { * There can be multiple LocalChannelWriters refering to the same stream if * needed. * - * It must outlive System. + * It must be outlived by System. */ class LocalChannelWriter : public ChannelWriter { public: friend class Channel; LocalChannelWriter(const std::string &reactor_name, - const std::string &channel_name, - const std::weak_ptr &queue, - ChannelFinder &system) + const std::string &channel_name, System &system) : reactor_name_(reactor_name), channel_name_(channel_name), - queue_(queue), system_(system) {} + template + void Send(Args &&... args) { + Send(std::unique_ptr( + std::make_unique(std::forward(args)...))); + } + void Send(std::unique_ptr m) override; std::string ReactorName() const override; std::string Name() const override; @@ -312,10 +293,19 @@ class Channel { private: std::string reactor_name_; std::string channel_name_; + // TODO: we shouldn't do this kind of caching inside of LocalChannelWriter. std::weak_ptr queue_; - ChannelFinder &system_; + System &system_; }; + explicit Channel(const Params ¶ms) + : channel_name_(params.channel_name), + reactor_name_(params.reactor_name), + mutex_(params.mutex), + cvar_(params.cvar), + stream_(mutex_, this), + reactor_(params.reactor) {} + /** * Implementation of the event stream. * @@ -431,14 +421,14 @@ class Reactor { friend class System; public: - Reactor(ChannelFinder &system, const std::string &name, - const std::function &setup, System &system2); + Reactor(System &system, const std::string &name, + const std::function &setup); ~Reactor(); std::pair> Open( const std::string &s); std::pair> Open(); - std::shared_ptr FindChannel(const std::string &channel_name); + std::shared_ptr FindChannel(const std::string &channel_name); /** * Close a channel by name. @@ -457,8 +447,7 @@ class Reactor { Reactor &operator=(const Reactor &other) = delete; Reactor &operator=(Reactor &&other) = default; - ChannelFinder &system_; - System &system2_; + System &system_; std::string name_; std::function setup_; @@ -498,50 +487,36 @@ class Reactor { PendingMessageInfo GetPendingMessages(); }; +using LocalChannelWriter = Channel::LocalChannelWriter; + /** * Placeholder for all reactors. * Make sure object of this class outlives all Reactors created by it. */ -class System : public ChannelFinder { +class System { public: friend class Reactor; System() = default; std::unique_ptr Spawn(const std::string &name, - std::function setup, - ChannelFinder *finder = nullptr) { - if (!finder) { - finder = this; - } + std::function setup) { std::unique_lock lock(mutex_); CHECK(reactors_.find(name) == reactors_.end()) << "Reactor with name: '" << name << "' already exists."; - auto reactor = std::make_unique(*finder, name, setup, *this); + auto reactor = std::make_unique(*this, name, setup); reactors_.emplace(name, reactor.get()); return reactor; } - std::shared_ptr FindChannel( - const std::string &reactor_name, - const std::string &channel_name) override { + // Next two functions shouldn't be exposed. + std::shared_ptr Resolve(const std::string &reactor_name, + const std::string &channel_name) { std::unique_lock lock(mutex_); auto it_reactor = reactors_.find(reactor_name); - if (it_reactor == reactors_.end()) - return std::shared_ptr(new Channel::LocalChannelWriter( - reactor_name, channel_name, {}, *this)); + if (it_reactor == reactors_.end()) return nullptr; return it_reactor->second->FindChannel(channel_name); } - std::shared_ptr FindChannel(const std::string &, uint16_t, - const std::string &, - const std::string &) override { - // TODO: This is awful design, but at this point I just want to make - // reactors work. We should templatize Reactor by system instead of dealing - // with interfaces then System would spawn Reactor and - // DistributedSystem would spawn Reactor. - LOG(FATAL) << "Tried to resolve remote channel in local System"; - } - void RemoveReactor(const std::string &name_) { std::unique_lock guard(mutex_); auto it = reactors_.find(name_); diff --git a/tests/unit/reactor_distributed.cpp b/tests/unit/reactor_distributed.cpp index 9d8fd8d6d..877c86d5b 100644 --- a/tests/unit/reactor_distributed.cpp +++ b/tests/unit/reactor_distributed.cpp @@ -66,10 +66,10 @@ TEST(SimpleTests, StartAndStopServices) { TEST(SimpleTests, SendEmptyMessage) { DistributedSystem system; - auto master = system.Spawn("master", [](Reactor &r) { + auto master = system.Spawn("master", [&](Reactor &r) { std::this_thread::sleep_for(100ms); - auto writer = r.system_.FindChannel("127.0.0.1", 10000, "worker", "main"); - writer->Send(); + RemoteChannelWriter writer("127.0.0.1", 10000, "worker", "main", system); + writer.Send(); r.CloseChannel("main"); }); @@ -96,20 +96,23 @@ TEST(SimpleTests, SendEmptyMessage) { TEST(SimpleTests, SendReturnAddressMessage) { DistributedSystem system; - auto master = system.Spawn("master", [](Reactor &r) { + auto master = system.Spawn("master", [&](Reactor &r) { std::this_thread::sleep_for(100ms); - auto writer = r.system_.FindChannel("127.0.0.1", 10000, "worker", "main"); - writer->Send(r.name(), "main"); + RemoteChannelWriter writer("127.0.0.1", 10000, "worker", "main", system); + writer.Send(r.name(), "main"); r.main_.first->OnEvent( [&](const MessageInt &message, const Subscription &) { EXPECT_EQ(message.x, 5); r.CloseChannel("main"); }); }); - auto worker = system.Spawn("worker", [](Reactor &r) { + auto worker = system.Spawn("worker", [&](Reactor &r) { r.main_.first->OnEvent( [&](const ReturnAddressMessage &message, const Subscription &) { - message.FindChannel(r.system_)->Send(5); + RemoteChannelWriter writer(message.address(), message.port(), + message.reactor_name(), + message.channel_name(), system); + writer.Send(5); r.CloseChannel("main"); }); }); @@ -128,10 +131,10 @@ TEST(SimpleTests, SendReturnAddressMessage) { TEST(SimpleTests, SendSerializableMessage) { DistributedSystem system; - auto master = system.Spawn("master", [](Reactor &r) { + auto master = system.Spawn("master", [&](Reactor &r) { std::this_thread::sleep_for(100ms); - auto writer = r.system_.FindChannel("127.0.0.1", 10000, "worker", "main"); - writer->Send(r.name(), "main", 123); + RemoteChannelWriter writer("127.0.0.1", 10000, "worker", "main", system); + writer.Send(r.name(), "main", 123); r.main_.first->OnEvent( [&](const MessageInt &message, const Subscription &) { ASSERT_EQ(message.x, 779); @@ -139,11 +142,14 @@ TEST(SimpleTests, SendSerializableMessage) { }); }); - auto worker = system.Spawn("worker", [](Reactor &r) { + auto worker = system.Spawn("worker", [&](Reactor &r) { r.main_.first->OnEvent( [&](const RequestMessage &message, const Subscription &) { ASSERT_EQ(message.x, 123); - message.FindChannel(r.system_)->Send(779); + RemoteChannelWriter writer(message.address(), message.port(), + message.reactor_name(), + message.channel_name(), system); + writer.Send(779); r.CloseChannel("main"); }); }); diff --git a/tests/unit/reactor_local.cpp b/tests/unit/reactor_local.cpp index 08f302aca..a71b45774 100644 --- a/tests/unit/reactor_local.cpp +++ b/tests/unit/reactor_local.cpp @@ -42,8 +42,8 @@ TEST(SimpleSendTest, OneCallback) { System system; auto master = system.Spawn("master", [](Reactor &r) { std::this_thread::sleep_for(100ms); - auto channel_writer = r.system_.FindChannel("worker", "main"); - channel_writer->Send(888); + LocalChannelWriter channel_writer("worker", "main", r.system_); + channel_writer.Send(888); r.CloseChannel("main"); }); @@ -69,13 +69,12 @@ TEST(SimpleSendTest, IgnoreAfterClose) { auto master = system.Spawn("master", [](Reactor &r) { std::this_thread::sleep_for(100ms); - std::shared_ptr channel_writer = - r.system_.FindChannel("worker", "main"); - channel_writer->Send(101); - channel_writer->Send(102); // should be ignored + LocalChannelWriter channel_writer("worker", "main", r.system_); + channel_writer.Send(101); + channel_writer.Send(102); // should be ignored std::this_thread::sleep_for(std::chrono::milliseconds(300)); - channel_writer->Send(103); // should be ignored - channel_writer->Send(104); // should be ignored + channel_writer.Send(103); // should be ignored + channel_writer.Send(104); // should be ignored // Write-end doesn't need to be closed because it's in RAII. r.CloseChannel("main"); }); @@ -102,16 +101,16 @@ TEST(SimpleSendTest, RecreateChannelAfterClosing) { auto master = system.Spawn("master", [](Reactor &r) { std::this_thread::sleep_for(100ms); - auto channel_writer = r.system_.FindChannel("worker", "main"); + LocalChannelWriter channel_writer("worker", "main", r.system_); // Original "worker" reactor will die after it process this message. - channel_writer->Send(101); + channel_writer.Send(101); std::this_thread::sleep_for(std::chrono::milliseconds(100)); // This message will be dropped since there is no reactor with name // "worker". - channel_writer->Send(102); + channel_writer.Send(102); std::this_thread::sleep_for(std::chrono::milliseconds(200)); // This message should recieved by new "worker" reactor. - channel_writer->Send(103); + channel_writer.Send(103); r.CloseChannel("main"); }); @@ -154,7 +153,10 @@ TEST(SimpleSendTest, DuringFirstEvent) { stream->OnEvent( [&](const Message &msg, const Subscription &subscription) { const MessageInt &msgint = dynamic_cast(msg); - if (msgint.x == 101) r.FindChannel("main")->Send(102); + if (msgint.x == 101) { + LocalChannelWriter channel_writer("master", "main", r.system_); + channel_writer.Send(102); + } if (msgint.x == 102) { subscription.Unsubscribe(); r.CloseChannel("main"); @@ -162,8 +164,8 @@ TEST(SimpleSendTest, DuringFirstEvent) { } }); - std::shared_ptr channel_writer = r.FindChannel("main"); - channel_writer->Send(101); + LocalChannelWriter channel_writer("master", "main", r.system_); + channel_writer.Send(101); }); f.wait(); @@ -184,17 +186,16 @@ TEST(MultipleSendTest, UnsubscribeService) { auto master = system.Spawn("master", [](Reactor &r) { std::this_thread::sleep_for(100ms); - std::shared_ptr channel_writer = - r.system_.FindChannel("worker", "main"); - channel_writer->Send(55); - channel_writer->Send(66); - channel_writer->Send(77); - channel_writer->Send(88); + LocalChannelWriter channel_writer("worker", "main", r.system_); + channel_writer.Send(55); + channel_writer.Send(66); + channel_writer.Send(77); + channel_writer.Send(88); std::this_thread::sleep_for(std::chrono::milliseconds(300)); - channel_writer->Send('a'); - channel_writer->Send('b'); - channel_writer->Send('c'); - channel_writer->Send('d'); + channel_writer.Send('a'); + channel_writer.Send('b'); + channel_writer.Send('c'); + channel_writer.Send('d'); r.CloseChannel("main"); }); @@ -238,13 +239,11 @@ TEST(MultipleSendTest, OnEvent) { System system; auto master = system.Spawn("master", [](Reactor &r) { std::this_thread::sleep_for(100ms); - std::shared_ptr channel_writer = - r.system_.FindChannel("worker", "main"); - - channel_writer->Send(101); - channel_writer->Send('a'); - channel_writer->Send(103); - channel_writer->Send('b'); + LocalChannelWriter channel_writer("worker", "main", r.system_); + channel_writer.Send(101); + channel_writer.Send('a'); + channel_writer.Send(103); + channel_writer.Send('b'); r.CloseChannel("main"); }); @@ -287,11 +286,10 @@ TEST(MultipleSendTest, Chaining) { auto master = system.Spawn("master", [](Reactor &r) { std::this_thread::sleep_for(100ms); - std::shared_ptr channel_writer = - r.system_.FindChannel("worker", "main"); - channel_writer->Send(55); - channel_writer->Send(66); - channel_writer->Send(77); + LocalChannelWriter channel_writer("worker", "main", r.system_); + channel_writer.Send(55); + channel_writer.Send(66); + channel_writer.Send(77); r.CloseChannel("main"); }); @@ -329,12 +327,11 @@ TEST(MultipleSendTest, ChainingInRightOrder) { auto master = system.Spawn("master", [](Reactor &r) { std::this_thread::sleep_for(100ms); - std::shared_ptr channel_writer = - r.system_.FindChannel("worker", "main"); - channel_writer->Send('a'); - channel_writer->Send(55); - channel_writer->Send('b'); - channel_writer->Send(77); + LocalChannelWriter channel_writer("worker", "main", r.system_); + channel_writer.Send('a'); + channel_writer.Send(55); + channel_writer.Send('b'); + channel_writer.Send(77); r.CloseChannel("main"); }); @@ -371,11 +368,10 @@ TEST(MultipleSendTest, ProcessManyMessages) { auto master = system.Spawn("master", [](Reactor &r) { std::this_thread::sleep_for(100ms); - std::shared_ptr channel_writer = - r.system_.FindChannel("worker", "main"); + LocalChannelWriter channel_writer("worker", "main", r.system_); for (int i = 0; i < kNumTests; ++i) { - channel_writer->Send(rand()); + channel_writer.Send(rand()); std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 5)); } r.CloseChannel("main");