client.h

/*
 * Copyright (c) 2015, 2025, Oracle and/or its affiliates.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License, version 2.0,
 * as published by the Free Software Foundation.
 *
 * This program is designed to work with certain software (including
 * but not limited to OpenSSL) that is licensed under separate terms,
 * as designated in a particular file or component or in included license
 * documentation. The authors of MySQL hereby grant you an additional
 * permission to link the program and your derivative works with the
 * separately licensed software that they have either included with
 * the program or referenced in the documentation.
 *
 * This program 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 General Public License, version 2.0, for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/

#ifndef PLUGIN_X_SRC_CLIENT_H_
#definePLUGIN_X_SRC_CLIENT_H_

#include<atomic>
#include<memory>
#include<string>

#include"plugin/x/src/capabilities/configurator.h"
#include"plugin/x/src/helper/chrono.h"
#include"plugin/x/src/helper/multithread/mutex.h"
#include"plugin/x/src/interface/client.h"
#include"plugin/x/src/interface/protocol_encoder.h"
#include"plugin/x/src/interface/protocol_monitor.h"
#include"plugin/x/src/interface/server.h"
#include"plugin/x/src/interface/session.h"
#include"plugin/x/src/interface/vio.h"
#include"plugin/x/src/ngs/memory.h"
#include"plugin/x/src/ngs/protocol/message.h"
#include"plugin/x/src/ngs/protocol_decoder.h"
#include"plugin/x/src/ngs/protocol_encoder_compression.h"

#ifndef WIN32
#include<netinet/in.h>
#endif

namespacexpl {

classClient : publiciface::Client {
public:
Client(std::shared_ptr<iface::Vio> connection, iface::Server *server,
 Client_id client_id, iface::Protocol_monitor *pmon);
~Client() override;

 Mutex &get_session_exit_mutex() override { return m_session_exit_mutex; }
 iface::Session *session() override { return m_session.get(); }
 std::shared_ptr<iface::Session> session_shared_ptr() constoverride {
return m_session;
 }

public:// impl iface::Client
voidrun() override;

voidactivate_tls() override;

voidreset_accept_time() override;

voidon_auth_timeout() override;
voidon_server_shutdown() override;
voidkill() override;

 iface::Server &server() constoverride { return *m_server; }
 iface::Protocol_encoder &protocol() constoverride { return *m_encoder; }
 iface::Vio &connection() constoverride { return *m_connection; }

voidon_session_auth_success(iface::Session *s) override;
voidon_session_close(iface::Session *s) override;
voidon_session_reset(iface::Session *s) override;

voiddisconnect_and_trigger_close() override;
boolis_handler_thd(const THD *) constoverride;

constchar *client_address() constoverride { return m_client_addr.c_str(); }
constchar *client_hostname() constoverride { return m_client_host.c_str(); }
constchar *client_hostname_or_address() constoverride;
constchar *client_id() constoverride { return m_id; }
 Client_id client_id_num() constoverride { return m_client_id; }
intclient_port() constoverride { return m_client_port; }

 Client::State get_state() constoverride { return m_state.load(); }
 chrono::Time_point get_accept_time() constoverride;

 iface::Waiting_for_io *get_idle_processing() override;

boolsupports_expired_passwords() constoverride {
return m_supports_expired_passwords;
 }
voidset_supports_expired_passwords(bool flag) override {
 m_supports_expired_passwords = flag;
 }

boolis_interactive() constoverride { return m_is_interactive; }
voidset_is_interactive(constbool flag) override;

voidset_wait_timeout(constuint32_t) override;
voidset_read_timeout(constuint32_t) override;
voidset_write_timeout(constuint32_t) override;

boolhandle_session_connect_attr_set(const ngs::Message_request &command);

voidconfigure_compression_opts(
const ngs::Compression_algorithm algo, constint64_t max_msg,
constbool combine, const Optional_value<int64_t> &level) override;

voidhandle_message(ngs::Message_request *message) override;

 std::string get_status_ssl_cipher_list() const;
 std::string get_status_compression_algorithm() const;
 std::string get_status_compression_level() const;

 ngs::Error_code read_one_message_and_dispatch();
voidset_encoder(iface::Protocol_encoder *enc);
voidset_session(iface::Session *session) { m_session.reset(session); }
voidset_idle_reporting(iface::Waiting_for_io *reporter) {
 m_idle_reporting.reset(reporter);
 }

private:
classMessage_dispatcher
 : public ngs::Message_decoder::Message_dispatcher_interface {
public:
explicitMessage_dispatcher(iface::Client *client) : m_client(client) {}

voidhandle(ngs::Message_request *message) override {
 m_client->handle_message(message);
 }

private:
 iface::Client *m_client;
 };

classClient_idle_reporting;

protected:
char m_id[2 + sizeof(Client_id) * 2 + 1]; // 64bits in hex, plus 0x plus \0
 Client_id m_client_id;
 iface::Server *m_server;

 std::unique_ptr<iface::Waiting_for_io> m_idle_reporting;
 std::shared_ptr<iface::Vio> m_connection;
 std::shared_ptr<ngs::Protocol_config> m_config;
// TODO(lkotula): benchmark m_memory_block_pool as global in Xpl (shouldn't be
// in review)
 ngs::Memory_block_pool m_memory_block_pool{{10, k_minimum_page_size}};
 Message_dispatcher m_dispatcher;
 ngs::Protocol_decoder m_decoder;

 chrono::Time_point m_accept_time;

 ngs::Memory_instrumented<iface::Protocol_encoder>::Unique_ptr m_encoder;
 std::string m_client_addr;
 std::string m_client_host;
uint16_t m_client_port;
 std::atomic<Client::State> m_state;
 std::atomic<Client::State> m_state_when_reason_changed;
 std::atomic<bool> m_removed;

 std::shared_ptr<iface::Session> m_session;

 iface::Protocol_monitor *m_protocol_monitor;

mutable Mutex m_session_exit_mutex;

enumclassClose_reason {
 k_none,
 k_net_error,
 k_error,
 k_reject,
 k_normal,
 k_server_shutdown,
 k_kill,
 k_connect_timeout,
 k_write_timeout,
 k_read_timeout
 };

 std::atomic<Close_reason> m_close_reason{Close_reason::k_none};

char *m_msg_buffer{nullptr};
size_t m_msg_buffer_size{0};
bool m_supports_expired_passwords{false};
bool m_is_interactive{false};
bool m_is_compression_encoder_injected{false};

uint32_t m_read_timeout{0};
uint32_t m_write_timeout{0};

 ngs::Compression_algorithm m_cached_compression_algorithm{
 ngs::Compression_algorithm::k_none};
int64_t m_cached_max_msg{-1};
bool m_cached_combine_msg{false};
int32_t m_cached_compression_level{3};

int32_tget_adjusted_compression_level(
const ngs::Compression_algorithm algo,
const Optional_value<int64_t> &level) const;

 Capabilities_configurator *capabilities_configurator();
voidget_capabilities(
const Mysqlx::Connection::CapabilitiesGet &msg) override;
voidset_capabilities(
const Mysqlx::Connection::CapabilitiesSet &msg) override;

voidremove_client_from_server();

 std::string resolve_hostname() const;
virtualvoidon_network_error(constint error);
voidon_read_timeout();

 iface::Protocol_monitor &get_protocol_monitor();

private:
 ngs::Protocol_encoder_compression *get_protocol_compression_or_install_it();

Client(const Client &) = delete;
 Client &operator=(const Client &) = delete;

voidget_last_error(int *out_error_code, std::string *out_message);
voidset_close_reason_if_non_fatal(const Close_reason reason);
voidupdate_counters();
voidqueue_up_disconnection_notice(const ngs::Error_code &error);
voidqueue_up_disconnection_notice_if_necessary();

voidon_client_addr();
voidon_accept();
boolcreate_session();
boolis_localhost(constchar *hostname) const;
};

} // namespace xpl

#endif// PLUGIN_X_SRC_CLIENT_H_