socket_server_manager_glibc_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 glibc {
 
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(addr.to_string().c_str(), port_num, min_message_size, max_message_size, timeout, priority, incoming_cpu),
   server_to_client_flow_(std::move(server_to_client_flow)) {
}
 
template<class LkT>
inline void
manager<LkT>::run() {
   acceptor.run();
}
 
template<class LkT>
inline void
manager<LkT>::stop() {
   acceptor.stop();
}
 
template<class LkT>
inline void
manager<LkT>::set_options(acceptor_t &acceptor, socket_t &skt) {
   acceptor.set_options(skt);
}
 
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;
   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;
 
   send_heartbeats(typename socket_t::socket_type accept_socket)
   : base_t(accept_socket) {
   }
 
   ~send_heartbeats() {
      this->stop();
   }
 
   static typename session::ptr_type make(typename socket_t::socket_type accept_socket) noexcept(false) {
      return std::make_shared<send_heartbeats>(accept_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](typename socket_t::socket_type accept_socket) {
         bool exit=false;
         {
            typename base_t::session::ptr_type client_connection(send_heartbeats::make(accept_socket));
            this->set_options(client_connection->socket());
            assert(client_connection.get());
            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](typename socket_t::socket_type accept_socket) {
         bool exit=false;
         {
            typename base_t::session::ptr_type client_connection(base_t::session::make(accept_socket));
            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);
         }
      }
   );
}
 
} } } } }