feat(grpc): add round robin load balancing policy

grpc/src/client/load_balancing/child_manager.rs

@@ -363,8 +363,8 @@ mod test {
         Child, ChildManager, ChildUpdate, ChildWorkScheduler, ResolverUpdateSharder,
     };
     use crate::client::load_balancing::test_utils::{
-        self, StubPolicy, StubPolicyFuncs, TestChannelController, TestEvent, TestSubchannel,
-        TestWorkScheduler,
+        self, StubPolicy, StubPolicyData, StubPolicyFuncs, TestChannelController, TestEvent,
+        TestSubchannel, TestWorkScheduler,
     };
     use crate::client::load_balancing::{
         ChannelController, LbPolicy, LbPolicyBuilder, LbPolicyOptions, LbState, ParsedJsonLbConfig,
@@ -444,7 +444,7 @@ mod test {
         let (tx_events, rx_events) = mpsc::unbounded_channel::<TestEvent>();
         let tcc = Box::new(TestChannelController { tx_events });
         let builder: Arc<dyn LbPolicyBuilder> = GLOBAL_LB_REGISTRY.get_policy(test_name).unwrap();
-        let endpoint_sharder = EndpointSharder { builder: builder };
+        let endpoint_sharder = EndpointSharder { builder };
         let child_manager = ChildManager::new(Box::new(endpoint_sharder), default_runtime());
         (rx_events, Box::new(child_manager), tcc)
     }
@@ -517,25 +517,29 @@ mod test {
     // Defines the functions resolver_update and subchannel_update to test
     // aggregate_states.
     fn create_verifying_funcs_for_aggregate_tests() -> StubPolicyFuncs {
+        let data = StubPolicyData::new();
         StubPolicyFuncs {
             // Closure for resolver_update. resolver_update should only receive
             // one endpoint and create one subchannel for the endpoint it
             // receives.
-            resolver_update: Some(move |update: ResolverUpdate, _, controller| {
-                assert_eq!(update.endpoints.iter().len(), 1);
-                let endpoint = update.endpoints.unwrap().pop().unwrap();
-                let subchannel = controller.new_subchannel(&endpoint.addresses[0]);
-                Ok(())
-            }),
+            resolver_update: Some(Arc::new(
+                move |data, update: ResolverUpdate, _, controller| {
+                    assert_eq!(update.endpoints.iter().len(), 1);
+                    let endpoint = update.endpoints.unwrap().pop().unwrap();
+                    let subchannel = controller.new_subchannel(&endpoint.addresses[0]);
+                    Ok(())
+                },
+            )),
             // Closure for subchannel_update. Sends a picker of the same state
             // that was passed to it.
-            subchannel_update: Some(move |updated_subchannel, state, controller| {
-                controller.update_picker(LbState {
-                    connectivity_state: state.connectivity_state,
-                    picker: Arc::new(QueuingPicker {}),
-                });
-            }),
-            ..Default::default()
+            subchannel_update: Some(Arc::new(
+                move |data, updated_subchannel, state, controller| {
+                    controller.update_picker(LbState {
+                        connectivity_state: state.connectivity_state,
+                        picker: Arc::new(QueuingPicker {}),
+                    });
+                },
+            )),
         }
     }
 

grpc/src/client/load_balancing/mod.rs

@@ -54,8 +54,10 @@ use crate::client::{
 
 pub mod child_manager;
 pub mod pick_first;
+pub mod round_robin;
 #[cfg(test)]
 pub mod test_utils;
+pub mod utils;
 
 pub(crate) mod registry;
 use super::{service_config::LbConfig, subchannel::SubchannelStateWatcher};