Separate channel creation and state probe, and error handling thereof. (#2465)

## Motivation

Currently the `Channel` object is calling into the `PersistentChannel`
internals to do things like provision an active channel and check state.
The division of where work and error handling is managed becomes muddied
in this situation, which complicates handling errors in ways that aren't
`unwrap`.

This was also motivated by work to implement the always-failing
PersistentChannel PR, but it seemed like it could be broken off into
it's own PR.

## Solution

Moving the actual mechanics into the `PersistentChannel` object, which
owns channel config, and allows `Channel` to be a lightweight handles
for users to interact with. Specifically state probing and channel
creation are now in `PersistentChannel`.

`PersistentChannel` calls are wrapped in `Result` objects that are
translated at the `Channel` level to the appropriate public-facing API
objects. `get_or_create_active_channel()` just becomes
`get_active_channel()` and passes the 'create' flag, which controls
whether a new channel is created if it does not already exist.

The external semantics should stay the same as they have been.

This does not yet address the error that can surface in `call()`,
because it's not yet clear what the error reporting story is.

Author: nathanielford

grpc/src/client/channel.rs

@@ -154,23 +154,10 @@ impl Channel {
 
     // TODO: enter_idle(&self) and graceful_stop()?
 
-    /// Returns the current state of the channel.
+    /// Returns the current state of the channel. If there is no underlying active channel,
+    /// returns Idle. If `connect` is true, will create a new active channel.
     pub fn state(&mut self, connect: bool) -> ConnectivityState {
-        let ac = if !connect {
-            // If !connect and we have no active channel already, return idle.
-            let ac = self.inner.active_channel.lock().unwrap();
-            if ac.is_none() {
-                return ConnectivityState::Idle;
-            }
-            ac.as_ref().unwrap().clone()
-        } else {
-            // Otherwise, get or create the active channel.
-            self.get_or_create_active_channel()
-        };
-        if let Some(s) = ac.connectivity_state.cur() {
-            return s;
-        }
-        ConnectivityState::Idle
+        self.inner.state(connect)
     }
 
     /// Waits for the state of the channel to change from source.  Times out and
@@ -183,20 +170,8 @@ impl Channel {
         todo!()
     }
 
-    fn get_or_create_active_channel(&self) -> Arc<ActiveChannel> {
-        let mut s = self.inner.active_channel.lock().unwrap();
-        if s.is_none() {
-            *s = Some(ActiveChannel::new(
-                self.inner.target.clone(),
-                &self.inner.options,
-                self.inner.runtime.clone(),
-            ));
-        }
-        s.clone().unwrap()
-    }
-
     pub async fn call(&self, method: String, request: Request) -> Response {
-        let ac = self.get_or_create_active_channel();
+        let ac = self.inner.get_active_channel();
         ac.call(method, request).await
     }
 }
@@ -213,8 +188,8 @@ struct PersistentChannel {
 }
 
 impl PersistentChannel {
-    // Channels begin idle so new is a simple constructor.  Required parameters
-    // are not in ChannelOptions.
+    // Channels begin idle so `new()` does not automatically connect.
+    // ChannelOption contain only optional parameters.
     fn new(
         target: &str,
         _credentials: Option<Box<dyn Credentials>>,
@@ -228,6 +203,43 @@ impl PersistentChannel {
             runtime,
         }
     }
+
+    /// Returns the current state of the channel. If there is no underlying active channel,
+    /// returns Idle. If `connect` is true, will create a new active channel iff none exists.
+    fn state(&self, connect: bool) -> ConnectivityState {
+        // Done this away to avoid potentially locking twice.
+        let active_channel = if connect {
+            self.get_active_channel()
+        } else {
+            match self.active_channel.lock().unwrap().clone() {
+                Some(x) => x,
+                None => {
+                    return ConnectivityState::Idle;
+                }
+            }
+        };
+
+        active_channel
+            .connectivity_state
+            .cur()
+            .unwrap_or(ConnectivityState::Idle)
+    }
+
+    /// Gets the underlying active channel. If there is no current connection, it will create one.
+    /// This cannot fail and will always return a valid active channel.
+    fn get_active_channel(&self) -> Arc<ActiveChannel> {
+        let mut active_channel = self.active_channel.lock().unwrap();
+
+        if active_channel.is_none() {
+            *active_channel = Some(ActiveChannel::new(
+                self.target.clone(),
+                &self.options,
+                self.runtime.clone(),
+            ));
+        }
+
+        active_channel.clone().unwrap() // We have ensured this is not None.
+    }
 }
 
 struct ActiveChannel {