<commit_before>/******************************************************************************
 *                       ____    _    _____                                   *
 *                      / ___|  / \  |  ___|    C++                           *
 *                     | |     / _ \ | |_       Actor                         *
 *                     | |___ / ___ \|  _|      Framework                     *
 *                      \____/_/   \_|_|                                      *
 *                                                                            *
 * Copyright (C) 2011 - 2015                                                  *
 * <NAME> <<EMAIL>ik.charousset (at) haw-hamburg.de>                *
 *                                                                            *
 * Distributed under the terms and conditions of the BSD 3-Clause License or  *
 * (at your option) under the terms and conditions of the Boost Software      *
 * License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE.       *
 *                                                                            *
 * If you did not receive a copy of the license files, see                    *
 * http://opensource.org/licenses/BSD-3-Clause and                            *
 * http://www.boost.org/LICENSE_1_0.txt.                                      *
 ******************************************************************************/

#include "caf/config.hpp"

#define CAF_SUITE io_basp
#include "caf/test/unit_test.hpp"

#include <array>
#include <mutex>
#include <memory>
#include <limits>
#include <iostream>
#include <condition_variable>

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

#include "caf/experimental/whereis.hpp"

#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/test_multiplexer.hpp"

namespace std {

ostream& operator<<(ostream& out, const caf::io::basp::message_type& x) {
  return out << to_string(x);
}

template <class T>
ostream& operator<<(ostream& out, const caf::variant<caf::anything, T>& x) {
  using std::to_string;
  using caf::to_string;
  using caf::io::basp::to_string;
  if (get<caf::anything>(&x) != nullptr)
    return out << "*";
  return out << to_string(get<T>(x));
}

} // namespace std

namespace caf {

template <class T>
bool operator==(const variant<anything, T>& x, const T& y) {
  return get<anything>(&x) != nullptr || get<T>(x) == y;
}

template <class T>
bool operator==(const T& x, const variant<anything, T>& y) {
  return (y == x);
}

template <class T>
std::string to_string(const variant<anything, T>& x) {
  return get<anything>(&x) == nullptr ? std::string{"*"} : to_string(get<T>(x));
}

} // namespace caf

using namespace std;
using namespace caf;
using namespace caf::io;
using namespace caf::experimental;

namespace {

#define THROW_ON_UNEXPECTED(selfref)                                           \
  others >> [&] {                                                              \
    throw std::logic_error("unexpected message: "                              \
                           + to_string(selfref->current_message()));           \
  },                                                                           \
  after(std::chrono::seconds(0)) >> [&] {                                      \
    throw std::logic_error("unexpected timeout");                              \
  }

static constexpr uint32_t num_remote_nodes = 2;

using buffer = std::vector<char>;

string hexstr(const buffer& buf) {
  ostringstream oss;
  oss << hex;
  oss.fill('0');
  for (auto& c : buf) {
    oss.width(2);
    oss << int{c};
  }
  return oss.str();
}

class fixture {
public:
  fixture() {
    mpx_ = new network::test_multiplexer;
    set_middleman(mpx_);
    auto mm = middleman::instance();
    aut_ = mm->get_named_broker<basp_broker>(atom("_BASP"));
    this_node_ = detail::singletons::get_node_id();
    CAF_MESSAGE("this node: " << to_string(this_node_));
    self_.reset(new scoped_actor);
    // run the initialization message of the BASP broker
    mpx_->exec_runnable();
    // handle the message from the configuration server
    mpx()->exec_runnable();
    ahdl_ = accept_handle::from_int(1);
    mpx_->assign_tcp_doorman(aut_.get(), ahdl_);
    registry_ = detail::singletons::get_actor_registry();
    registry_->put((*self_)->id(),
                  actor_cast<abstract_actor_ptr>((*self_)->address()));
    // first remote node is everything of this_node + 1, then +2, etc.
    for (uint32_t i = 0; i < num_remote_nodes; ++i) {
      node_id::host_id_type tmp = this_node_.host_id();
      for (auto& c : tmp)
        c = static_cast<uint8_t>(c + i + 1);
      remote_node_[i] = node_id{this_node_.process_id() + i + 1, tmp};
      remote_hdl_[i] = connection_handle::from_int(i + 1);
      auto& ptr = pseudo_remote_[i];
      ptr.reset(new scoped_actor);
      // register all pseudo remote actors in the registry
      registry_->put((*ptr)->id(),
                    actor_cast<abstract_actor_ptr>((*ptr)->address()));
    }
  }

  uint32_t serialized_size(const message& msg) {
    buffer buf;
    binary_serializer bs{back_inserter(buf), &get_namespace()};
    bs << msg;
    return static_cast<uint32_t>(buf.size());
  }

  ~fixture() {
    this_node_ = invalid_node_id;
    self_.reset();
    for (auto& nid : remote_node_)
      nid = invalid_node_id;
    for (auto& ptr : pseudo_remote_)
      ptr.reset();
    await_all_actors_done();
    shutdown();
  }

