send_to_all.hpp

Go to the documentation of this file.

 
#ifndef SEND_TO_ALL_HPP_INCLUDED
#define SEND_TO_ALL_HPP_INCLUDED
 
#include <cstddef> // std::size_t
#include <utility> // std::move
 
#include <mutex>
#include <vector>
 
#include "net_ip/basic_io_interface.hpp"
#include "net_ip/basic_io_output.hpp"
 
#include "net_ip_component/output_queue_stats.hpp"
 
#include "buffer/shared_buffer.hpp"
 
namespace chops {
namespace net {
 
template <typename IOT>
class send_to_all {
private:
  using lock_guard  = std::scoped_lock<std::mutex>;
  using io_out      = chops::net::basic_io_output<IOT>;
  using io_outs     = std::vector<io_out>;
  using io_interface = chops::net::basic_io_interface<IOT>;
 
private:
  mutable std::mutex    m_mutex;
  io_outs               m_io_outs;
 
public:
  void add_io_output(io_out io) {
    lock_guard gd { m_mutex };
    m_io_outs.push_back(io);
  }
 
  void remove_io_output(io_out io) {
    lock_guard gd { m_mutex };
    std::erase_if (m_io_outs, [io] (auto out) { return io == out; } );
  }
 
  void operator() (io_interface io, std::size_t, bool starting) {
    auto ret = io.make_io_output();
    if (!ret) {
      return;
    }
    
    if (starting) {
      add_io_output(*ret);
    }
    else {
      remove_io_output(*ret);
    }
  }
 
  void send(chops::const_shared_buffer buf) const {
    lock_guard gd { m_mutex };
    for (const auto& io : m_io_outs) {
      io.send(buf);
    }
  }
 
  void send(chops::const_shared_buffer buf, io_out cur_io) const { // TG
    lock_guard gd { m_mutex };
    for (const auto& io : m_io_outs) {
      if ( !(cur_io == io) ) {
        io.send(buf);
      }
    }
  }
 
  void send(const void* buf, std::size_t sz) const {
    send(chops::const_shared_buffer(buf, sz));
  }
 
  void send(const void* buf, std::size_t sz, io_out cur_io) const { // TG
    send(chops::const_shared_buffer(buf, sz), cur_io);
  }
 
  void send(chops::mutable_shared_buffer&& buf) const { 
    send(chops::const_shared_buffer(std::move(buf)));
  }
 
  void send(chops::mutable_shared_buffer&& buf, io_out cur_io) const { 
    send(chops::const_shared_buffer(std::move(buf)), cur_io);
  }
 
  std::size_t size() const noexcept {
    lock_guard gd { m_mutex };
    return m_io_outs.size();
  }
 
  auto get_total_output_queue_stats() const noexcept {
    lock_guard gd { m_mutex };
    return accumulate_output_queue_stats(m_io_outs.cbegin(), m_io_outs.cend());
  }
};
 
} // end net namespace
} // end chops namespace
 
#endif