use std::any::type_name;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::LazyLock;
use std::sync::Mutex;
use crate::client::load_balancing::ChannelController;
use crate::client::load_balancing::DynLbConfig;
use crate::client::load_balancing::DynLbPolicy;
use crate::client::load_balancing::DynLbPolicyBuilder;
use crate::client::load_balancing::LbPolicy;
use crate::client::load_balancing::LbPolicyBuilder;
use crate::client::load_balancing::LbPolicyOptions;
use crate::client::load_balancing::ParsedJsonLbConfig;
use crate::client::load_balancing::subchannel::Subchannel;
use crate::client::load_balancing::subchannel::SubchannelState;
use crate::client::name_resolution::ResolverUpdate;
pub(crate) struct LbPolicyRegistry {
m: Arc<Mutex<HashMap<String, Arc<DynLbPolicyBuilder>>>>,
}
impl LbPolicyRegistry {
pub fn new() -> Self {
Self { m: Arc::default() }
}
pub(crate) fn add_builder<B: LbPolicyBuilder>(&self, builder: B) {
self.m
.lock()
.unwrap()
.insert(builder.name().to_string(), DynAdapter::new_arc(builder));
}
pub(crate) fn add_dyn_builder(&self, builder: Arc<DynLbPolicyBuilder>) {
self.m
.lock()
.unwrap()
.insert(builder.name().to_string(), builder);
}
pub(crate) fn get_policy(&self, name: &str) -> Option<Arc<DynLbPolicyBuilder>> {
self.m.lock().unwrap().get(name).cloned()
}
}
impl Default for LbPolicyRegistry {
fn default() -> Self {
Self::new()
}
}
pub(crate) static GLOBAL_LB_REGISTRY: LazyLock<LbPolicyRegistry> =
LazyLock::new(LbPolicyRegistry::new);
#[derive(Debug)]
struct DynAdapter<T>(T);
impl<T: LbPolicyBuilder> LbPolicyBuilder for DynAdapter<T> {
type LbPolicy = Box<DynLbPolicy>;
fn build(&self, options: LbPolicyOptions) -> Self::LbPolicy {
Box::new(DynAdapter(self.0.build(options)))
}
fn name(&self) -> &'static str {
self.0.name()
}
fn parse_config(&self, config: &ParsedJsonLbConfig) -> Result<Option<DynLbConfig>, String> {
let cfg = self.0.parse_config(config)?;
Ok(cfg.map(|c| Arc::new(c) as DynLbConfig))
}
}
impl<T: LbPolicy> LbPolicy for DynAdapter<T> {
type LbConfig = DynLbConfig;
fn resolver_update(
&mut self,
update: ResolverUpdate,
config: Option<&DynLbConfig>,
channel_controller: &mut dyn ChannelController,
) -> Result<(), String> {
let config = config.map(|c| {
c.downcast_ref::<T::LbConfig>().unwrap_or_else(|| {
panic!("LB config type should be {}", type_name::<T::LbConfig>())
})
});
self.0.resolver_update(update, config, channel_controller)
}
fn subchannel_update(
&mut self,
subchannel: Arc<dyn Subchannel>,
state: &SubchannelState,
channel_controller: &mut dyn ChannelController,
) {
self.0
.subchannel_update(subchannel, state, channel_controller);
}
fn work(&mut self, channel_controller: &mut dyn ChannelController) {
self.0.work(channel_controller);
}
fn exit_idle(&mut self, channel_controller: &mut dyn ChannelController) {
self.0.exit_idle(channel_controller);
}
}
impl<T> DynAdapter<T> {
fn new_arc(policy: T) -> Arc<Self> {
Arc::new(DynAdapter(policy))
}
}