  // our "virtual communication backend"
  network::test_multiplexer* mpx() {<commit_msg>Fix bug in BASP test caused by recent changes<commit_after>/******************************************************************************
 *                       ____    _    _____                                   *
 *                      / ___|  / \  |  ___|    C++                           *
 *                     | |     / _ \ | |_       Actor                         *
 *                     | |___ / ___ \|  _|      Framework                     *
 *                      \____/_/   \_|_|                                      *
 *                                                                            *
 * Copyright (C) 2011 - 2015                                                  *
 * <NAME> <dominik.charousset (at) haw-hamburg.de>                *
 *                                                                            *
 * Distributed under the terms and conditions of the BSD 3-Clause License or  *
 * (at your option) under the terms and conditions of the Boost Software      *
 * License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE.       *
 *                                                                            *
 * If you did not receive a copy of the license files, see                    *
 * http://opensource.org/licenses/BSD-3-Clause and                            *
 * http://www.boost.org/LICENSE_1_0.txt.                                      *
 ******************************************************************************/

#include "caf/config.hpp"

#define CAF_SUITE io_basp
#include "caf/test/unit_test.hpp"

#include <array>
#include <mutex>
#include <memory>
#include <limits>
#include <iostream>
#include <condition_variable>

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

#include "caf/experimental/whereis.hpp"

#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/test_multiplexer.hpp"

namespace std {

ostream& operator<<(ostream& out, const caf::io::basp::message_type& x) {
  return out << to_string(x);
}

template <class T>
ostream& operator<<(ostream& out, const caf::variant<caf::anything, T>& x) {
  using std::to_string;
  using caf::to_string;
  using caf::io::basp::to_string;
  if (get<caf::anything>(&x) != nullptr)
    return out << "*";
  return out << to_string(get<T>(x));
}

} // namespace std

namespace caf {

template <class T>
bool operator==(const variant<anything, T>& x, const T& y) {
  return get<anything>(&x) != nullptr || get<T>(x) == y;
}

template <class T>
bool operator==(const T& x, const variant<anything, T>& y) {
  return (y == x);
}

template <class T>
std::string to_string(const variant<anything, T>& x) {
  return get<anything>(&x) == nullptr ? std::string{"*"} : to_string(get<T>(x));
}

} // namespace caf

using namespace std;
using namespace caf;
using namespace caf::io;
using namespace caf::experimental;

namespace {

#define THROW_ON_UNEXPECTED(selfref)                                           \
  others >> [&] {                                                              \
    throw std::logic_error("unexpected message: "                              \
                           + to_string(selfref->current_message()));           \
  },                                                                           \
  after(std::chrono::seconds(0)) >> [&] {                                      \
    throw std::logic_error("unexpected timeout");                              \
  }

static constexpr uint32_t num_remote_nodes = 2;

using buffer = std::vector<char>;

string hexstr(const buffer& buf) {
  ostringstream oss;
  oss << hex;
  oss.fill('0');
  for (auto& c : buf) {
    oss.width(2);
    oss << int{c};
  }
  return oss.str();
}

class fixture {
public:
  fixture() {
    mpx_ = new network::test_multiplexer;
    set_middleman(mpx_);
    auto mm = middleman::instance();
    aut_ = mm->get_named_broker<basp_broker>(atom("_BASP"));
    this_node_ = detail::singletons::get_node_id();
    CAF_MESSAGE("this node: " << to_string(this_node_));
    self_.reset(new scoped_actor);
    ahdl_ = accept_handle::from_int(1);
    mpx_->assign_tcp_doorman(aut_.get(), ahdl_);
    registry_ = detail::singletons::get_actor_registry();
    registry_->put((*self_)->id(),
                  actor_cast<abstract_actor_ptr>((*self_)->address()));
    // first remote node is everything of this_node + 1, then +2, etc.
    for (uint32_t i = 0; i < num_remote_nodes; ++i) {
      node_id::host_id_type tmp = this_node_.host_id();
      for (auto& c : tmp)
        c = static_cast<uint8_t>(c + i + 1);
      remote_node_[i] = node_id{this_node_.process_id() + i + 1, tmp};
      remote_hdl_[i] = connection_handle::from_int(i + 1);
      auto& ptr = pseudo_remote_[i];
      ptr.reset(new scoped_actor);
      // register all pseudo remote actors in the registry
      registry_->put((*ptr)->id(),
                    actor_cast<abstract_actor_ptr>((*ptr)->address()));
    }
    // make sure all init messages are handled properly
    mpx_->flush_runnables();
  }

  uint32_t serialized_size(const message& msg) {
    buffer buf;
    binary_serializer bs{back_inserter(buf), &get_namespace()};
    bs << msg;
    return static_cast<uint32_t>(buf.size());
  }

  ~fixture() {
    this_node_ = invalid_node_id;
    self_.reset();
    for (auto& nid : remote_node_)
      nid = invalid_node_id;
    for (auto& ptr : pseudo_remote_)
      ptr.reset();
    await_all_actors_done();
    shutdown();
  }

  // our "virtual communication backend"
  network::test_multiplexer* mpx() {
{'fraction_non_alphanumeric': 0.15579612088407757, 'fraction_numerical': 0.006134415877311682, 'mean_word_length': 1.9089477827341905, 'pattern_counts': {'":': 0, '<': 98, '<?xml version=': 0, '>': 89, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 44, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2937158', 'n_tokens_mistral': 3338, 'n_tokens_neox': 3185, 'n_words': 710}