tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Remove FindChannel function

Browse Source 
Reviewers: buda, mtomic

Reviewed By: mtomic

Subscribers: pullbot

Differential Revision: https://phabricator.memgraph.io/D990
This commit is contained in:
Mislav Bradac
2017-11-16 13:53:19 +01:00
parent f16246604f
commit 651806c14f
11 changed files with 172 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/communication/raft/raft.cpp
View File

@@ -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<MLeaderTimeout>();
}
LocalChannelWriter channel(id_, "main", system_);
channel.Send<MLeaderTimeout>();
}),
heartbeat_watchdog_(
config_.heartbeat_interval, config_.heartbeat_interval,

19
src/communication/raft/raft_network.hpp
View File

@@ -1,5 +1,6 @@
#pragma once
#include <experimental/optional>
#include <mutex>
#include "communication/reactor/reactor_local.hpp"
@@ -84,11 +85,11 @@ class LocalReactorNetworkInterface : public RaftNetworkInterface {
private:
template <class TMessage>
bool SendMessage(const std::string &recipient, const TMessage &msg) {
auto channel = system_.FindChannel(recipient, "main");
if (!channel) {
return false;
}
channel->Send<TMessage>(msg);
reactor::LocalChannelWriter channel(recipient, "main", system_);
channel.Send<TMessage>(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<TMessage>(msg);
}
reactor::LocalChannelWriter channel(recipient, "main", system_);
channel.Send<TMessage>(msg);
}
return ok;

4
src/communication/reactor/common_messages.hpp
View File

@@ -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.

49
src/communication/reactor/protocol.cpp
View File

@@ -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<uint8_t *>(&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> 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);
}
}

14
src/communication/reactor/protocol.hpp
View File

@@ -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> message);
}

14
src/communication/reactor/reactor_distributed.cpp
View File

@@ -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) {}
}

54
src/communication/reactor/reactor_distributed.hpp
View File

