Merge branch 'wl12497'

Author: Bogdan Degtyariov

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

@@ -550,6 +550,7 @@ class Session
   void check_protocol_fields();
   bool has_prepared_statements();
   void set_has_prepared_statements(bool);
+  bool has_keep_open();
 
   /*
     Clear diagnostic information that accumulated for the session.
@@ -562,7 +563,6 @@ class Session
   { m_da.clear(); }
 
   void reset();
-
   void close();
 
   /*

cdk/include/mysql/cdk/protocol/mysqlx.h

@@ -460,7 +460,8 @@ struct Protocol_fields
     Enum values will be used as binary flags,
     so they must be as 2^N
   */
-  enum value { ROW_LOCKING = 1 , UPSERT = 2, PREPARED_STATEMENTS = 4 };
+  enum value { ROW_LOCKING = 1 , UPSERT = 2, PREPARED_STATEMENTS = 4,
+               KEEP_OPEN = 8 };
 };
 
 }  // api namespace
@@ -533,8 +534,12 @@ class Protocol
   Op& snd_CapabilitiesSet(const api::Any::Document& caps);
   Op& snd_AuthenticateStart(const char* mechanism, bytes data, bytes response);
   Op& snd_AuthenticateContinue(bytes data);
-  Op& snd_SessionReset();
-  Op& snd_Close();
+  // Depending on whether the server supports keep_open in
+  // SESS_RESET message the session close has to be handled
+  // differently.
+  Op& snd_SessionReset(bool keep_open = false);
+  Op& snd_SessionClose();
+  Op& snd_ConnectionClose();
 
   /**
     Send protocol command which executes a statement.

cdk/mysqlx/session.cc

@@ -60,7 +60,7 @@ struct Proto_field_checker : public cdk::protocol::mysqlx::api::Expectations
     unsigned int m_code = 0;
 
     void error(unsigned int code, short int,
-                cdk::protocol::mysqlx::sql_state_t, const string &)
+               cdk::protocol::mysqlx::sql_state_t, const string &)
     {
       m_code = code;
     }
@@ -79,26 +79,29 @@ struct Proto_field_checker : public cdk::protocol::mysqlx::api::Expectations
   }
 
   /*
-    This method sets the expectation and returns
-    the field flag if it is supported, otherwise 0 is returned.
+  This method sets the expectation and returns
+  the field flag if it is supported, otherwise 0 is returned.
   */
   uint64_t is_supported(Protocol_fields::value v)
   {
     switch (v)
     {
-      case Protocol_fields::ROW_LOCKING:
-        // Find=17, locking=12
-        m_data = bytes("17.12");
-        break;
-      case Protocol_fields::UPSERT:
-        // Insert=18, upsert=6
-        m_data = bytes("18.6");
-        break;
-      case Protocol_fields::PREPARED_STATEMENTS:
-        m_data = bytes("40");
-        break;
-      default:
-        return 0;
+    case Protocol_fields::ROW_LOCKING:
+      // Find=17, locking=12
+      m_data = bytes("17.12");
+      break;
+    case Protocol_fields::UPSERT:
+      // Insert=18, upsert=6
+      m_data = bytes("18.6");
+      break;
+    case Protocol_fields::PREPARED_STATEMENTS:
+      m_data = bytes("40");
+      break;
+    case Protocol_fields::KEEP_OPEN:
+      m_data = bytes("6.1");
+      break;
+    default:
+      return 0;
     }
     m_proto.snd_Expect_Open(*this, false).wait();
 
@@ -109,8 +112,8 @@ struct Proto_field_checker : public cdk::protocol::mysqlx::api::Expectations
     if (prc.m_code == 0 || prc.m_code == 5168)
     {
       /*
-        The expectation block needs to be closed if no error
-        or expectation failed error (5168)
+      The expectation block needs to be closed if no error
+      or expectation failed error (5168)
       */
       m_proto.snd_Expect_Close().wait();
       m_proto.rcv_Reply(prc).wait();
@@ -346,6 +349,7 @@ class AuthSha256Memory
 
 };
 
+
 /*
    Class Session
 */
@@ -457,17 +461,18 @@ option_t Session::is_valid()
 
 void Session::check_protocol_fields()
 {
+  wait();
+  if (0 < entry_count())
+    get_error().rethrow();
   if (m_proto_fields == UINT64_MAX)
   {
-    wait();
-    if (0 < entry_count())
-      get_error().rethrow();
     Proto_field_checker field_checker(m_protocol);
     m_proto_fields = 0;
     /* More fields checks will be added here */
     m_proto_fields |= field_checker.is_supported(Protocol_fields::ROW_LOCKING);
     m_proto_fields |= field_checker.is_supported(Protocol_fields::UPSERT);
     m_proto_fields |= field_checker.is_supported(Protocol_fields::PREPARED_STATEMENTS);
+    m_proto_fields |= field_checker.is_supported(Protocol_fields::KEEP_OPEN);
   }
 }
 
@@ -485,6 +490,12 @@ void Session::set_has_prepared_statements(bool x)
     m_proto_fields &= ~Protocol_fields::PREPARED_STATEMENTS;
 }
 
