[Merged by Bors] - Relax mutability requirements of FluvioAdmin client

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Platform Version 0.8.5 - UNRELEASED
 * Add unstable Admin Watch API for topics, partitions, and SPUs ([#1136](https://github.com/infinyon/fluvio/pull/1136))
 * Make recipes for smoke tests no longer build by default, helps caching. ([#1165](https://github.com/infinyon/fluvio/pull/1165))
+* Relax requirement of `FluvioAdmin` methods from `&mut self` to `&self`. ([#1178](https://github.com/infinyon/fluvio/pull/1178))
 
 ## Platform Version 0.8.4 - 2020-05-29
 * Don't hang when check for non exist topic. ([#697](https://github.com/infinyon/fluvio/pull/697))

src/cli/src/consumer/partition/list.rs

@@ -27,7 +27,7 @@ impl ListPartitionOpt {
         O: Terminal,
     {
         let output = self.output.format;
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
 
         let partitions = admin.list::<PartitionSpec, _>(vec![]).await?;
 

src/cli/src/consumer/topic/create.rs

@@ -88,7 +88,7 @@ impl CreateTopicOpt {
         let (name, topic_spec) = self.validate()?;
 
         debug!("creating topic: {} spec: {:#?}", name, topic_spec);
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
         admin.create(name.clone(), dry_run, topic_spec).await?;
         println!("topic \"{}\" created", name);
 

src/cli/src/consumer/topic/delete.rs

@@ -21,7 +21,7 @@ pub struct DeleteTopicOpt {
 impl DeleteTopicOpt {
     pub async fn process(self, fluvio: &Fluvio) -> Result<()> {
         debug!("deleting topic: {}", &self.topic);
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
         admin.delete::<TopicSpec, _>(&self.topic).await?;
         println!("topic \"{}\" deleted", &self.topic);
         Ok(())

src/cli/src/consumer/topic/describe.rs

@@ -35,7 +35,7 @@ impl DescribeTopicsOpt {
         let output_type = self.output.format;
         debug!("describe topic: {}, {}", topic, output_type);
 
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
         let topics = admin.list::<TopicSpec, _>(vec![topic]).await?;
 
         display::describe_topics(topics, output_type, out).await?;

src/cli/src/consumer/topic/list.rs

@@ -30,7 +30,7 @@ impl ListTopicsOpt {
     pub async fn process<O: Terminal>(self, out: Arc<O>, fluvio: &Fluvio) -> Result<()> {
         let output_type = self.output.format;
         debug!("list topics {:#?} ", output_type);
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
 
         let topics = admin.list::<TopicSpec, _>(vec![]).await?;
         display::format_response_output(out, topics, output_type)?;

src/client/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fluvio"
-version = "0.8.5"
+version = "0.8.6"
 edition = "2018"
 license = "Apache-2.0"
 authors = ["Fluvio Contributors <team@fluvio.io>"]

src/client/src/admin.rs

@@ -125,7 +125,7 @@ impl FluvioAdmin {
     }
 
     #[instrument(skip(self, request))]
-    async fn send_receive<R>(&mut self, request: R) -> Result<R::Response, FlvSocketError>
+    async fn send_receive<R>(&self, request: R) -> Result<R::Response, FlvSocketError>
     where
         R: AdminRequest + Send + Sync,
     {
@@ -134,12 +134,7 @@ impl FluvioAdmin {
 
     /// create new object
     #[instrument(skip(self, name, dry_run, spec))]
-    pub async fn create<S>(
-        &mut self,
-        name: String,
-        dry_run: bool,
-        spec: S,
-    ) -> Result<(), FluvioError>
+    pub async fn create<S>(&self, name: String, dry_run: bool, spec: S) -> Result<(), FluvioError>
     where
         S: Into<AllCreatableSpec>,
     {
@@ -157,7 +152,7 @@ impl FluvioAdmin {
     /// delete object by key
     /// key is depend on spec, most are string but some allow multiple types
     #[instrument(skip(self, key))]
-    pub async fn delete<S, K>(&mut self, key: K) -> Result<(), FluvioError>
+    pub async fn delete<S, K>(&self, key: K) -> Result<(), FluvioError>
     where
         S: DeleteSpec,
         K: Into<S::DeleteKey>,
@@ -168,7 +163,7 @@ impl FluvioAdmin {
     }
 
     #[instrument(skip(self, filters))]
-    pub async fn list<S, F>(&mut self, filters: F) -> Result<Vec<Metadata<S>>, FluvioError>
+    pub async fn list<S, F>(&self, filters: F) -> Result<Vec<Metadata<S>>, FluvioError>
     where
         S: ListSpec + Encoder + Decoder,
         S::Status: Encoder + Decoder,

src/client/src/producer.rs

@@ -122,7 +122,7 @@ impl TopicProducer {
         let requests = assemble_requests(&self.topic, partitions_by_spu);
 
         for (leader, request) in requests {
-            let mut spu_client = self.pool.create_serial_socket_from_leader(leader).await?;
+            let spu_client = self.pool.create_serial_socket_from_leader(leader).await?;
             spu_client.send_receive(request).await?;
         }
 

src/client/src/sockets.rs

@@ -209,7 +209,7 @@ impl VersionedSerialSocket {
 
     /// send and wait for reply serially
     #[instrument(level = "trace", skip(self, request))]
-    pub async fn send_receive<R>(&mut self, request: R) -> Result<R::Response, FlvSocketError>
+    pub async fn send_receive<R>(&self, request: R) -> Result<R::Response, FlvSocketError>
     where
         R: Request + Send + Sync,
     {

src/cluster/src/cli/group/create.rs

@@ -44,7 +44,7 @@ impl CreateManagedSpuGroupOpt {
         let (name, spec) = self.validate();
         debug!("creating spg: {}, spec: {:#?}", name, spec);
 
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
         admin.create(name, false, spec).await?;
 
         Ok(())

src/cluster/src/cli/group/delete.rs

@@ -23,7 +23,7 @@ pub struct DeleteManagedSpuGroupOpt {
 
 impl DeleteManagedSpuGroupOpt {
     pub async fn process(self, fluvio: &Fluvio) -> Result<(), ClusterCliError> {
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
         admin.delete::<SpuGroupSpec, _>(&self.name).await?;
         Ok(())
     }

src/cluster/src/cli/group/list.rs

@@ -26,7 +26,7 @@ impl ListManagedSpuGroupsOpt {
         out: Arc<O>,
         fluvio: &Fluvio,
     ) -> Result<(), ClusterCliError> {
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
         let lists = admin.list::<SpuGroupSpec, _>(vec![]).await?;
 
         output::spu_group_response_to_output(out, lists, self.output.format)

src/cluster/src/cli/spu/list.rs

@@ -34,7 +34,7 @@ impl ListSpusOpt {
         out: Arc<O>,
         fluvio: &Fluvio,
     ) -> Result<(), ClusterCliError> {
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
 
         let spus = if self.custom {
             // List custom SPUs only

src/cluster/src/cli/spu/register.rs

@@ -38,7 +38,7 @@ pub struct RegisterCustomSpuOpt {
 impl RegisterCustomSpuOpt {
     pub async fn process(self, fluvio: &Fluvio) -> Result<(), ClusterCliError> {
         let (name, spec) = self.validate()?;
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
         admin.create(name, false, spec).await?;
         Ok(())
     }

src/cluster/src/cli/spu/unregister.rs

@@ -35,7 +35,7 @@ pub struct UnregisterCustomSpuOpt {
 impl UnregisterCustomSpuOpt {
     pub async fn process(self, fluvio: &Fluvio) -> Result<(), ClusterCliError> {
         let delete_key = self.validate()?;
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
         admin.delete::<CustomSpuSpec, _>(delete_key).await?;
         Ok(())
     }

src/cluster/src/start/k8.rs

@@ -1334,7 +1334,7 @@ impl ClusterInstaller {
         debug!("trying to create managed spu: {:#?}", cluster);
         let name = self.config.group_name.clone();
         let fluvio = Fluvio::connect_with_config(cluster).await?;
-        let mut admin = fluvio.admin().await;
+        let admin = fluvio.admin().await;
 
         let spu_spec = SpuGroupSpec {
             replicas: self.config.spu_replicas,

src/cluster/src/start/local.rs

@@ -661,7 +661,7 @@ impl LocalInstaller {
         sleep(Duration::from_secs(delay)).await;
 
         let client = Fluvio::connect().await?;
-        let mut admin = client.admin().await;
+        let admin = client.admin().await;
 
         // wait for list of spu
         for _ in 0..30u16 {

tests/runner/src/tests/smoke/consume.rs

@@ -171,7 +171,7 @@ async fn validate_consume_message_api(
     // wait 500m second and ensure partition list
     sleep(Duration::from_millis(500)).await;
 
-    let mut admin = client.admin().await;
+    let admin = client.admin().await;
     let partitions = admin
         .list::<PartitionSpec, _>(vec![])
         .await

tests/runner/src/utils/test_runner.rs

@@ -36,7 +36,7 @@ impl FluvioTest {
             println!("Creating the topic: {}", &option.topic_name);
         }
 
-        let mut admin = client.admin().await;
+        let admin = client.admin().await;
         let topic_spec = TopicSpec::new_computed(1, option.replication() as i32, None);
 
         let topic_create = admin