<commit_before>// Copyright 2016 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "mojo/edk/system/node_controller.h"

#include <algorithm>

#include "base/bind.h"
#include "base/location.h"
#include "base/logging.h"
#include "crypto/random.h"
#include "mojo/edk/embedder/platform_channel_pair.h"
#include "mojo/edk/system/core.h"
#include "mojo/edk/system/ports_message.h"

namespace mojo {
namespace edk {

namespace {

template <typename T>
void GenerateRandomName(T* out) { crypto::RandBytes(out, sizeof(T)); }

ports::NodeName GetRandomNodeName() {
  ports::NodeName name;
  GenerateRandomName(&name);
  return name;
}

}  // namespace

NodeController::PendingTokenConnection::PendingTokenConnection() {}

NodeController::PendingTokenConnection::~PendingTokenConnection() {}

NodeController::ReservedPort::ReservedPort() {}

NodeController::ReservedPort::~ReservedPort() {}

NodeController::~NodeController() {}

NodeController::NodeController(Core* core)
    : core_(core),
      name_(GetRandomNodeName()),
      node_(new ports::Node(name_, this)) {
  DVLOG(1) << "Initializing node " << name_;
}

void NodeController::SetIOTaskRunner(
    scoped_refptr<base::TaskRunner> task_runner) {
  io_task_runner_ = task_runner;
}

void NodeController::ConnectToChild(ScopedPlatformHandle platform_handle) {
  io_task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&NodeController::ConnectToChildOnIOThread,
                 base::Unretained(this),
                 base::Passed(&platform_handle)));
}

void NodeController::ConnectToParent(ScopedPlatformHandle platform_handle) {
  io_task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&NodeController::ConnectToParentOnIOThread,
                 base::Unretained(this),
                 base::Passed(&platform_handle)));
}

void NodeController::SetPortObserver(
    const ports::PortRef& port,
    std::shared_ptr<PortObserver> observer) {
  DCHECK(observer);
  node_->SetUserData(port, std::move(observer));
}

scoped_ptr<PortsMessage> NodeController::AllocMessage(size_t num_payload_bytes,
                                                      size_t num_ports) {
  ports::ScopedMessage m;
  int rv = node_->AllocMessage(num_payload_bytes, num_ports, &m);
  if (rv != ports::OK)
    return nullptr;
  DCHECK(m);

  return make_scoped_ptr(static_cast<PortsMessage*>(m.release()));
}

int NodeController::SendMessage(const ports::PortRef& port,
                                scoped_ptr<PortsMessage> message) {
  ports::ScopedMessage ports_message(message.release());
  return node_->SendMessage(port, std::move(ports_message));
}

void NodeController::ReservePortForToken(const ports::PortName& port_name,
                                         const std::string& token,
                                         const base::Closure& on_connect) {
  io_task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&NodeController::ReservePortForTokenOnIOThread,
                 base::Unretained(this), port_name, token, on_connect));
}

void NodeController::ConnectToParentPortByToken(
    const std::string& token,
    const ports::PortName& local_port,
    const base::Closure& on_connect) {
  io_task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&NodeController::ConnectToParentPortByTokenOnIOThread,
                 base::Unretained(this), token, local_port, on_connect));
}

void NodeController::ConnectToChildOnIOThread(<commit_msg>flush NodeController's incoming message queue after SendMessage<commit_after>// Copyright 2016 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "mojo/edk/system/node_controller.h"

#include <algorithm>

#include "base/bind.h"
#include "base/location.h"
#include "base/logging.h"
#include "crypto/random.h"
#include "mojo/edk/embedder/platform_channel_pair.h"
#include "mojo/edk/system/core.h"
#include "mojo/edk/system/ports_message.h"

namespace mojo {
namespace edk {

namespace {

template <typename T>
void GenerateRandomName(T* out) { crypto::RandBytes(out, sizeof(T)); }

ports::NodeName GetRandomNodeName() {
  ports::NodeName name;
  GenerateRandomName(&name);
  return name;
}

}  // namespace

NodeController::PendingTokenConnection::PendingTokenConnection() {}

NodeController::PendingTokenConnection::~PendingTokenConnection() {}

NodeController::ReservedPort::ReservedPort() {}

NodeController::ReservedPort::~ReservedPort() {}

NodeController::~NodeController() {}

NodeController::NodeController(Core* core)
    : core_(core),
      name_(GetRandomNodeName()),
      node_(new ports::Node(name_, this)) {
  DVLOG(1) << "Initializing node " << name_;
}

void NodeController::SetIOTaskRunner(
    scoped_refptr<base::TaskRunner> task_runner) {
  io_task_runner_ = task_runner;
}

void NodeController::ConnectToChild(ScopedPlatformHandle platform_handle) {
  io_task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&NodeController::ConnectToChildOnIOThread,
                 base::Unretained(this),
                 base::Passed(&platform_handle)));
}

void NodeController::ConnectToParent(ScopedPlatformHandle platform_handle) {
  io_task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&NodeController::ConnectToParentOnIOThread,
                 base::Unretained(this),
                 base::Passed(&platform_handle)));
}

void NodeController::SetPortObserver(
    const ports::PortRef& port,
    std::shared_ptr<PortObserver> observer) {
  DCHECK(observer);
  node_->SetUserData(port, std::move(observer));
}

scoped_ptr<PortsMessage> NodeController::AllocMessage(size_t num_payload_bytes,
                                                      size_t num_ports) {
  ports::ScopedMessage m;
  int rv = node_->AllocMessage(num_payload_bytes, num_ports, &m);
  if (rv != ports::OK)
    return nullptr;
  DCHECK(m);

  return make_scoped_ptr(static_cast<PortsMessage*>(m.release()));
}

int NodeController::SendMessage(const ports::PortRef& port,
                                scoped_ptr<PortsMessage> message) {
  ports::ScopedMessage ports_message(message.release());
  int rv = node_->SendMessage(port, std::move(ports_message));

  // Flush the incoming message queue in case the call to SendMessage queued
  // any messages locally.
  AcceptIncomingMessages();

  return rv;
}

void NodeController::ReservePortForToken(const ports::PortName& port_name,
                                         const std::string& token,
                                         const base::Closure& on_connect) {
  io_task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&NodeController::ReservePortForTokenOnIOThread,
                 base::Unretained(this), port_name, token, on_connect));
}

void NodeController::ConnectToParentPortByToken(
    const std::string& token,
    const ports::PortName& local_port,
    const base::Closure& on_connect) {
  io_task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&NodeController::ConnectToParentPortByTokenOnIOThread,
                 base::Unretained(this), token, local_port, on_connect));
}

void NodeController::ConnectToChildOnIOThread(
{'fraction_non_alphanumeric': 0.11811023622047244, 'fraction_numerical': 0.001381406271584473, 'mean_word_length': 4.077138849929874, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5267148', 'n_tokens_mistral': 2134, 'n_tokens_neox': 2129, 'n_words': 390}