cluster: un-pub Cluster and make it cloneable

The Cluster struct by itself only serves as a facade for the
ClusterWorker, i.e. it has channels that allow sending requests to the
worker, receives the ClusterData via the `data` field etc. Apart from
the `_worker_handle` field, all other fields are cloneable. Two tasks
working on two copies of the same Cluster object should behave the same
as if they shared and operated on a single Cluster object (e.g. via
Arc<Cluster>).

This commit makes the Cluster object cloneable - the `_worker_handle` is
shared via an Arc. This will be very useful in the next commit - we will
do a similar thing for the GenericSession object.

scylla/src/transport/cluster.rs

@@ -40,20 +40,29 @@ use super::topology::Strategy;
 
 /// Cluster manages up to date information and connections to database nodes.
 /// All data can be accessed by cloning Arc<ClusterData> in the `data` field
-pub struct Cluster {
+//
+// NOTE: This structure was intentionally made cloneable. The reason for this
+// is to make it possible to use two different Session APIs in the same program
+// that share the same session resources.
+//
+// It is safe to do because the Cluster struct is just a facade for the real,
+// "semantic" Cluster object. Cloned instance of this struct will use the same
+// ClusterData and worker and will observe the same state.
+#[derive(Clone)]
+pub(crate) struct Cluster {
     // `ArcSwap<ClusterData>` is wrapped in `Arc` to support sharing cluster data
     // between `Cluster` and `ClusterWorker`
     data: Arc<ArcSwap<ClusterData>>,
 
     refresh_channel: tokio::sync::mpsc::Sender<RefreshRequest>,
     use_keyspace_channel: tokio::sync::mpsc::Sender<UseKeyspaceRequest>,
 
-    _worker_handle: RemoteHandle<()>,
+    _worker_handle: Arc<RemoteHandle<()>>,
 }
 
 /// Enables printing [Cluster] struct in a neat way, by skipping the rather useless
 /// print of channels state and printing [ClusterData] neatly.
-pub struct ClusterNeatDebug<'a>(pub &'a Cluster);
+pub(crate) struct ClusterNeatDebug<'a>(pub &'a Cluster);
 impl<'a> std::fmt::Debug for ClusterNeatDebug<'a> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let cluster = self.0;
@@ -193,7 +202,7 @@ impl Cluster {
             data: cluster_data,
             refresh_channel: refresh_sender,
             use_keyspace_channel: use_keyspace_sender,
-            _worker_handle: worker_handle,
+            _worker_handle: Arc::new(worker_handle),
         };
 
         Ok(result)