WL12495: DevAPI: Support session attributes

Author: silvakid

CMakeLists.txt

@@ -98,6 +98,12 @@ else()
 endif()
 
 
+#
+# Connector System Configuration
+#
+include(config.cmake)
+
+
 #
 # Install settings
 # ================

cdk/foundation/connection_tcpip.cc

@@ -130,6 +130,42 @@ namespace cdk {
 namespace foundation {
 namespace connection {
 
+class Socket_system_initializer
+{
+  Socket_system_initializer()
+  {
+    detail::initialize_socket_system();
+  }
+
+  ~Socket_system_initializer()
+  {
+    try
+    {
+      detail::uninitialize_socket_system();
+    }
+    catch (...)
+    {
+      // Ignoring errors in destructor.
+    }
+  }
+
+  friend void socket_system_initialize();
+};
+
+void socket_system_initialize()
+{
+  static Socket_system_initializer initializer;
+}
+
+
+std::string get_local_hostname()
+{
+  // This will initialize socket system (e.g. Winsock) during construction of first CDK connection.
+  socket_system_initialize();
+
+  return detail::get_local_hostname();
+}
+
 
 TCPIP::TCPIP(const std::string& host,
              unsigned short port,

cdk/foundation/connection_tcpip_base.h

@@ -42,28 +42,11 @@ namespace foundation {
 namespace connection {
 
 
+void socket_system_initialize();
+
+
 class Socket_base::Impl
 {
-  class Socket_system_initializer
-  {
-  public:
-    Socket_system_initializer()
-    {
-      detail::initialize_socket_system();
-    }
-
-    ~Socket_system_initializer()
-    {
-      try
-      {
-        detail::uninitialize_socket_system();
-      }
-      catch (...)
-      {
-        // Ignoring errors in destructor.
-      }
-    }
-  };
 
 public:
   typedef detail::Socket socket;
@@ -74,7 +57,7 @@ class Socket_base::Impl
     : m_sock(detail::NULL_SOCKET)
   {
     // This will initialize socket system (e.g. Winsock) during construction of first CDK connection.
-    static Socket_system_initializer initializer;
+    socket_system_initialize();
   }
 
   bool is_open() const

cdk/foundation/socket_detail.cc

@@ -968,5 +968,14 @@ size_t send_some(Socket socket, const byte *buffer, size_t buffer_size, bool wai
   return bytes_sent;
 }
 
+std::string get_local_hostname()
+{
+  char buf[1024] = {0};
+  if (gethostname(buf, sizeof(buf)) < 0) {
+    throw_socket_error();
+  }
+  return buf;
+}
+
 
 }}}} // cdk::foundation::connection::detail

cdk/foundation/socket_detail.h

@@ -442,6 +442,11 @@ size_t recv_some(Socket socket, byte *buffer, size_t buffer_size, bool wait);
 size_t send_some(Socket socket, const byte *buffer, size_t buffer_size, bool wait);
 
 
+/**
+   @brief get_local_hostname returns hostname of the current machine
+ */
+std::string get_local_hostname();
+
 }}}} // cdk::foundation::connection::detail
 
 

cdk/include/mysql/cdk/data_source.h

@@ -37,6 +37,7 @@ PUSH_SYS_WARNINGS_CDK
 #include <functional>
 #include <algorithm>
 #include <set>
+#include "api/expression.h"
 POP_SYS_WARNINGS_CDK
 
 
