Commit c2cf530b authored by Samir Halilcevic's avatar Samir Halilcevic

Rewrite caf::after example

parent fa5fd1c4
...@@ -2,58 +2,56 @@ ...@@ -2,58 +2,56 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#include <algorithm>
#include <chrono> #include <chrono>
#include <iostream> #include <iostream>
#include <random> #include <random>
using std::cout; using std::cout;
using std::endl; using std::endl;
using namespace caf; using namespace caf;
// Sends random characters to buddy, and then waits for a letter back // Sends a random number of printable characters to buddy and exits
behavior noisy_buddy(event_based_actor* self, actor buddy) { void generator(event_based_actor* self, actor buddy) {
using namespace std::chrono_literals;
std::random_device rd; std::random_device rd;
std::minstd_rand gen{rd()}; std::minstd_rand gen{rd()};
const auto count = std::uniform_int_distribution<>(20, 100)(gen); const auto count = std::uniform_int_distribution<>{20, 100}(gen);
std::uniform_int_distribution<> dis(33, 126); std::uniform_int_distribution<> dis{33, 126};
for (auto i = 0; i < count; i++) { for (auto i = 0; i < count; i++) {
self->send(buddy, static_cast<char>(dis(gen))); self->send(buddy, static_cast<char>(dis(gen)));
} }
return {[self](std::string letter) {
cout << "Received a letter:" << endl << letter << endl;
self->quit();
}};
} }
struct aggregator_state { // Collects the incoming characters until either the awaited_size of characters
std::string letter; // is received, or no new characters arrive for 100ms
caf::actor dest; behavior collector(stateful_actor<std::string>* self, size_t awaited_size) {
};
// Aggregates incoming characters and stores the sender, replies with the
// reversed string when inactive for 100ms
behavior aggregator(stateful_actor<aggregator_state>* self) {
using namespace std::chrono_literals; using namespace std::chrono_literals;
self->state.reserve(awaited_size);
return { return {
[=](char c) mutable { [=](char c) {
self->state.dest = caf::actor_cast<caf::actor>(self->current_sender()); self->state.push_back(c);
self->state.letter.push_back(c); if (self->state.size() == awaited_size) {
cout << "Received message length: " << self->state.size() << endl
<< "Message content: " << self->state << endl;
self->quit();
}
}, },
// trigger if we dont receive a message for 100ms // trigger if we dont receive a message for 100ms
caf::after(100ms) >> caf::after(100ms) >>
[=] { [self]() {
std::reverse(self->state.letter.begin(), self->state.letter.end()); cout << "Timeout reached!" << endl;
self->send(self->state.dest, self->state.letter); if (!self->state.empty()) {
cout << "bye" << endl; cout << "Received message length: " << self->state.size() << endl
<< "Message content: " << self->state << endl;
}
self->quit(); self->quit();
}}; },
};
} }
void caf_main(actor_system& system) { void caf_main(actor_system& system) {
auto agg = system.spawn(aggregator); auto col = system.spawn(collector, 60);
system.spawn(noisy_buddy, agg); system.spawn(generator, col);
} }
CAF_MAIN() CAF_MAIN()
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