<commit_before>#include <thread>
#include <string>
#include <cstring>
#include <sstream>
#include <iostream>
#include <functional>

#include "test.hpp"
#include "ping_pong.hpp"

#include "caf/all.hpp"
#include "caf/io/all.hpp"

#include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp"

using namespace std;
using namespace caf;

namespace {

using string_pair = std::pair<std::string, std::string>;

using actor_vector = vector<actor>;

void reflector(event_based_actor* self) {
    self->become(others() >> [=] {
        CAF_PRINT("reflect and quit");
        self->quit();
        return self->last_dequeued();
    });
}

void spawn5_server_impl(event_based_actor* self, actor client, group grp) {
    CAF_LOGF_TRACE(CAF_TARG(client, to_string) << ", "
                                                 << CAF_TARG(grp, to_string));
    CAF_CHECK(grp != invalid_group);
    self->spawn_in_group(grp, reflector);
    self->spawn_in_group(grp, reflector);
    CAF_PRINT("send {'Spawn5'} and await {'ok', actor_vector}");
    self->sync_send(client, atom("Spawn5"), grp)
        .then(on(atom("ok"), arg_match) >> [=](const actor_vector& vec) {
                  CAF_PRINT("received vector with " << vec.size()
                                                     << " elements");
                  self->send(grp, "Hello reflectors!", 5.0);
                  if (vec.size() != 5) {
                      CAF_PRINTERR(
                          "remote client did not spawn five reflectors!");
                  }
                  for (auto& a : vec) {
                      CAF_PRINT("monitor actor: " << to_string(a));
                      self->monitor(a);
                  }
              },
              others() >> [=] {
                  CAF_UNEXPECTED_MSG(self);
                  self->quit(exit_reason::unhandled_exception);
              },
              after(chrono::seconds(10)) >> [=] {
             CAF_UNEXPECTED_TOUT();
             self->quit(exit_reason::unhandled_exception);
         })
        .continue_with([=] {
             CAF_PRINT("wait for reflected messages");
             // receive seven reply messages (2 local, 5 remote)
             auto replies = std::make_shared<int>(0);
             self->become(
                 on("Hello reflectors!", 5.0) >> [=] {
                     if (++*replies == 7) {
                         CAF_PRINT("wait for DOWN messages");
                         auto downs = std::make_shared<int>(0);
                         self->become(
                             [=](const down_msg& dm) {
                                 if (dm.reason != exit_reason::normal) {
                                     CAF_PRINTERR(
                                         "reflector exited for non-normal exit "
                                         "reason!");
                                 }
                                 if (++*downs == 5) {
                                     CAF_CHECKPOINT();
                                     self->send(client, atom("Spawn5Done"));
                                     self->quit();
                                 }
                             },
                             others() >> [=] {
                                 CAF_UNEXPECTED_MSG(self);
                                 // self->quit(exit_reason::unhandled_exception);
                             },
                             after(chrono::seconds(2)) >> [=] {
                                 CAF_UNEXPECTED_TOUT();
                                 CAF_LOGF_ERROR("did only receive "
                                                 << *downs << " down messages");
                                 // self->quit(exit_reason::unhandled_exception);
                             });
                     }
                 },
                 after(std::chrono::seconds(2)) >> [=] {
                     CAF_UNEXPECTED_TOUT();
                     CAF_LOGF_ERROR("did only receive "
                                     << *replies
                                     << " responses to 'Hello reflectors!'");
                     // self->quit(exit_reason::unhandled_exception);
                 });
         });
}

// receive seven reply messages (2 local, 5 remote)
void spawn5_server(event_based_actor* self, actor client, bool inverted) {
    if (!inverted)
        spawn5_server_impl(self, client, group::get("local", "foobar"));
    else {
        CAF_PRINT("request group");
        self->sync_send(client, atom("GetGroup"))
            .then([=](const group& remote_group) {
                 spawn5_server_impl(self, client, remote_group);
             });
    }
}

void spawn5_client(event_based_actor* self) {
    self->become(
        on(atom("GetGroup")) >> []()->group {
            CAF_PRINT("received {'GetGroup'}");
            return group::get("local", "foobar");
        },
        on(atom("Spawn5"), arg_match) >> [=](const group & grp)->message {
            CAF_PRINT("received {'Spawn5'}");
            actor_vector vec;
            for (int i = 0; i < 5; ++i) {
                CAF_CHECKPOINT();
                vec.push_back(spawn_in_group(grp, reflector));
            }
            CAF_CHECKPOINT();
            return make_message(atom("ok"), std::move(vec));
        },
        on(atom("Spawn5Done")) >> [=] {
            CAF_PRINT("received {'Spawn5Done'}");
            self->quit();
        });
}

template<typename F>
void await_down(event_based_actor* self, actor ptr, F continuation) {
    self->become(
        [=](const down_msg & dm) -> bool {
            if (dm.source == ptr) {
                continuation();
                return true;
            }
            return false; // not the 'DOWN' message we are waiting for
        }
    );
}

static constexpr size_t num_pings = 10;

class client : public event_based_actor {

