connection: introduce SelfIdentity and new options

SelfIdentity is intended to group STARTUP options that can be used
to uniquely identify driver, application and particular client instance.

DRIVER_NAME and DRIVER_VERSION were already present, but they were
hardcoded to "scylla-rust-driver" and current crate version, retrieved
from Cargo. Now it is possible to set custom values for them, which
can be useful for custom driver builds, as well as for drivers running
on top of Rust driver (think cpp-rust-driver).

The remaining options are inspired by cpp-driver, and added for
cpp-rust-driver purposes as well.

APPLICATION_NAME and APPLICATION_VERSION identify a single build of
an application using the driver.

CLIENT_ID uniquely identifies a single instance of an application.

All the above information are visible in Cassandra in
`system_views.clients` in `client_options` column.
DRIVER_NAME and DRIVER_VERSION are visible in ScyllaDB in
`system.clients`.

scylla-cql/src/frame/request/options.rs

@@ -20,6 +20,9 @@ pub const COMPRESSION: &str = "COMPRESSION";
 pub const CQL_VERSION: &str = "CQL_VERSION";
 pub const DRIVER_NAME: &str = "DRIVER_NAME";
 pub const DRIVER_VERSION: &str = "DRIVER_VERSION";
+pub const APPLICATION_NAME: &str = "APPLICATION_NAME";
+pub const APPLICATION_VERSION: &str = "APPLICATION_VERSION";
+pub const CLIENT_ID: &str = "CLIENT_ID";
 
 /* Value names for options in SUPPORTED/STARTUP */
 pub const DEFAULT_CQL_PROTOCOL_VERSION: &str = "4.0.0";

scylla/src/transport/connection.rs

@@ -350,6 +350,136 @@ mod ssl_config {
     }
 }
 
