<commit_before>// Copyright 2014 Open Source Robotics Foundation, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#ifndef RCLCPP__PUBLISHER_HPP_
#define RCLCPP__PUBLISHER_HPP_

#include <rmw/error_handling.h>
#include <rmw/rmw.h>

#include <iostream>
#include <memory>
#include <mutex>
#include <sstream>
#include <string>

#include "rcl_interfaces/msg/intra_process_message.hpp"
#include "rmw/impl/cpp/demangle.hpp"

#include "rclcpp/allocator/allocator_common.hpp"
#include "rclcpp/allocator/allocator_deleter.hpp"
#include "rclcpp/macros.hpp"
#include "rclcpp/visibility_control.hpp"

namespace rclcpp
{

// Forward declaration for friend statement
namespace node
{
class Node;
}  // namespace node

namespace publisher
{

class PublisherBase
{
public:
  RCLCPP_SMART_PTR_DEFINITIONS(PublisherBase);
  /// Default constructor.
  /**
   * Typically, a publisher is not created through this method, but instead is created through a
   * call to `Node::create_publisher`.
   * \param[in] node_handle The corresponding rmw representation of the owner node.
   * \param[in] publisher_handle The rmw publisher handle corresponding to this publisher.
   * \param[in] topic The topic that this publisher publishes on.
   * \param[in] queue_size The maximum number of unpublished messages to queue.
   */
  RCLCPP_PUBLIC
  PublisherBase(
    std::shared_ptr<rmw_node_t> node_handle,
    rmw_publisher_t * publisher_handle,
    std::string topic,
    size_t queue_size);

  /// Default destructor.
  RCLCPP_PUBLIC
  virtual ~PublisherBase();

  /// Get the topic that this publisher publishes on.
  // \return The topic name.
  RCLCPP_PUBLIC
  const std::string &
  get_topic_name() const;

  /// Get the queue size for this publisher.
  // \return The queue size.
  RCLCPP_PUBLIC
  size_t
  get_queue_size() const;

  /// Get the global identifier for this publisher (used in rmw and by DDS).
  // \return The gid.
  RCLCPP_PUBLIC
  const rmw_gid_t &
  get_gid() const;

  /// Get the global identifier for this publisher used by intra-process communication.
  // \return The intra-process gid.
  RCLCPP_PUBLIC
  const rmw_gid_t &
  get_intra_process_gid() const;

  /// Compare this publisher to a gid.
  /**
   * Note that this function calls the next function.
   * \param[in] gid Reference to a gid.
   * \return True if the publisher's gid matches the input.
   */
  RCLCPP_PUBLIC
  bool
  operator==(const rmw_gid_t & gid) const;

  /// Compare this publisher to a pointer gid.
  /**
   * A wrapper for comparing this publisher's gid to the input using rmw_compare_gids_equal.
   * \param[in] gid A pointer to a gid.
   * \return True if this publisher's gid matches the input.
   */
  RCLCPP_PUBLIC
  bool
  operator==(const rmw_gid_t * gid) const;

  typedef std::function<uint64_t(uint64_t, void *, const std::type_info &)> StoreMessageCallbackT;

protected:
  RCLCPP_PUBLIC
  void
  setup_intra_process(
    uint64_t intra_process_publisher_id,
    StoreMessageCallbackT callback,
    rmw_publisher_t * intra_process_publisher_handle);

  std::shared_ptr<rmw_node_t> node_handle_;

  rmw_publisher_t * publisher_handle_;
  rmw_publisher_t * intra_process_publisher_handle_;

  std::string topic_;
  size_t queue_size_;

  uint64_t intra_process_publisher_id_;
  StoreMessageCallbackT store_intra_process_message_;

  rmw_gid_t rmw_gid_;
  rmw_gid_t intra_process_rmw_gid_;

  std::mutex intra_process_publish_mutex_;
};

/// A publisher publishes messages of any type to a topic.
template<typename MessageT, typename Alloc = std::allocator<void>>
class Publisher : public PublisherBase
{
  friend rclcpp::node::Node;

public:
  using MessageAllocTraits = allocator::AllocRebind<MessageT, Alloc>;
  using MessageAlloc = typename MessageAllocTraits::allocator_type;
  using MessageDeleter = allocator::Deleter<MessageAlloc, MessageT>;
  using MessageUniquePtr = std::unique_ptr<MessageT, MessageDeleter>;

  RCLCPP_SMART_PTR_DEFINITIONS(Publisher<MessageT, Alloc>);