 public:

    client(actor server) : m_server(std::move(server)) {}

    behavior make_behavior() override { return spawn_ping(); }

 private:

    behavior spawn_ping() {
        CAF_PRINT("send {'SpawnPing'}");
        send(m_server, atom("SpawnPing"));
        return (on(atom("PingPtr"), arg_match) >> [=](const actor& ping) {
            auto pptr = spawn<monitored + detached + blocking_api>(pong, ping);
            await_down(this, pptr, [=] { send_sync_msg(); });
        });
    }

    void send_sync_msg() {
        CAF_PRINT("sync send {'SyncMsg', 4.2fSyncMsg}");
        sync_send(m_server, atom("SyncMsg"), 4.2f)
            .then(on(atom("SyncReply")) >> [=] { send_foobars(); });
    }

    void send_foobars(int i = 0) {
        if (i == 0) {
            CAF_PRINT("send foobars");
        }<commit_msg>Add a bit more verbosity to remote actor test<commit_after>#include <thread>
#include <string>
#include <cstring>
#include <sstream>
#include <iostream>
#include <functional>

#include "test.hpp"
#include "ping_pong.hpp"

#include "caf/all.hpp"
#include "caf/io/all.hpp"

#include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp"

using namespace std;
using namespace caf;

namespace {

using string_pair = std::pair<std::string, std::string>;

using actor_vector = vector<actor>;

void reflector(event_based_actor* self) {
    self->become(others() >> [=] {
        CAF_PRINT("reflect and quit");
        self->quit();
        return self->last_dequeued();
    });
}

void spawn5_server_impl(event_based_actor* self, actor client, group grp) {
    CAF_LOGF_TRACE(CAF_TARG(client, to_string) << ", "
                                                 << CAF_TARG(grp, to_string));
    CAF_CHECK(grp != invalid_group);
    self->spawn_in_group(grp, reflector);
    self->spawn_in_group(grp, reflector);
    CAF_PRINT("send {'Spawn5'} and await {'ok', actor_vector}");
    self->sync_send(client, atom("Spawn5"), grp)
        .then(on(atom("ok"), arg_match) >> [=](const actor_vector& vec) {
                  CAF_PRINT("received vector with " << vec.size()
                                                     << " elements");
                  self->send(grp, "Hello reflectors!", 5.0);
                  if (vec.size() != 5) {
                      CAF_PRINTERR(
                          "remote client did not spawn five reflectors!");
                  }
                  for (auto& a : vec) {
                      CAF_PRINT("monitor actor: " << to_string(a));
                      self->monitor(a);
                  }
              },
              others() >> [=] {
                  CAF_UNEXPECTED_MSG(self);
                  self->quit(exit_reason::unhandled_exception);
              },
              after(chrono::seconds(10)) >> [=] {
             CAF_UNEXPECTED_TOUT();
             self->quit(exit_reason::unhandled_exception);
         })
        .continue_with([=] {
             CAF_PRINT("wait for reflected messages");
             // receive seven reply messages (2 local, 5 remote)
             auto replies = std::make_shared<int>(0);
             self->become(
                 on("Hello reflectors!", 5.0) >> [=] {
                     if (++*replies == 7) {
                         CAF_PRINT("wait for DOWN messages");
                         auto downs = std::make_shared<int>(0);
                         self->become(
                             [=](const down_msg& dm) {
                                 if (dm.reason != exit_reason::normal) {
                                     CAF_PRINTERR(
                                         "reflector exited for non-normal exit "
                                         "reason!");
                                 }
                                 if (++*downs == 5) {
                                     CAF_CHECKPOINT();
                                     self->send(client, atom("Spawn5Done"));
                                     self->quit();
                                 }
                             },
                             others() >> [=] {
                                 CAF_UNEXPECTED_MSG(self);
                                 // self->quit(exit_reason::unhandled_exception);
                             },
                             after(chrono::seconds(2)) >> [=] {
                                 CAF_UNEXPECTED_TOUT();
                                 CAF_LOGF_ERROR("did only receive "
                                                 << *downs << " down messages");
                                 // self->quit(exit_reason::unhandled_exception);
                             });
                     }
                 },
                 after(std::chrono::seconds(2)) >> [=] {
                     CAF_UNEXPECTED_TOUT();
                     CAF_LOGF_ERROR("did only receive "
                                     << *replies
                                     << " responses to 'Hello reflectors!'");
                     // self->quit(exit_reason::unhandled_exception);
                 });
         });
}

// receive seven reply messages (2 local, 5 remote)
void spawn5_server(event_based_actor* self, actor client, bool inverted) {
    if (!inverted)
        spawn5_server_impl(self, client, group::get("local", "foobar"));
    else {
        CAF_PRINT("request group");
        self->sync_send(client, atom("GetGroup"))
            .then([=](const group& remote_group) {
                 spawn5_server_impl(self, client, remote_group);
             });
    }
}

void spawn5_client(event_based_actor* self) {
    self->become(
        on(atom("GetGroup")) >> []()->group {
            CAF_PRINT("received {'GetGroup'}");
            return group::get("local", "foobar");
        },
        on(atom("Spawn5"), arg_match) >> [=](const group & grp)->message {
            CAF_PRINT("received {'Spawn5'}");
            actor_vector vec;
            for (int i = 0; i < 5; ++i) {
                CAF_CHECKPOINT();
                vec.push_back(spawn_in_group(grp, reflector));
            }
            CAF_CHECKPOINT();
            return make_message(atom("ok"), std::move(vec));
        },
        on(atom("Spawn5Done")) >> [=] {
            CAF_PRINT("received {'Spawn5Done'}");
            self->quit();
        });
}

template<typename F>
void await_down(event_based_actor* self, actor ptr, F continuation) {
    self->become(
        [=](const down_msg & dm) -> bool {
            if (dm.source == ptr) {
                continuation();
                return true;
            }
            return false; // not the 'DOWN' message we are waiting for
        }
    );
}

static constexpr size_t num_pings = 10;

class client : public event_based_actor {

 public:

    client(actor server) : m_server(std::move(server)) {
        // nop
    }

    behavior make_behavior() override {
        return spawn_ping();
    }

 private:

    behavior spawn_ping() {
        CAF_PRINT("send {'SpawnPing'}");
        send(m_server, atom("SpawnPing"));
        return {
            on(atom("PingPtr"), arg_match) >> [=](const actor& ping) {
                CAF_PRINT("received ping pointer, spawn pong");
                auto pptr = spawn<monitored + detached + blocking_api>(pong, ping);
                await_down(this, pptr, [=] { send_sync_msg(); });
            }
        };
    }

    void send_sync_msg() {
        CAF_PRINT("sync send {'SyncMsg', 4.2fSyncMsg}");
        sync_send(m_server, atom("SyncMsg"), 4.2f)
            .then(on(atom("SyncReply")) >> [=] { send_foobars(); });
    }

    void send_foobars(int i = 0) {
        if (i == 0) {
            CAF_PRINT("send foobars");
        }
{'fraction_non_alphanumeric': 0.13487133984028393, 'fraction_numerical': 0.005767524401064774, 'mean_word_length': 1.4382549125653505, 'pattern_counts': {'":': 0, '<': 65, '<?xml version=': 0, '>': 127, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 48, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4421547', 'n_tokens_mistral': 3859, 'n_tokens_neox': 3744, 'n_words': 708}