+bool Session::has_keep_open()
+{
+  check_protocol_fields();
+  return (m_proto_fields & Protocol_fields::KEEP_OPEN) != 0;
+}
+
 
 option_t Session::check_valid()
 {
@@ -495,21 +506,23 @@ option_t Session::check_valid()
 
 void Session::reset()
 {
+  check_protocol_fields(); // This will be used for lazy checks
   clear_errors();
 
   m_reply_op_queue.clear();
 
   if (is_valid())
   {
-    m_protocol.snd_SessionReset().wait();
-    m_protocol.rcv_Reply(*this).wait();
+    m_protocol.snd_SessionReset(has_keep_open()).wait();
 
-    m_isvalid = false;
+    m_protocol.rcv_Reply(*this).wait();
 
-    // Re-authenticate
-    send_auth();
+    if (!has_keep_open())
+    {
+      m_isvalid = false;
+      send_auth();  // Re-authenticate for servers not supporting keep-open
+    }
   }
-
 }
 
 
@@ -519,7 +532,7 @@ void Session::close()
 
   if (is_valid())
   {
-    m_protocol.snd_Close().wait();
+    m_protocol.snd_ConnectionClose().wait();
     m_protocol.rcv_Reply(*this).wait();
   }
   m_isvalid = false;

cdk/protocol/mysqlx/pb/mysqlx_session.proto

@@ -90,9 +90,13 @@ message AuthenticateOk {
 }
 
 // reset the current session
+// param keep_open: if is true the session will be reset, but stays authenticated.
+//       otherwise, the session will be closed and needs to be authenticated again.
 //
 // :Returns: :protobuf:msg:`Mysqlx::Ok`
 message Reset {
+  optional bool keep_open = 1 [ default = false ];
+
   option (client_message_id) = SESS_RESET; // comment_out_if PROTOBUF_LITE
 }
 

cdk/protocol/mysqlx/protocol.cc

@@ -855,13 +855,20 @@ class Rcv_reply : public Op_rcv
 };
 
 
-Protocol::Op& Protocol::snd_SessionReset()
+Protocol::Op& Protocol::snd_SessionReset(bool keep_open)
 {
   Mysqlx::Session::Reset reset;
+  reset.set_keep_open(keep_open);
   return get_impl().snd_start(reset, msg_type::cli_SessionReset);
 }
 
-Protocol::Op& Protocol::snd_Close()
+Protocol::Op& Protocol::snd_SessionClose()
+{
+  Mysqlx::Session::Close close;
+  return get_impl().snd_start(close, msg_type::cli_SessionClose);
+}
+
+Protocol::Op& Protocol::snd_ConnectionClose()
 {
   Mysqlx::Connection::Close close;
   return get_impl().snd_start(close, msg_type::cli_Close);
@@ -940,7 +947,6 @@ Protocol::Op& Protocol_server::rcv_Command(Cmd_processor &prc)
   return get_impl().rcv_start<Rcv_command>(prc);
 }
 
-
 // ------------------------------------------------------------
 
 size_t Processor_base::message_begin_internal(msg_type_t type, bool &flag)

common/session.cc

@@ -536,7 +536,6 @@ void Session_pool::release_session(cdk::shared_ptr<cdk::Session> &sess)
 
     if (el != m_pool.end())
       try {
-      el->first->reset();
       el->second = system_clock::now() + m_time_to_live;
     }
     catch (...) {
@@ -577,6 +576,7 @@ std::shared_ptr<cdk::Session> Session_pool::get_session()
     if (it->first.unique())
     {
       try {
+        it->first->reset();
         if(!it->first->is_valid())
         {
           throw "Remove this";