Merge branch 'wl13947'

Author: silvakid

cdk/include/mysql/cdk/data_source.h

@@ -197,20 +197,36 @@ class Protocol_options
     REQUIRED
   };
 
+  enum compression_algorithm_t {
+    NONE,
+    DEFLATE_STREAM,
+    LZ4_MESSAGE,
+    ZSTD_STREAM
+  };
+
   virtual auth_method_t auth_method() const = 0;
   virtual compression_mode_t compression() const = 0;
 
+  using Compression_algorithms = std::vector<compression_algorithm_t>;
+  virtual const Compression_algorithms& compression_algorithms() const = 0;
+
 };
 
 
 class Options
   : public ds::Options<Protocol_options>,
     public foundation::connection::Socket_base::Options
 {
+public:
+
+  typedef std::vector<compression_algorithm_t> Compression_algorithms;
+
 protected:
 
   auth_method_t m_auth_method = DEFAULT;
-  compression_mode_t m_compression = DISABLED;
+  compression_mode_t m_compression = PREFERRED;
+  bool m_has_compression_alg = false;
+  Compression_algorithms m_compression_algorithms;
 
 public:
 
@@ -241,6 +257,23 @@ class Options
     return m_compression;
   }
 
+  void add_compression_alg(compression_algorithm_t val)
+  {
+    m_has_compression_alg = true;
+    m_compression_algorithms.push_back(val);
+  }
+
+  const Compression_algorithms& compression_algorithms() const
+  {
+    if (m_has_compression_alg)
+      return m_compression_algorithms;
+
+    static Compression_algorithms default_compression_algorithms =
+    { ZSTD_STREAM, LZ4_MESSAGE, DEFLATE_STREAM };
+
+    return default_compression_algorithms ;
+  }
+
 };
 
 

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

@@ -253,6 +253,8 @@ class Session
 
   typedef ds::Options<ds::mysqlx::Protocol_options> Options;
   using compression_mode_t = ds::mysqlx::Protocol_options::compression_mode_t;
+  using compression_algorithm_t =
+    ds::mysqlx::Protocol_options::compression_algorithm_t;
 
   template <class C>
   Session(C &conn, const Options &options)
