exchange_to_client_processor_impl.hpp

Go to the documentation of this file.

/******************************************************************************
** Copyright © 2020 by J.M.McGuiness, isimud@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 isimud { namespace ISIMUD_VER_NAMESPACE { namespace exchanges { namespace common {
 
template<class ClientCxnPtr, class EchgCxn>
class exchange_to_client_processor<ClientCxnPtr, EchgCxn>::flow final {
public:
   template<class LatencyTimestamps>
   flow(std::atomic<bool> &e, exchg_link_t &exchg_link, LatencyTimestamps &timestamps) noexcept(true);
 
   ~flow() noexcept(false);
 
   /// A client has connected so make sure the connection is passed to the exchg_to_client_thread, so it can forward messages as necessary. Also must be used to stop sending messages to that client.
   /**
    * When a client is disconnected, this must be done early to ensure that this class does not attempt to delete the contained socket, as that is actually owned by the boost::io_context and if not reset could cause a double-free error.
    */
   void connect_client(client_connection_t client_connection) noexcept(true);
 
   std::string to_string() const noexcept(false);
 
private:
   using thread_t=libjmmcg::ppd::jthread;
 
   std::atomic<bool> &exit_;
   boost::exception_ptr client_err{};
   exchg_link_t &exchg_link_;
   client_connection_t client_connection_{};
   /**
      The thread that spins waiting for messages from the exchange to forward to the connected client.
   */
   thread_t exchg_to_client_thread;
 
   /**
    * There may not be a client connected yet, but we still need to respond to heartbeats from the exchange.
    */
   template<class LatencyTimestamps>
   void process_msgs(LatencyTimestamps &timestamps) noexcept(true);
};
 
template<class ClientCxnPtr, class EchgCxn>
inline std::ostream &
operator<<(std::ostream &os, typename exchange_to_client_processor<ClientCxnPtr, EchgCxn>::flow const &s) noexcept(false) {
   os<<s.to_string();
   return os;
}
 
template<class ClientCxnPtr, class EchgCxn>
template<class LatencyTimestamps> inline
exchange_to_client_processor<ClientCxnPtr, EchgCxn>::exchange_to_client_processor(std::atomic<bool> &e, ctor_args const &exchange_cxn_details, socket_priority to_exchg_priority, proc_rules_t const &proc_rules, LatencyTimestamps &timestamps)
:  exchg_link(exchange_cxn_details, to_exchg_priority, common::thread_traits::exchange_to_client_thread.core, proc_rules),
   exchange_to_client(e, exchg_link, timestamps) {
}
 
template<class ClientCxnPtr, class EchgCxn> inline bool
exchange_to_client_processor<ClientCxnPtr, EchgCxn>::is_logged_on() const noexcept(true) {
   return exchg_link.is_logged_on();
}
 
template<class ClientCxnPtr, class EchgCxn> inline void
exchange_to_client_processor<ClientCxnPtr, EchgCxn>::exchange_to_client_processor::connect_client(client_connection_t client_connection) noexcept(true) {
   exchange_to_client.connect_client(client_connection);
}
 
template<class ClientCxnPtr, class EchgCxn> inline typename exchange_to_client_processor<ClientCxnPtr, EchgCxn>::socket_t &
exchange_to_client_processor<ClientCxnPtr, EchgCxn>::socket() noexcept(true) {
   return exchg_link.socket();
}
 
template<class ClientCxnPtr, class EchgCxn> inline std::string
exchange_to_client_processor<ClientCxnPtr, EchgCxn>::to_string() const noexcept(false) {
   std::ostringstream os;
   os<<"exchange connection: "<<exchg_link
      <<", exchange-to-client details: "<<exchange_to_client.to_string();
   return os.str();
}
 
template<class ClientCxnPtr, class EchgCxn>
inline std::ostream &
operator<<(std::ostream &os, exchange_to_client_processor<ClientCxnPtr, EchgCxn> const &s) noexcept(false) {
   os<<s.to_string();
   return os;
}
 
template<class ClientCxnPtr, class EchgCxn>
template<class LatencyTimestamps> inline
exchange_to_client_processor<ClientCxnPtr, EchgCxn>::flow::flow(std::atomic<bool> &e, exchg_link_t &exchg_link, LatencyTimestamps &timestamps) noexcept(true)
:  exit_(e),
   exchg_link_(exchg_link),
   exchg_to_client_thread(
      [this, &timestamps]() {
         this->process_msgs<LatencyTimestamps>(timestamps);
      }
   ) {
   exchg_to_client_thread.kernel_priority(common::thread_traits::exchange_to_client_thread.priority);
   exchg_to_client_thread.kernel_affinity(
      typename thread_t::thread_traits::api_params_type::processor_mask_type(common::thread_traits::exchange_to_client_thread.core)
   );
   exchg_to_client_thread.set_name("exchg_to_client");
}
 
template<class ClientCxnPtr, class EchgCxn> inline
exchange_to_client_processor<ClientCxnPtr, EchgCxn>::flow::~flow() noexcept(false) {
   connect_client(client_connection_t{});
   exit_=true;
   if (client_err) {
      boost::rethrow_exception(client_err);
   }
}
 
template<class ClientCxnPtr, class EchgCxn> inline void
exchange_to_client_processor<ClientCxnPtr, EchgCxn>::flow::connect_client(client_connection_t client_connection) noexcept(true) {
   client_connection_=client_connection;
   assert(!client_connection_ || client_connection_->socket().is_open());
}
 
template<class ClientCxnPtr, class EchgCxn> inline std::string
exchange_to_client_processor<ClientCxnPtr, EchgCxn>::flow::to_string() const noexcept(false) {
   std::ostringstream os;
   os
      <<", exchange-to-client core="<<common::thread_traits::exchange_to_client_thread
      <<", client processing-error: '"<<client_err<<"'";
   if (client_connection_.get()) {
      os<<", client link: "<<client_connection_->socket();
   }
   return os.str();
}
 
template<class ClientCxnPtr, class EchgCxn>
template<class LatencyTimestamps> inline void
exchange_to_client_processor<ClientCxnPtr, EchgCxn>::flow::process_msgs(LatencyTimestamps &timestamps) noexcept(true) {
   try {
      while (LIKELY(!static_cast<bool>(exit_))) {
         try {
            if (UNLIKELY(exchg_link_.read_and_process_a_msg(client_connection_, timestamps))) {
               return;
            }
         } catch (...) {
            // TODO Should really return a msg to the client indicating failed to translate, with details...
            client_err=boost::current_exception();
         }
      }
   } catch (...) {
      client_err=boost::current_exception();
   }
}
 
} } } }