socket_server_manager_asio_impl.hpp

Go to the documentation of this file.

/******************************************************************************
** Copyright © 2015 by J.M.McGuiness, coder@hussar.me.uk
**
** This library is free software; you can redistribute it and/or
** modify it under the terms of the GNU Lesser General Public
** License as published by the Free Software Foundation; either
** version 2.1 of the License, or (at your option) any later version.
**
** This library is distributed in the hope that it will be useful,
** but WITHOUT ANY WARRANTY; without even the implied warranty of
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
** Lesser General Public License for more details.
**
** You should have received a copy of the GNU Lesser General Public
** License along with this library; if not, write to the Free Software
** Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*/
 
namespace jmmcg { namespace LIBJMMCG_VER_NAMESPACE { namespace socket { namespace server_manager { namespace asio {
 
template<class LkT>
inline std::string
manager<LkT>::session::to_string() const noexcept(false) {
   std::ostringstream ss;
   ss<<socket_;
   return ss.str();
}
 
template<class LkT>
inline std::ostream &
operator<<(std::ostream &os, typename manager<LkT>::session const &ec) noexcept(false) {
   os<<ec.to_string();
   return os;
}
 
template<class LkT>
inline
manager<LkT>::manager(boost::asio::ip::address const &addr, unsigned short port_num, std::size_t /*min_message_size*/, std::size_t max_message_size, unsigned short timeout, socket_priority /*priority*/, std::size_t /*incoming_cpu*/, server_to_client_flow_t &&server_to_client_flow)
: acceptor(io_context, boost::asio::ip::tcp::endpoint(addr, port_num)),
   server_to_client_flow_(std::move(server_to_client_flow)) {
// TODO "protocol not available":   acceptor.set_option(boost::asio::socket_base::send_low_watermark(min_message_size));
   acceptor.set_option(boost::asio::socket_base::send_buffer_size(max_message_size));
// TODO "protocol not available":   acceptor.set_option(boost::asio::socket_base::receive_low_watermark(min_message_size));
   acceptor.set_option(boost::asio::socket_base::receive_buffer_size(max_message_size));
   acceptor.set_option(boost::asio::socket_base::linger(true, timeout));
   acceptor.set_option(boost::asio::ip::tcp::no_delay(true));
   acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
}
 
template<class LkT>
inline void
manager<LkT>::run() {
   io_context.run();
}
 
template<class LkT>
inline void
manager<LkT>::stop() {
   io_context.stop();
   while (!io_context.stopped()) {
      thread_traits::sleep(0);
   }
}
 
template<class LkT>
inline void
manager<LkT>::set_options(boost::asio::ip::tcp::acceptor &acceptor, socket_t &skt) {
   boost::asio::socket_base::send_buffer_size send_buffer_size;
   acceptor.get_option(send_buffer_size);
   boost::asio::socket_base::receive_buffer_size receive_buffer_size;
   acceptor.get_option(receive_buffer_size);
   boost::asio::socket_base::linger linger_details;
   acceptor.get_option(linger_details);
   skt.set_options(1, std::max(send_buffer_size.value(), receive_buffer_size.value()), linger_details.timeout(), socket_t::socket_priority::low, 0);
   boost::asio::ip::tcp::no_delay no_delay;
   acceptor.get_option(no_delay);
}
 
template<class LkT>
inline void
manager<LkT>::set_options(socket_t &skt) {
   set_options(acceptor, skt);
}
 
template<class LkT>
inline std::string
manager<LkT>::to_string() const noexcept(false) {
   std::ostringstream ss;
// TODO  ss<<acceptor;
   return ss.str();
}
 
template<class LkT>
inline std::ostream &
operator<<(std::ostream &os, manager<LkT> const &ec) noexcept(false) {
   os<<ec.to_string();
   return os;
}
 
template<class SvrHBs, class LkT>
class loopback<SvrHBs, LkT>::send_heartbeats final : public manager<LkT>::session {
public:
   using base_t=typename manager<LkT>::session;
   using session=base_t;
 
   explicit send_heartbeats(boost::asio::ip::tcp::socket &&socket)
   : base_t(std::move(socket)) {
   }
 
   ~send_heartbeats() {
      this->stop();
   }
 
   static typename session::ptr_type make(boost::asio::ip::tcp::socket &&socket) noexcept(false) {
      return std::make_shared<send_heartbeats>(std::move(socket));
   }
 
   void start() override {
      assert(!heatbeating);
      heatbeating.reset(new heartbeats_t(this->socket_));
   }
 
   void stop() override {
      if (heatbeating) {
         heatbeating->stop();
      }
      base_t::stop();
   }
 
private:
   std::unique_ptr<heartbeats_t> heatbeating;
};
 
template<class SvrHBs, class LkT>
inline
loopback<SvrHBs, LkT>::loopback(boost::asio::ip::address const &addr, unsigned short port_num, std::size_t min_message_size, std::size_t max_message_size, unsigned short, socket_priority priority, std::size_t incoming_cpu, server_to_client_flow_t &&server_to_client_flow)
: base_t(addr, port_num, min_message_size, max_message_size, (heartbeats_t::max_missed_heartbeats*heartbeats_t::heartbeat_interval).count(), priority, incoming_cpu, std::move(server_to_client_flow)) {
}
 
template<class SvrHBs, class LkT>
template<class RecvProcMsgs> inline void
loopback<SvrHBs, LkT>::start_accept(RecvProcMsgs proc_fn) noexcept(false) {
   this->acceptor.async_accept(
      [this, proc_fn](boost::system::error_code const &error, boost::asio::ip::tcp::socket &&client_conn) {
         bool exit=false;
         if (!error) {
            typename base_t::session::ptr_type client_connection(send_heartbeats::make(std::move(client_conn)));
            assert(client_connection.get());
            this->set_options(client_connection->socket());
            assert(client_connection->socket().is_open());
            this->server_to_client_flow_(client_connection);
            exit=client_connection->process(proc_fn);
            this->server_to_client_flow_(typename base_t::session::ptr_type{});
         }
         if (!exit) {
            start_accept(proc_fn);
         }
      }
   );
}
 
template<class LkT> inline
forwarding<LkT>::forwarding(boost::asio::ip::address const &addr, unsigned short port_num, std::size_t min_message_size, std::size_t max_message_size, unsigned short timeout, socket_priority priority, std::size_t incoming_cpu, server_to_client_flow_t &&server_to_client_flow, socket_t &dest)
: base_t(addr, port_num, min_message_size, max_message_size, timeout, priority, incoming_cpu, std::move(server_to_client_flow)), dest_socket_(dest) {
}
 
template<class LkT>
template<class RecvProcMsgs> inline void
forwarding<LkT>::start_accept(RecvProcMsgs proc_fn) noexcept(false) {
   this->acceptor.async_accept(
      [this, proc_fn](boost::system::error_code const &error, boost::asio::ip::tcp::socket &&client_conn) {
         bool exit=false;
         if (!error) {
            typename base_t::session::ptr_type client_connection(base_t::session::make(std::move(client_conn)));
            assert(client_connection.get());
            this->set_options(client_connection->socket());
            this->set_options(dest_socket_);
            assert(client_connection->socket().is_open());
            this->server_to_client_flow_(client_connection);
            exit=client_connection->process(proc_fn, dest_socket_);
            this->server_to_client_flow_(typename base_t::session::ptr_type{});
         }
         if (!exit) {
            start_accept(proc_fn);
         }
      }
   );
}
 
} } } } }