@@ -267,7 +269,7 @@ class Session
 
     if (options.compression() != compression_mode_t::DISABLED)
     {
-      compression = negotiate_compression();
+      compression = negotiate_compression(options.compression_algorithms());
 
       if (compression == Compression_type::NONE &&
           options.compression() == compression_mode_t::REQUIRED)
@@ -286,16 +288,13 @@ class Session
 
   /*
     Get the most suitable compression type supported by the server.
-    Function prioritizes compression types as follows:
-      LZ4
-      DEFLATE
 
-    If the higher priority type is not available the function will
-    check a lower priority type.
+    Vector order is used for priorization, so first element has the higher
+    priority.
 
     NONE is returned if the server does not support compression
   */
-  Compression_type::value negotiate_compression();
+  Compression_type::value negotiate_compression(const std::vector<compression_algorithm_t>& algorithms);
 
   virtual ~Session();
 

cdk/mysqlx/session.cc

@@ -43,7 +43,8 @@ POP_SYS_WARNINGS_CDK
 namespace cdk {
 namespace mysqlx {
 
-Compression_type::value Session::negotiate_compression()
+Compression_type::value Session::negotiate_compression(
+      const std::vector<cdk::mysqlx::Session::compression_algorithm_t>& algorithms)
 {
   Compression_type::value compression = Compression_type::NONE;
 
@@ -84,27 +85,59 @@ Compression_type::value Session::negotiate_compression()
   } cap_prc;
 
   /*
-    The compression types must be attempted with increaing
+    The compression types must be attempted with increasing
     priority. The last successful type will be applied.
   */
-  compress_caps.m_algorithm = "deflate_stream";
-  m_protocol.snd_CapabilitiesSet(compress_caps).wait();
-  compress_caps.m_algorithm = "lz4_message";
-  m_protocol.snd_CapabilitiesSet(compress_caps).wait();
-  compress_caps.m_algorithm = "zstd_stream";
-  m_protocol.snd_CapabilitiesSet(compress_caps).wait();
-
-  m_protocol.rcv_Reply(cap_prc).wait();
-  if (cap_prc.m_compression_ok)
-    compression = Compression_type::DEFLATE;
-
-  m_protocol.rcv_Reply(cap_prc).wait();
-  if (cap_prc.m_compression_ok)
-    compression = Compression_type::LZ4;
-
-  m_protocol.rcv_Reply(cap_prc).wait();
-  if (cap_prc.m_compression_ok)
-    compression = Compression_type::ZSTD;
+
+  for(auto alg = algorithms.rbegin(); alg != algorithms.rend(); ++alg)
+  {
+    switch (*alg) {
+    case cdk::mysqlx::Session::compression_algorithm_t::DEFLATE_STREAM:
+      compress_caps.m_algorithm = "deflate_stream";
+      m_protocol.snd_CapabilitiesSet(compress_caps).wait();
+      break;
+    case cdk::mysqlx::Session::compression_algorithm_t::LZ4_MESSAGE:
+      compress_caps.m_algorithm = "lz4_message";
+      m_protocol.snd_CapabilitiesSet(compress_caps).wait();
+      break;
+    case cdk::mysqlx::Session::compression_algorithm_t::ZSTD_STREAM:
+      compress_caps.m_algorithm = "zstd_stream";
+      m_protocol.snd_CapabilitiesSet(compress_caps).wait();
+      break;
+   case cdk::mysqlx::Session::compression_algorithm_t::NONE:
+      break;
+    default:
+      //Add algorithm name here
+      assert(false);
+    }
+  }
+
+
+  for(auto alg = algorithms.rbegin(); alg != algorithms.rend(); ++alg)
+  {
+    switch (*alg) {
+    case cdk::mysqlx::Session::compression_algorithm_t::DEFLATE_STREAM:
+      m_protocol.rcv_Reply(cap_prc).wait();
+      if (cap_prc.m_compression_ok)
+        compression = Compression_type::DEFLATE;
+      break;
+    case cdk::mysqlx::Session::compression_algorithm_t::LZ4_MESSAGE:
+      m_protocol.rcv_Reply(cap_prc).wait();
+      if (cap_prc.m_compression_ok)
+        compression = Compression_type::LZ4;
+      break;
+    case cdk::mysqlx::Session::compression_algorithm_t::ZSTD_STREAM:
+      m_protocol.rcv_Reply(cap_prc).wait();
+      if (cap_prc.m_compression_ok)
+        compression = Compression_type::ZSTD;
+      break;
+    case cdk::mysqlx::Session::compression_algorithm_t::NONE:
+       break;
+    default:
+      //New algorithm adder....
+      assert(false);
+    }
+  }
 
   return compression;
 }

common/session.cc

@@ -241,6 +241,29 @@ TCPIP_options::compression_mode_t get_compression(unsigned m)
   return CDK_type(0); // quiet compiler warnings
 }
 
+TCPIP_options::compression_algorithm_t get_compression_algorithm(std::string alg)
+{
+  std::string algorithm_name = to_upper(alg);
+
+  using CDK_type = TCPIP_options::compression_algorithm_t;
+
+  static std::map<std::string,CDK_type> alg_map= {
+    {"DEFLATE_STREAM",CDK_type::DEFLATE_STREAM},
+    {"DEFLATE",CDK_type::DEFLATE_STREAM}, //ALIAS
+    {"LZ4_MESSAGE",CDK_type::LZ4_MESSAGE},
+    {"LZ4",CDK_type::LZ4_MESSAGE}, //ALIAS
+    {"ZSTD_STREAM",CDK_type::ZSTD_STREAM},
+    {"ZSTD",CDK_type::ZSTD_STREAM}, //ALIAS
+  };
+
+  auto it = alg_map.find(algorithm_name);
+
+  if(it == alg_map.end())
+    return CDK_type::NONE;
+
+  return it->second;
+}
+
 TLS_options::SSL_MODE get_ssl_mode(unsigned m)
 {
   using DevAPI_type = Settings_impl::SSL_mode;
@@ -395,6 +418,28 @@ void prepare_options(
       (unsigned)settings.get(Option::COMPRESSION).get_uint()));
   }
 