+/// Driver and application self-identifying information,
+/// to be sent in STARTUP message.
+#[derive(Debug, Clone, Default)]
+pub struct SelfIdentity<'id> {
+    // Custom driver identity can be set if a custom driver build is running,
+    // or an entirely different driver is operating on top of Rust driver
+    // (e.g. cpp-rust-driver).
+    custom_driver_name: Option<Cow<'id, str>>,
+    custom_driver_version: Option<Cow<'id, str>>,
+
+    // Application identity can be set to distinguish different applications
+    // connected to the same cluster.
+    application_name: Option<Cow<'id, str>>,
+    application_version: Option<Cow<'id, str>>,
+
+    // A (unique) client ID can be set to distinguish different instances
+    // of the same application connected to the same cluster.
+    client_id: Option<Cow<'id, str>>,
+}
+
+impl<'id> SelfIdentity<'id> {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Advertises a custom driver name, which can be used if a custom driver build is running,
+    /// or an entirely different driver is operating on top of Rust driver
+    /// (e.g. cpp-rust-driver).
+    pub fn set_custom_driver_name(&mut self, custom_driver_name: Cow<'id, str>) {
+        self.custom_driver_name = Some(custom_driver_name);
+    }
+
+    /// Advertises a custom driver name. See [Self::set_custom_driver_name] for use cases.
+    pub fn with_custom_driver_name(mut self, custom_driver_name: Cow<'id, str>) -> Self {
+        self.custom_driver_name = Some(custom_driver_name);
+        self
+    }
+
+    /// Advertises a custom driver version. See [Self::set_custom_driver_name] for use cases.
+    pub fn set_custom_driver_version(&mut self, custom_driver_version: Cow<'id, str>) {
+        self.custom_driver_version = Some(custom_driver_version);
+    }
+
+    /// Advertises a custom driver version. See [Self::set_custom_driver_name] for use cases.
+    pub fn with_custom_driver_version(mut self, custom_driver_version: Cow<'id, str>) -> Self {
+        self.custom_driver_version = Some(custom_driver_version);
+        self
+    }
+
+    /// Advertises an application name, which can be used to distinguish different applications
+    /// connected to the same cluster.
+    pub fn set_application_name(&mut self, application_name: Cow<'id, str>) {
+        self.application_name = Some(application_name);
+    }
+
+    /// Advertises an application name. See [Self::set_application_name] for use cases.
+    pub fn with_application_name(mut self, application_name: Cow<'id, str>) -> Self {
+        self.application_name = Some(application_name);
+        self
+    }
+
+    /// Advertises an application version. See [Self::set_application_name] for use cases.
+    pub fn set_application_version(&mut self, application_version: Cow<'id, str>) {
+        self.application_version = Some(application_version);
+    }
+
+    /// Advertises an application version. See [Self::set_application_name] for use cases.
+    pub fn with_application_version(mut self, application_version: Cow<'id, str>) -> Self {
+        self.application_version = Some(application_version);
+        self
+    }
+
+    /// Advertises a client ID, which can be set to distinguish different instances
+    /// of the same application connected to the same cluster.
+    pub fn set_client_id(&mut self, client_id: Cow<'id, str>) {
+        self.client_id = Some(client_id);
+    }
+
+    /// Advertises a client ID. See [Self::set_client_id] for use cases.
+    pub fn with_client_id(mut self, client_id: Cow<'id, str>) -> Self {
+        self.client_id = Some(client_id);
+        self
+    }
+}
+
+impl<'id: 'map, 'map> SelfIdentity<'id> {
+    fn add_startup_options(&'id self, options: &'map mut HashMap<Cow<'id, str>, Cow<'id, str>>) {
+        /* Driver identity. */
+        let driver_name = self
+            .custom_driver_name
+            .as_deref()
+            .unwrap_or(options::DEFAULT_DRIVER_NAME);
+        options.insert(
+            Cow::Borrowed(options::DRIVER_NAME),
+            Cow::Borrowed(driver_name),
+        );
+
+        let driver_version = self
+            .custom_driver_version
+            .as_deref()
+            .or(options::DEFAULT_DRIVER_VERSION);
+        if let Some(driver_version) = driver_version {
+            options.insert(
+                Cow::Borrowed(options::DRIVER_VERSION),
+                Cow::Borrowed(driver_version),
+            );
+        }
+
+        /* Application identity. */
+        if let Some(application_name) = self.application_name.as_deref() {
+            options.insert(
+                Cow::Borrowed(options::APPLICATION_NAME),
+                Cow::Borrowed(application_name),
+            );
+        }
+
+        if let Some(application_version) = self.application_version.as_deref() {
+            options.insert(
+                Cow::Borrowed(options::APPLICATION_VERSION),
+                Cow::Borrowed(application_version),
+            );
+        }
+
+        /* Client identity. */
+        if let Some(client_id) = self.client_id.as_deref() {
+            options.insert(Cow::Borrowed(options::CLIENT_ID), Cow::Borrowed(client_id));
+        }
+    }
+}
+
 #[derive(Clone)]
 pub(crate) struct ConnectionConfig {
     pub(crate) compression: Option<Compression>,
@@ -370,6 +500,8 @@ pub(crate) struct ConnectionConfig {
     pub(crate) keepalive_interval: Option<Duration>,
     pub(crate) keepalive_timeout: Option<Duration>,
     pub(crate) tablet_sender: Option<mpsc::Sender<(TableSpec<'static>, RawTablet)>>,
+
+    pub(crate) identity: SelfIdentity<'static>,
 }
 
 impl Default for ConnectionConfig {
@@ -394,6 +526,8 @@ impl Default for ConnectionConfig {
             keepalive_timeout: None,
 
             tablet_sender: None,
+
+            identity: SelfIdentity::default(),
         }
     }
 }
@@ -1527,25 +1661,9 @@ pub(crate) async fn open_connection(
     source_port: Option<u16>,
     config: &ConnectionConfig,
 ) -> Result<(Connection, ErrorReceiver), QueryError> {
+    /* Translate the address, if applicable. */
     let addr = maybe_translated_addr(endpoint, config.address_translator.as_deref()).await?;
-    open_named_connection(
-        addr,
-        source_port,
-        config,
-        Some("scylla-rust-driver"),
-        option_env!("CARGO_PKG_VERSION"),
-    )
-    .await
-}
 
-/// The same as `open_connection`, but with customizable driver name and version.
-pub(crate) async fn open_named_connection(
-    addr: SocketAddr,
-    source_port: Option<u16>,
-    config: &ConnectionConfig,
-    driver_name: Option<&str>,
-    driver_version: Option<&str>,
-) -> Result<(Connection, ErrorReceiver), QueryError> {
     /* Setup connection on TCP level and prepare for sending/receiving CQL frames. */
     let (mut connection, error_receiver) =
         Connection::new(addr, source_port, config.clone()).await?;
@@ -1605,19 +1723,8 @@ pub(crate) async fn open_named_connection(
         Cow::Borrowed(options::DEFAULT_CQL_PROTOCOL_VERSION),
     );
 
-    // Driver identity.
-    if let Some(driver_name) = driver_name {
-        options.insert(
-            Cow::Borrowed(options::DRIVER_NAME),
-            Cow::Borrowed(driver_name),
-        );
-    }
-    if let Some(driver_version) = driver_version {
-        options.insert(
-            Cow::Borrowed(options::DRIVER_VERSION),
-            Cow::Borrowed(driver_version),
-        );
-    }
+    // Application & driver's identity.
+    config.identity.add_startup_options(&mut options);
 
     // Optional compression.
     if let Some(compression) = &config.compression {

scylla/src/transport/mod.rs

@@ -19,6 +19,7 @@ pub mod speculative_execution;
 pub mod topology;
 
 pub use crate::frame::{Authenticator, Compression};
+pub use connection::SelfIdentity;
 pub use execution_profile::ExecutionProfile;
 pub use scylla_cql::errors;
 

scylla/src/transport/session.rs

@@ -46,7 +46,7 @@ use super::node::KnownNode;
 use super::partitioner::PartitionerName;
 use super::query_result::MaybeFirstRowTypedError;
 use super::topology::UntranslatedPeer;
-use super::NodeRef;
+use super::{NodeRef, SelfIdentity};
 use crate::cql_to_rust::FromRow;
 use crate::frame::response::cql_to_rust::FromRowError;
 use crate::frame::response::result;
@@ -515,6 +515,8 @@ impl Session {
             keepalive_interval: config.keepalive_interval,
             keepalive_timeout: config.keepalive_timeout,
             tablet_sender: Some(tablet_sender),
+            // A temporary stub, removed in the next commit.
+            identity: SelfIdentity::default(),
         };
 
         let pool_config = PoolConfig {