@@ -46,12 +47,29 @@ namespace cdk {
 
 namespace ds {
 
+
+struct Attr_processor
+{
+  virtual ~Attr_processor() {}
+  virtual void attr(const string &key, const string &val)=0;
+};
+
+class Session_attributes
+  : public cdk::api::Expr_base<Attr_processor>
+{};
+
+
 /*
  * Generic session options which are valid for any data source.
  */
 
+
+
 template <class Base>
-class Options : public Base
+class Options
+    : public Base
+    , public Session_attributes
+    , public Attr_processor
 {
 public:
 
@@ -88,6 +106,31 @@ class Options : public Base
     m_has_db = true;
   }
 
+  void set_attributes(std::map<std::string,std::string> &connection_attr)
+  {
+    m_connection_attr = connection_attr;
+  }
+
+  const Session_attributes* attributes() const
+  {
+    if (m_connection_attr.empty())
+      return nullptr;
+    return this;
+  }
+
+  void process(Processor &prc) const override
+  {
+    for (auto &el :  m_connection_attr)
+    {
+      prc.attr(el.first, el.second);
+    }
+  }
+
+  void attr(const string &key, const string &val) override
+  {
+    m_connection_attr[key]=val;
+  }
+
 protected:
 
   string m_usr;
@@ -96,6 +139,7 @@ class Options : public Base
 
   bool    m_has_db;
   string  m_db;
+  std::map<std::string,std::string> m_connection_attr;
 
 };
 

cdk/include/mysql/cdk/foundation/connection_tcpip.h

@@ -49,6 +49,12 @@ namespace connection {
 class TCPIP;
 class TLS;
 
+/**
+   @brief get_local_hostname returns hostname of the current machine
+ */
+std::string get_local_hostname();
+
+
 
 /*
   Defining cdkio error category

cdk/include/mysql/cdk/mysqlx/session.h

@@ -524,7 +524,9 @@ class Session
   {
     m_prepare_prc.m_session = this;
     m_stmt_stats.clear();
+    send_connection_attr(options);
     authenticate(options, conn.is_secure());
+
     // TODO: make "lazy" checks instead, deferring to the time when given
     // feature is used.
     check_protocol_fields();
@@ -697,6 +699,8 @@ class Session
 
   Reply_init & set_command(Proto_prepare_op *cmd);
 
+  // Send Connection Attributes
+  void send_connection_attr(const Options &options);
   // Authentication (cdk::protocol::mysqlx::Auth_processor)
   void authenticate(const Options &options, bool secure = false);
   void do_authenticate(const Options &options, int auth_method, bool secure);

cdk/mysqlx/session.cc

@@ -354,6 +354,73 @@ class AuthSha256Memory
    Class Session
 */
 
+void Session::send_connection_attr(const Options &options)
+{
+
+  struct Attr_converter
+      : cdk::protocol::mysqlx::api::Any::Document
+      , ds::Attr_processor
+  {
+    Attr_converter(const ds::Session_attributes* attr)
+      :m_attr(attr)
+    {}
+
+    const ds::Session_attributes * m_attr;
+    Processor::Any_prc::Doc_prc *m_attr_prc;
+
+    void process(Processor &prc) const override
+    {
+      auto *self  = const_cast<Attr_converter*>(this);
+      prc.doc_begin();
+      self->m_attr_prc = prc.key_val("session_connect_attrs")->doc();
+      self->m_attr_prc->doc_begin();
+      m_attr->process(*self);
+      self->m_attr_prc->doc_end();
+      prc.doc_end();
+    }
+
+    void attr(const string &key, const string &val) override
+    {
+      m_attr_prc->key_val(key)->scalar()->str(bytes(val));
+    }
+
+  } ;
+
+  if (options.attributes())
+  {
+    m_protocol.snd_CapabilitiesSet(Attr_converter(options.attributes())).wait();
+
+    struct Check_reply_prc : cdk::protocol::mysqlx::Reply_processor
+    {
+      string m_msg;
+      unsigned int m_code = 0;
+      cdk::protocol::mysqlx::sql_state_t m_sql_state;
+      void error(unsigned int code, short int,
+                 cdk::protocol::mysqlx::sql_state_t state, const string &msg) override
+      {
+        m_code = code;
+        m_sql_state = state;
+        m_msg = msg;
+      }
+
+      void ok(string) override
+      {}
+    };
+
+    Check_reply_prc prc;
+
+    m_protocol.rcv_Reply(prc).wait();
+
+    if(prc.m_code != 0 &&    prc.m_code != 5002)
+    {
+      //code: 5002
+      //msg: "Capability \'session_connect_attrs\' doesn\'t exist"
+      throw Server_error(prc.m_code, prc.m_sql_state, prc.m_msg);
+    }
+
+  }
+}
+
 
 void Session::send_auth()
 {