  Publisher(
    std::shared_ptr<rmw_node_t> node_handle,
    rmw_publisher_t * publisher_handle,
    std::string topic,
    size_t queue_size,
    std::shared_ptr<Alloc> allocator)
  : PublisherBase(node_handle, publisher_handle, topic, queue_size)
  {
    message_allocator_ = std::make_shared<MessageAlloc>(*allocator.get());
    allocator::set_allocator_for_deleter(&message_deleter_, message_allocator_.get());
  }


  /// Send a message to the topic for this publisher.
  /**
   * This function is templated on the input message type, MessageT.
   * \param[in] msg A shared pointer to the message to send.
   */
  void
  publish(std::unique_ptr<MessageT, MessageDeleter> & msg)
  {
    this->do_inter_process_publish(msg.get());
    if (store_intra_process_message_) {
      // Take the pointer from the unique_msg, release it and pass as a void *
      // to the ipm. The ipm should then capture it again as a unique_ptr of
      // the correct type.
      // TODO(wjwwood):
      //   investigate how to transfer the custom deleter (if there is one)
      //   from the incoming unique_ptr through to the ipm's unique_ptr.
      //   See: http://stackoverflow.com/questions/11002641/dynamic-casting-for-unique-ptr
      MessageT * msg_ptr = msg.get();
      msg.release();
      uint64_t message_seq;
      {
        std::lock_guard<std::mutex> lock(intra_process_publish_mutex_);
        message_seq =
          store_intra_process_message_(intra_process_publisher_id_, msg_ptr, typeid(MessageT));
      }
      rcl_interfaces::msg::IntraProcessMessage ipm;
      ipm.publisher_id = intra_process_publisher_id_;
      ipm.message_sequence = message_seq;
      auto status = rmw_publish(intra_process_publisher_handle_, &ipm);
      if (status != RMW_RET_OK) {
        // *INDENT-OFF* (prevent uncrustify from making unecessary indents here)
        throw std::runtime_error(
          std::string("failed to publish intra process message: ") + rmw_get_error_string_safe());
        // *INDENT-ON*
      }
    } else {
      // Always destroy the message, even if we don't consume it, for consistency.
      msg.reset();
    }
  }<commit_msg>Remove mutex in publisher<commit_after>// Copyright 2014 Open Source Robotics Foundation, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#ifndef RCLCPP__PUBLISHER_HPP_
#define RCLCPP__PUBLISHER_HPP_

#include <rmw/error_handling.h>
#include <rmw/rmw.h>

#include <iostream>
#include <memory>
#include <mutex>
#include <sstream>
#include <string>

#include "rcl_interfaces/msg/intra_process_message.hpp"
#include "rmw/impl/cpp/demangle.hpp"

#include "rclcpp/allocator/allocator_common.hpp"
#include "rclcpp/allocator/allocator_deleter.hpp"
#include "rclcpp/macros.hpp"
#include "rclcpp/visibility_control.hpp"

namespace rclcpp
{

// Forward declaration for friend statement
namespace node
{
class Node;
}  // namespace node

namespace publisher
{

class PublisherBase
{
public:
  RCLCPP_SMART_PTR_DEFINITIONS(PublisherBase);
  /// Default constructor.
  /**
   * Typically, a publisher is not created through this method, but instead is created through a
   * call to `Node::create_publisher`.
   * \param[in] node_handle The corresponding rmw representation of the owner node.
   * \param[in] publisher_handle The rmw publisher handle corresponding to this publisher.
   * \param[in] topic The topic that this publisher publishes on.
   * \param[in] queue_size The maximum number of unpublished messages to queue.
   */
  RCLCPP_PUBLIC
  PublisherBase(
    std::shared_ptr<rmw_node_t> node_handle,
    rmw_publisher_t * publisher_handle,
    std::string topic,
    size_t queue_size);

  /// Default destructor.
  RCLCPP_PUBLIC
  virtual ~PublisherBase();

  /// Get the topic that this publisher publishes on.
  // \return The topic name.
  RCLCPP_PUBLIC
  const std::string &
  get_topic_name() const;

  /// Get the queue size for this publisher.
  // \return The queue size.
  RCLCPP_PUBLIC
  size_t
  get_queue_size() const;

  /// Get the global identifier for this publisher (used in rmw and by DDS).
  // \return The gid.
  RCLCPP_PUBLIC
  const rmw_gid_t &
  get_gid() const;

  /// Get the global identifier for this publisher used by intra-process communication.
  // \return The intra-process gid.
  RCLCPP_PUBLIC
  const rmw_gid_t &
  get_intra_process_gid() const;

