<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/none.hpp"

#include "caf/config.hpp"

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

#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"

#include "caf/detail/make_counted.hpp"
#include "caf/detail/actor_registry.hpp"
#include "caf/detail/sync_request_bouncer.hpp"

namespace caf {
namespace io {

broker::servant::~servant() {
  CAF_LOG_TRACE("");
}

broker::servant::servant(broker* ptr) : m_disconnected(false), m_broker(ptr) {
  // nop
}

void broker::servant::set_broker(broker* new_broker) {
  if (!m_disconnected) {
    m_broker = new_broker;
  }
}

void broker::servant::disconnect(bool invoke_disconnect_message) {
  CAF_LOG_TRACE("");
  if (!m_disconnected) {
    CAF_LOG_DEBUG("disconnect servant from broker");
    m_disconnected = true;
    remove_from_broker();
    if (invoke_disconnect_message) {
      auto msg = disconnect_message();
      m_broker->invoke_message(m_broker->address(),invalid_message_id, msg);
    }
  }
}

broker::scribe::scribe(broker* ptr, connection_handle conn_hdl)
    : servant(ptr),
      m_hdl(conn_hdl) {
  std::vector<char> tmp;
  m_read_msg = make_message(new_data_msg{m_hdl, std::move(tmp)});
}

void broker::scribe::remove_from_broker() {
  CAF_LOG_TRACE("hdl = " << hdl().id());
  m_broker->m_scribes.erase(hdl());
}

broker::scribe::~scribe() {
  CAF_LOG_TRACE("");
}

message broker::scribe::disconnect_message() {
  return make_message(connection_closed_msg{hdl()});
}

void broker::scribe::consume(const void*, size_t num_bytes) {
  CAF_LOG_TRACE(CAF_ARG(num_bytes));
  auto& buf = rd_buf();
  buf.resize(num_bytes);                       // make sure size is correct
  read_msg().buf.swap(buf);                    // swap into message
  m_broker->invoke_message(invalid_actor_addr, // call client
                           invalid_message_id, m_read_msg);
  read_msg().buf.swap(buf); // swap buffer back to stream
  flush();                  // implicit flush of wr_buf()
}

void broker::scribe::io_failure(network::operation op) {
  CAF_LOG_TRACE("id = " << hdl().id()
                << ", " << CAF_TARG(op, static_cast<int>));
  // keep compiler happy when compiling w/o logging
  static_cast<void>(op);
  disconnect(true);
}

broker::doorman::doorman(broker* ptr, accept_handle acc_hdl)
    : servant(ptr), m_hdl(acc_hdl) {
  auto hdl2 = connection_handle::from_int(-1);
  m_accept_msg = make_message(new_connection_msg{m_hdl, hdl2});
}

broker::doorman::~doorman() {
  CAF_LOG_TRACE("");<commit_msg>Discard data on disconnected scribes<commit_after>/******************************************************************************
 *                       ____    _    _____                                   *
 *                      / ___|  / \  |  ___|    C++                           *
 *                     | |     / _ \ | |_       Actor                         *
 *                     | |___ / ___ \|  _|      Framework                     *
 *                      \____/_/   \_|_|                                      *
 *                                                                            *
 * Copyright (C) 2011 - 2015                                                  *
 * <NAME> <<EMAIL>inik.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/none.hpp"

#include "caf/config.hpp"

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

#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"

#include "caf/detail/make_counted.hpp"
#include "caf/detail/actor_registry.hpp"
#include "caf/detail/sync_request_bouncer.hpp"

namespace caf {
namespace io {

broker::servant::~servant() {
  CAF_LOG_TRACE("");
}

broker::servant::servant(broker* ptr) : m_disconnected(false), m_broker(ptr) {
  // nop
}

void broker::servant::set_broker(broker* new_broker) {
  if (!m_disconnected) {
    m_broker = new_broker;
  }
}

void broker::servant::disconnect(bool invoke_disconnect_message) {
  CAF_LOG_TRACE("");
  if (!m_disconnected) {
    CAF_LOG_DEBUG("disconnect servant from broker");
    m_disconnected = true;
    remove_from_broker();
    if (invoke_disconnect_message) {
      auto msg = disconnect_message();
      m_broker->invoke_message(m_broker->address(),invalid_message_id, msg);
    }
  }
}

broker::scribe::scribe(broker* ptr, connection_handle conn_hdl)
    : servant(ptr),
      m_hdl(conn_hdl) {
  std::vector<char> tmp;
  m_read_msg = make_message(new_data_msg{m_hdl, std::move(tmp)});
}

void broker::scribe::remove_from_broker() {
  CAF_LOG_TRACE("hdl = " << hdl().id());
  m_broker->m_scribes.erase(hdl());
}

broker::scribe::~scribe() {
  CAF_LOG_TRACE("");
}

message broker::scribe::disconnect_message() {
  return make_message(connection_closed_msg{hdl()});
}

void broker::scribe::consume(const void*, size_t num_bytes) {
  CAF_LOG_TRACE(CAF_ARG(num_bytes));
  if (m_disconnected) {
    // we are already disconnected from the broker while the multiplexer
    // did not yet remove the socket, this can happen if an IO event causes
    // the broker to call close_all() while the pollset contained
    // further activities for the broker
    return;
  }
  auto& buf = rd_buf();
  buf.resize(num_bytes);                       // make sure size is correct
  read_msg().buf.swap(buf);                    // swap into message
  m_broker->invoke_message(invalid_actor_addr, // call client
                           invalid_message_id, m_read_msg);
  read_msg().buf.swap(buf); // swap buffer back to stream
  flush();                  // implicit flush of wr_buf()
}

void broker::scribe::io_failure(network::operation op) {
  CAF_LOG_TRACE("id = " << hdl().id()
                << ", " << CAF_TARG(op, static_cast<int>));
  // keep compiler happy when compiling w/o logging
  static_cast<void>(op);
  disconnect(true);
}

broker::doorman::doorman(broker* ptr, accept_handle acc_hdl)
    : servant(ptr), m_hdl(acc_hdl) {
  auto hdl2 = connection_handle::from_int(-1);
  m_accept_msg = make_message(new_connection_msg{m_hdl, hdl2});
}

broker::doorman::~doorman() {
  CAF_LOG_TRACE("");
{'fraction_non_alphanumeric': 0.1581798483206934, 'fraction_numerical': 0.004092933670398459, 'mean_word_length': 2.074759437453738, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 32, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2453325', 'n_tokens_mistral': 2553, 'n_tokens_neox': 2470, 'n_words': 452}