+  if (settings.has_option(Option::COMPRESSION_ALGORITHMS))
+  {
+    bool has_algs = false;
+    for (const auto &opt_val : settings)
+    {
+      switch(opt_val.first)
+      {
+      case Option::COMPRESSION_ALGORITHMS:
+        has_algs = true;
+        opts.add_compression_alg(
+              get_compression_algorithm(
+                opt_val.second.get_string())
+              );
+      }
+    }
+    if(!has_algs)
+    {
+      //Inform that option was used but nothing was set
+      opts.add_compression_alg(TCPIP_options::NONE);
+    }
+  }
+
   // DNS+SRV
 
   if(settings.has_option(Option::DNS_SRV))

common/settings.h

@@ -313,6 +313,11 @@ class Settings_impl::Setter
       m_data.m_tls_vers = true;
       break;
 
+    case Session_option_impl::COMPRESSION_ALGORITHMS:
+      m_multi = !m_data.m_compression_algorithms;
+      m_data.m_compression_algorithms = true;
+      break;
+
     default:
       {
         std::stringstream err_msg;
@@ -1076,6 +1081,23 @@ Settings_impl::Setter::set_option<
     add_option((int)Settings_impl::Session_option_impl::TLS_VERSIONS, val);
 }
 
+template<>
+inline void
+Settings_impl::Setter::set_option<
+  Settings_impl::Session_option_impl::COMPRESSION_ALGORITHMS
+>(const std::string &val)
+{
+  m_data.m_compression_algorithms = true;  // record that the option was set
+
+  // If in multi mode, the value is a single list element, otherwise
+  // the value can be a comma separated list
+
+  if (!m_multi)
+    set_comma_separated((int)Settings_impl::Session_option_impl::COMPRESSION_ALGORITHMS, val);
+  else
+    add_option((int)Settings_impl::Session_option_impl::COMPRESSION_ALGORITHMS, val);
+}
+
 
 // Generic add_option() method.
 
@@ -1098,6 +1120,7 @@ void Settings_impl::Setter::add_option(int opt, const T &val)
 
   case Session_option_impl::TLS_CIPHERSUITES:
   case Session_option_impl::TLS_VERSIONS:
+  case Session_option_impl::COMPRESSION_ALGORITHMS:
     if (m_multi)
     {
       options.emplace_back(opt, val);
@@ -1237,6 +1260,9 @@ void Settings_impl::Setter::null()
   case Session_option_impl::USER:
     throw_error("Option ... can not be unset");
     break;
+  case Session_option_impl::COMPRESSION_ALGORITHMS:
+    //It has compression algorithms, so don't use default
+    m_data.m_compression_algorithms = true;
   case Session_option_impl::LAST:
     break;
   default:
@@ -1373,7 +1399,7 @@ void Settings_impl::Setter::key_val(const std::string &key, const std::string &v
         }
         break;
       default:
-        key_val(get_uri_option(key))->scalar()->str(val);
+        key_val(option)->scalar()->str(val);
     }
   }
   catch (const std::out_of_range&)
@@ -1427,6 +1453,7 @@ void Settings_impl::Setter::key_val(const std::string &key,
 
       case Settings_impl::Session_option_impl::TLS_CIPHERSUITES:
       case Settings_impl::Session_option_impl::TLS_VERSIONS:
+      case Settings_impl::Session_option_impl::COMPRESSION_ALGORITHMS:
         {
           auto *prc = key_val(option)->arr();
           if (!prc)