  /// Compare this publisher to a gid.
  /**
   * Note that this function calls the next function.
   * \param[in] gid Reference to a gid.
   * \return True if the publisher's gid matches the input.
   */
  RCLCPP_PUBLIC
  bool
  operator==(const rmw_gid_t & gid) const;

  /// Compare this publisher to a pointer gid.
  /**
   * A wrapper for comparing this publisher's gid to the input using rmw_compare_gids_equal.
   * \param[in] gid A pointer to a gid.
   * \return True if this publisher's gid matches the input.
   */
  RCLCPP_PUBLIC
  bool
  operator==(const rmw_gid_t * gid) const;

  typedef std::function<uint64_t(uint64_t, void *, const std::type_info &)> StoreMessageCallbackT;

protected:
  RCLCPP_PUBLIC
  void
  setup_intra_process(
    uint64_t intra_process_publisher_id,
    StoreMessageCallbackT callback,
    rmw_publisher_t * intra_process_publisher_handle);

  std::shared_ptr<rmw_node_t> node_handle_;

  rmw_publisher_t * publisher_handle_;
  rmw_publisher_t * intra_process_publisher_handle_;

  std::string topic_;
  size_t queue_size_;

  uint64_t intra_process_publisher_id_;
  StoreMessageCallbackT store_intra_process_message_;

  rmw_gid_t rmw_gid_;
  rmw_gid_t intra_process_rmw_gid_;
};

/// A publisher publishes messages of any type to a topic.
template<typename MessageT, typename Alloc = std::allocator<void>>
class Publisher : public PublisherBase
{
  friend rclcpp::node::Node;

public:
  using MessageAllocTraits = allocator::AllocRebind<MessageT, Alloc>;
  using MessageAlloc = typename MessageAllocTraits::allocator_type;
  using MessageDeleter = allocator::Deleter<MessageAlloc, MessageT>;
  using MessageUniquePtr = std::unique_ptr<MessageT, MessageDeleter>;

  RCLCPP_SMART_PTR_DEFINITIONS(Publisher<MessageT, Alloc>);

  Publisher(
    std::shared_ptr<rmw_node_t> node_handle,
    rmw_publisher_t * publisher_handle,
    std::string topic,
    size_t queue_size,
    std::shared_ptr<Alloc> allocator)
  : PublisherBase(node_handle, publisher_handle, topic, queue_size)
  {
    message_allocator_ = std::make_shared<MessageAlloc>(*allocator.get());
    allocator::set_allocator_for_deleter(&message_deleter_, message_allocator_.get());
  }


  /// Send a message to the topic for this publisher.
  /**
   * This function is templated on the input message type, MessageT.
   * \param[in] msg A shared pointer to the message to send.
   */
  void
  publish(std::unique_ptr<MessageT, MessageDeleter> & msg)
  {
    this->do_inter_process_publish(msg.get());
    if (store_intra_process_message_) {
      // Take the pointer from the unique_msg, release it and pass as a void *
      // to the ipm. The ipm should then capture it again as a unique_ptr of
      // the correct type.
      // TODO(wjwwood):
      //   investigate how to transfer the custom deleter (if there is one)
      //   from the incoming unique_ptr through to the ipm's unique_ptr.
      //   See: http://stackoverflow.com/questions/11002641/dynamic-casting-for-unique-ptr
      MessageT * msg_ptr = msg.get();
      msg.release();
      uint64_t message_seq =
        store_intra_process_message_(intra_process_publisher_id_, msg_ptr, typeid(MessageT));
      rcl_interfaces::msg::IntraProcessMessage ipm;
      ipm.publisher_id = intra_process_publisher_id_;
      ipm.message_sequence = message_seq;
      auto status = rmw_publish(intra_process_publisher_handle_, &ipm);
      if (status != RMW_RET_OK) {
        // *INDENT-OFF* (prevent uncrustify from making unecessary indents here)
        throw std::runtime_error(
          std::string("failed to publish intra process message: ") + rmw_get_error_string_safe());
        // *INDENT-ON*
      }
    } else {
      // Always destroy the message, even if we don't consume it, for consistency.
      msg.reset();
    }
  }
{'fraction_non_alphanumeric': 0.09088138950255199, 'fraction_numerical': 0.003961301135065133, 'mean_word_length': 3.7912408759124085, 'pattern_counts': {'":': 0, '<': 44, '<?xml version=': 0, '>': 46, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7040667', 'n_tokens_mistral': 4327, 'n_tokens_neox': 4135, 'n_words': 1196}