7f1c7a46cc
Summary: 1. added distributed tests, currently only one can be run (TODO) 2. usability changes: add Close() to Subscription 3. EventStream::Subscription -> Subscription 4. Add Subscription to ChainOnce 5. Added README.md conventions5 Reviewers: sasa.stanko Reviewed By: sasa.stanko Subscribers: pullbot, buda Differential Revision: https://phabricator.memgraph.io/D701
103 lines
2.9 KiB
C++
103 lines
2.9 KiB
C++
// command to run:
|
|
// gnome-terminal --tab -e './network_chat --port 10000 --minloglevel 2' --tab -e './network_chat --port 10001 --minloglevel 2'
|
|
|
|
#include "reactors_distributed.hpp"
|
|
|
|
class ChatMessage : public ReturnAddressMsg {
|
|
public:
|
|
ChatMessage() : ReturnAddressMsg(), message_("") {}
|
|
|
|
ChatMessage(std::string reactor, std::string channel, std::string message)
|
|
: ReturnAddressMsg(reactor, channel), message_(message) {}
|
|
|
|
std::string Message() const { return message_; }
|
|
|
|
template <class Archive>
|
|
void serialize(Archive &ar) {
|
|
ar(cereal::base_class<ReturnAddressMsg>(this), message_);
|
|
}
|
|
|
|
private:
|
|
std::string message_;
|
|
};
|
|
CEREAL_REGISTER_TYPE(ChatMessage);
|
|
|
|
class ChatACK : public ChatMessage {
|
|
public:
|
|
ChatACK() : ChatMessage() {}
|
|
|
|
ChatACK(std::string reactor, std::string channel, std::string message)
|
|
: ChatMessage(reactor, channel, message) {}
|
|
|
|
template <class Archive>
|
|
void serialize(Archive &ar) {
|
|
ar(cereal::base_class<ChatMessage>(this));
|
|
}
|
|
};
|
|
CEREAL_REGISTER_TYPE(ChatACK);
|
|
|
|
class ChatServer : public Reactor {
|
|
public:
|
|
ChatServer(std::string name)
|
|
: Reactor(name) {}
|
|
|
|
virtual void Run() {
|
|
std::cout << "ChatServer is active" << std::endl;
|
|
|
|
auto chat = Open("chat").first;
|
|
|
|
chat->OnEvent<ChatACK>([](const ChatACK& ack, const Subscription&) {
|
|
std::cout << "Received ACK from " << ack.Address() << ":"
|
|
<< ack.Port() << " -> '" << ack.Message() << "'"
|
|
<< std::endl;
|
|
});
|
|
|
|
chat->OnEvent<ChatMessage>([this](const ChatMessage& msg, const Subscription&) {
|
|
std::cout << "Received message from " << msg.Address() << ":"
|
|
<< msg.Port() << " -> '" << msg.Message() << "'"
|
|
<< std::endl;
|
|
auto channel = msg.GetReturnChannelWriter();
|
|
if (channel != nullptr) {
|
|
channel->Send<ChatACK>("server", "chat", msg.Message());
|
|
}
|
|
});
|
|
}
|
|
};
|
|
|
|
class ChatClient : public Reactor {
|
|
public:
|
|
ChatClient(std::string name)
|
|
: Reactor(name) {}
|
|
|
|
virtual void Run() {
|
|
std::cout << "ChatClient is active" << std::endl;
|
|
|
|
std::string address, message;
|
|
uint16_t port;
|
|
while (true) {
|
|
std::cout << "Enter IP, port and message to send." << std::endl;
|
|
std::cin >> address >> port >> message;
|
|
|
|
auto channel =
|
|
Distributed::GetInstance().network().Resolve(address, port, "server", "chat");
|
|
if (channel != nullptr) {
|
|
channel->Send<ChatMessage>("server", "chat", message);
|
|
} else {
|
|
std::cerr << "Couldn't resolve that server!" << std::endl;
|
|
}
|
|
}
|
|
}
|
|
};
|
|
|
|
int main(int argc, char *argv[]) {
|
|
gflags::ParseCommandLineFlags(&argc, &argv, true);
|
|
System& system = System::GetInstance();
|
|
Distributed &distributed = Distributed::GetInstance();
|
|
distributed.StartServices();
|
|
system.Spawn<ChatServer>("server");
|
|
system.Spawn<ChatClient>("client");
|
|
system.AwaitShutdown();
|
|
distributed.StopServices();
|
|
return 0;
|
|
}
|