@@ -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 <typename TMessage, typename... Args>
void Send(Args &&... args) {
Send(std::unique_ptr<Message>(
std::make_unique<TMessage>(std::forward<Args>(args)...)));
}
void Send(std::unique_ptr<Message> message) override {
std::lock_guard<SpinLock> lock(network_->mutex_);
network_->queue_.Emplace(address_, port_, reactor_, channel_,
@@ -184,11 +186,13 @@ class Network {
std::unique_ptr<ServerT> 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<Reactor> Spawn(const std::string &name,
std::function<void(Reactor &)> 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<ChannelWriter> 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<ChannelWriter> FindChannel(
const std::string &address, uint16_t port,
const std::string &reactor_name,
const std::string &channel_name) override {
return std::make_shared<Network::RemoteChannelWriter>(
&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;

43
src/communication/reactor/reactor_local.cpp
View File

@@ -25,47 +25,39 @@ std::string Channel::LocalChannelWriter::ReactorName() const {
void Channel::LocalChannelWriter::Send(std::unique_ptr<Message> 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<Channel> 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<Channel::LocalChannelWriter>(new_channel);
CHECK(t) << "t is of unexpected type";
queue_ = t->queue_;
std::shared_ptr<Channel> 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::LocalChannelWriter> 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<LocalChannelWriter>(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<void(Reactor &)> &setup, System &system2)
Reactor::Reactor(System &system, const std::string &name,
const std::function<void(Reactor &)> &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<EventStream *, std::shared_ptr<ChannelWriter>> Reactor::Open() {
} while (true);
}
std::shared_ptr<ChannelWriter> Reactor::FindChannel(
const std::string &channel_name) {
std::shared_ptr<Channel> Reactor::FindChannel(const std::string &channel_name) {
std::unique_lock<std::mutex> 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) {

87
src/communication/reactor/reactor_local.hpp
View File

@@ -63,20 +63,6 @@ class ChannelWriter {
virtual std::string Name() const = 0;
};
class ChannelFinder {
public:
virtual ~ChannelFinder() {}
// Find local channel.
virtual std::shared_ptr<ChannelWriter> FindChannel(
const std::string &reactor_name, const std::string &channel_name) = 0;
// Find remote channel.
virtual std::shared_ptr<ChannelWriter> 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 &params)
: 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<Channel> &queue,
ChannelFinder &system)
const std::string &channel_name, System &system)
: reactor_name_(reactor_name),
channel_name_(channel_name),
queue_(queue),
system_(system) {}
template <typename TMessage, typename... Args>
void Send(Args &&... args) {
Send(std::unique_ptr<Message>(
std::make_unique<TMessage>(std::forward<Args>(args)...)));
}
void Send(std::unique_ptr<Message> 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<Channel> queue_;
ChannelFinder &system_;
System &system_;
};
explicit Channel(const Params &params)
: 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<void(Reactor &)> &setup, System &system2);
Reactor(System &system, const std::string &name,
const std::function<void(Reactor &)> &setup);
~Reactor();
std::pair<EventStream *, std::shared_ptr<ChannelWriter>> Open(
const std::string &s);
std::pair<EventStream *, std::shared_ptr<ChannelWriter>> Open();
std::shared_ptr<ChannelWriter> FindChannel(const std::string &channel_name);
std::shared_ptr<Channel> 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<void(Reactor &)> 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<Reactor> Spawn(const std::string &name,
std::function<void(Reactor &)> setup,
ChannelFinder *finder = nullptr) {
if (!finder) {
finder = this;
}
std::function<void(Reactor &)> setup) {
std::unique_lock<std::mutex> lock(mutex_);
CHECK(reactors_.find(name) == reactors_.end())
<< "Reactor with name: '" << name << "' already exists.";
auto reactor = std::make_unique<Reactor>(*finder, name, setup, *this);
auto reactor = std::make_unique<Reactor>(*this, name, setup);
reactors_.emplace(name, reactor.get());
return reactor;
}
std::shared_ptr<ChannelWriter> FindChannel(
const std::string &reactor_name,
const std::string &channel_name) override {
// Next two functions shouldn't be exposed.
std::shared_ptr<Channel> Resolve(const std::string &reactor_name,
const std::string &channel_name) {
std::unique_lock<std::mutex> lock(mutex_);
auto it_reactor = reactors_.find(reactor_name);
if (it_reactor == reactors_.end())
return std::shared_ptr<ChannelWriter>(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<ChannelWriter> 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<System> and
// DistributedSystem would spawn Reactor<DistributedSystem>.
LOG(FATAL) << "Tried to resolve remote channel in local System";
}
void RemoveReactor(const std::string &name_) {
std::unique_lock<std::mutex> guard(mutex_);
auto it = reactors_.find(name_);

32
tests/unit/reactor_distributed.cpp
View File

@@ -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<Message>();
RemoteChannelWriter writer("127.0.0.1", 10000, "worker", "main", system);
writer.Send<Message>();
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<ReturnAddressMessage>(r.name(), "main");
RemoteChannelWriter writer("127.0.0.1", 10000, "worker", "main", system);
writer.Send<ReturnAddressMessage>(r.name(), "main");
r.main_.first->OnEvent<MessageInt>(
[&](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<ReturnAddressMessage>(
[&](const ReturnAddressMessage &message, const Subscription &) {
message.FindChannel(r.system_)->Send<MessageInt>(5);
RemoteChannelWriter writer(message.address(), message.port(),
message.reactor_name(),
message.channel_name(), system);
writer.Send<MessageInt>(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<RequestMessage>(r.name(), "main", 123);
RemoteChannelWriter writer("127.0.0.1", 10000, "worker", "main", system);
writer.Send<RequestMessage>(r.name(), "main", 123);
r.main_.first->OnEvent<MessageInt>(
[&](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<RequestMessage>(
[&](const RequestMessage &message, const Subscription &) {
ASSERT_EQ(message.x, 123);
message.FindChannel(r.system_)->Send<MessageInt>(779);
RemoteChannelWriter writer(message.address(), message.port(),
message.reactor_name(),
message.channel_name(), system);
writer.Send<MessageInt>(779);
r.CloseChannel("main");
});
});

88
tests/unit/reactor_local.cpp
View File

@@ -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<MessageInt>(888);
LocalChannelWriter channel_writer("worker", "main", r.system_);
channel_writer.Send<MessageInt>(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<ChannelWriter> channel_writer =
r.system_.FindChannel("worker", "main");
channel_writer->Send<MessageInt>(101);
channel_writer->Send<MessageInt>(102); // should be ignored
LocalChannelWriter channel_writer("worker", "main", r.system_);
channel_writer.Send<MessageInt>(101);
channel_writer.Send<MessageInt>(102); // should be ignored
std::this_thread::sleep_for(std::chrono::milliseconds(300));
channel_writer->Send<MessageInt>(103); // should be ignored
channel_writer->Send<MessageInt>(104); // should be ignored
channel_writer.Send<MessageInt>(103); // should be ignored
channel_writer.Send<MessageInt>(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<MessageInt>(101);
channel_writer.Send<MessageInt>(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<MessageInt>(102);
channel_writer.Send<MessageInt>(102);
std::this_thread::sleep_for(std::chrono::milliseconds(200));
// This message should recieved by new "worker" reactor.
channel_writer->Send<MessageInt>(103);
channel_writer.Send<MessageInt>(103);
r.CloseChannel("main");
});
@@ -154,7 +153,10 @@ TEST(SimpleSendTest, DuringFirstEvent) {
stream->OnEvent<MessageInt>(
[&](const Message &msg, const Subscription &subscription) {
const MessageInt &msgint = dynamic_cast<const MessageInt &>(msg);
if (msgint.x == 101) r.FindChannel("main")->Send<MessageInt>(102);
if (msgint.x == 101) {
LocalChannelWriter channel_writer("master", "main", r.system_);
channel_writer.Send<MessageInt>(102);
}
if (msgint.x == 102) {
subscription.Unsubscribe();
r.CloseChannel("main");
@@ -162,8 +164,8 @@ TEST(SimpleSendTest, DuringFirstEvent) {
}
});
std::shared_ptr<ChannelWriter> channel_writer = r.FindChannel("main");
channel_writer->Send<MessageInt>(101);
LocalChannelWriter channel_writer("master", "main", r.system_);
channel_writer.Send<MessageInt>(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<ChannelWriter> channel_writer =
r.system_.FindChannel("worker", "main");
channel_writer->Send<MessageInt>(55);
channel_writer->Send<MessageInt>(66);
channel_writer->Send<MessageInt>(77);
channel_writer->Send<MessageInt>(88);
LocalChannelWriter channel_writer("worker", "main", r.system_);
channel_writer.Send<MessageInt>(55);
channel_writer.Send<MessageInt>(66);
channel_writer.Send<MessageInt>(77);
channel_writer.Send<MessageInt>(88);
std::this_thread::sleep_for(std::chrono::milliseconds(300));
channel_writer->Send<MessageChar>('a');
channel_writer->Send<MessageChar>('b');
channel_writer->Send<MessageChar>('c');
channel_writer->Send<MessageChar>('d');
channel_writer.Send<MessageChar>('a');
channel_writer.Send<MessageChar>('b');
channel_writer.Send<MessageChar>('c');
channel_writer.Send<MessageChar>('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<ChannelWriter> channel_writer =
r.system_.FindChannel("worker", "main");
channel_writer->Send<MessageInt>(101);
channel_writer->Send<MessageChar>('a');
channel_writer->Send<MessageInt>(103);
channel_writer->Send<MessageChar>('b');
LocalChannelWriter channel_writer("worker", "main", r.system_);
channel_writer.Send<MessageInt>(101);
channel_writer.Send<MessageChar>('a');
channel_writer.Send<MessageInt>(103);
channel_writer.Send<MessageChar>('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<ChannelWriter> channel_writer =
r.system_.FindChannel("worker", "main");
channel_writer->Send<MessageInt>(55);
channel_writer->Send<MessageInt>(66);
channel_writer->Send<MessageInt>(77);
LocalChannelWriter channel_writer("worker", "main", r.system_);
channel_writer.Send<MessageInt>(55);
channel_writer.Send<MessageInt>(66);
channel_writer.Send<MessageInt>(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<ChannelWriter> channel_writer =
r.system_.FindChannel("worker", "main");
channel_writer->Send<MessageChar>('a');
channel_writer->Send<MessageInt>(55);
channel_writer->Send<MessageChar>('b');
channel_writer->Send<MessageInt>(77);
LocalChannelWriter channel_writer("worker", "main", r.system_);
channel_writer.Send<MessageChar>('a');
channel_writer.Send<MessageInt>(55);
channel_writer.Send<MessageChar>('b');
channel_writer.Send<MessageInt>(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<ChannelWriter> channel_writer =
r.system_.FindChannel("worker", "main");
LocalChannelWriter channel_writer("worker", "main", r.system_);
for (int i = 0; i < kNumTests; ++i) {
channel_writer->Send<MessageInt>(rand());
channel_writer.Send<MessageInt>(rand());
std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 5));
}
r.CloseChannel("main");