io_output_delivery.hpp

Go to the documentation of this file.

 
#ifndef IO_OUTPUT_DELIVERY_HPP_INCLUDED
#define IO_OUTPUT_DELIVERY_HPP_INCLUDED
 
#include <cstddef> // std::size_t
#include <utility> // std::move, std::pair, std::forward
#include <system_error>
#include <memory>
#include <exception>
#include <future>
 
#include "net_ip/net_entity.hpp"
#include "net_ip/basic_io_interface.hpp"
#include "net_ip/basic_io_output.hpp"
#include "net_ip/io_type_decls.hpp"
 
#include "queue/wait_queue.hpp"
 
namespace chops {
namespace net {
 
template <typename IOT>
struct io_state_chg_data {
  basic_io_output<IOT>    io_out;
  std::size_t             num_handlers;
  bool                    starting;
 
  io_state_chg_data(basic_io_output<IOT> io, std::size_t num, bool s) :
      io_out(std::move(io)), num_handlers(num), starting(s) { }
};
 
template <typename IOT>
using io_wait_q = chops::wait_queue<io_state_chg_data<IOT> >;
 
using tcp_io_wait_q = io_wait_q<chops::net::tcp_io>;
using udp_io_wait_q = io_wait_q<chops::net::udp_io>;
 
template <typename IOT, typename IOS, typename EF>
auto start_with_io_wait_queue (net_entity ne, 
                               IOS&& io_start,
                               io_wait_q<IOT>& wq, 
                               EF&& err_func) {
  return ne.start( [io_start = std::move(io_start), &wq]
                   (basic_io_interface<IOT> io, std::size_t num, bool starting) {
      if (starting) {
        io_start(io, num, starting);
      }
      wq.emplace_push(*(io.make_io_output()), num, starting);
    },
    std::forward<EF>(err_func)
  );
}
 
template <typename IOT>
using io_output_future = std::future<basic_io_output<IOT> >;
 
using tcp_io_output_future = io_output_future<tcp_io>;
using udp_io_output_future = io_output_future<udp_io>;
 
 
template <typename IOT>
struct io_output_future_pair {
  io_output_future<IOT>   start_fut;
  io_output_future<IOT>   stop_fut;
};
 
using tcp_io_output_future_pair = io_output_future_pair<tcp_io>;
using udp_io_output_future_pair = io_output_future_pair<udp_io>;
 
 
template <typename IOT, typename IOS, typename EF>
io_output_future<IOT> make_io_output_future(net_entity& ent,
                                            IOS&& io_start,
                                            EF&& err_func) {
  // io state change function object must be copyable since it will
  // be stored in a std::function, therefore wrap promise in shared ptr
  auto start_prom_ptr = std::make_shared<std::promise<basic_io_output<IOT> > >();
  auto start_fut = start_prom_ptr->get_future();
 
  auto lam = [io_start = std::move(io_start), start_prom_ptr] 
                    (basic_io_interface<IOT> io, std::size_t num, bool starting) {
    if (starting) {
      io_start(io, num, starting);
      start_prom_ptr->set_value(*(io.make_io_output()));
    }
  };
  auto e = ent.start(lam, std::forward<EF>(err_func));
  if (!e) { // error return from net_entity start method
    start_prom_ptr->set_exception(std::make_exception_ptr(std::system_error(e.error())));
  }
  return start_fut;
}
 
template <typename IOT, typename IOS, typename EF>
io_output_future_pair<IOT> make_io_output_future_pair(net_entity& ent,
                                                      IOS&& io_start,
                                                      EF&& err_func) {
 
  auto start_prom_ptr = std::make_shared<std::promise<basic_io_output<IOT> > >();
  auto start_fut = start_prom_ptr->get_future();
  auto stop_prom_ptr = std::make_shared<std::promise<basic_io_output<IOT> > >();
  auto stop_fut = stop_prom_ptr->get_future();
 
  auto lam = [io_start = std::move(io_start), start_prom_ptr, stop_prom_ptr] 
                (basic_io_interface<IOT> io, std::size_t num, bool starting) {
    if (starting) {
      io_start(io, num, starting);
      start_prom_ptr->set_value(*(io.make_io_output()));
    }
    else {
      stop_prom_ptr->set_value(*(io.make_io_output()));
    }
  };
 
  auto e = ent.start(lam, std::forward<EF>(err_func));
  if (!e) { // error return from net_entity start method
    auto ep = std::make_exception_ptr(std::system_error(e.error()));
    start_prom_ptr->set_exception(ep);
    stop_prom_ptr->set_exception(ep);
  }
  return io_output_future_pair<IOT> { std::move(start_fut), std::move(stop_fut) };
}
 
} // end net namespace
} // end chops namespace
 
#endif