use super::{
BTreeMap, BTreeSet, ErasedDomainResult, ErasedValue, ExecutionClassId, InvocationContext,
LocalBoxFuture, ModuleLifecycle, NativeEventEndpoint, NativeStreamEndpoint, Rc,
ResolvedAppPlan, RuntimeFailure,
};
pub trait NativeRequestEndpoint: std::fmt::Debug {
fn capability_id(&self) -> &'static str;
fn descriptor_version(&self) -> &'static str;
fn operations(&self) -> &'static [&'static str];
fn invoke(
&self,
operation: &str,
request: ErasedValue,
context: InvocationContext,
) -> LocalBoxFuture<'static, Result<ErasedDomainResult, RuntimeFailure>>;
}
#[derive(Clone, Debug, Default)]
pub struct NativeEndpointSet {
request: Vec<Rc<dyn NativeRequestEndpoint>>,
stream: Vec<Rc<dyn NativeStreamEndpoint>>,
event: Vec<Rc<dyn NativeEventEndpoint>>,
}
impl NativeEndpointSet {
pub fn new(
request: Vec<Rc<dyn NativeRequestEndpoint>>,
stream: Vec<Rc<dyn NativeStreamEndpoint>>,
event: Vec<Rc<dyn NativeEventEndpoint>>,
) -> Self {
Self {
request,
stream,
event,
}
}
pub fn request(&self) -> &[Rc<dyn NativeRequestEndpoint>] {
&self.request
}
pub fn stream(&self) -> &[Rc<dyn NativeStreamEndpoint>] {
&self.stream
}
pub fn event(&self) -> &[Rc<dyn NativeEventEndpoint>] {
&self.event
}
}
#[derive(Debug)]
pub struct PreparedNativeModule {
pub(super) endpoints: NativeEndpointSet,
pub(super) lifecycle: Rc<dyn ModuleLifecycle>,
}
impl PreparedNativeModule {
pub fn new(
endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
lifecycle: impl ModuleLifecycle,
) -> Self {
Self {
endpoints: NativeEndpointSet::new(endpoints, Vec::new(), Vec::new()),
lifecycle: Rc::new(lifecycle),
}
}
pub fn with_lifecycle(
endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
lifecycle: Rc<dyn ModuleLifecycle>,
) -> Self {
Self {
endpoints: NativeEndpointSet::new(endpoints, Vec::new(), Vec::new()),
lifecycle,
}
}
pub fn with_endpoints(
endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
lifecycle: impl ModuleLifecycle,
) -> Self {
Self {
endpoints: NativeEndpointSet::new(endpoints, stream_endpoints, Vec::new()),
lifecycle: Rc::new(lifecycle),
}
}
pub fn with_endpoint_set_lifecycle(
endpoints: NativeEndpointSet,
lifecycle: Rc<dyn ModuleLifecycle>,
) -> Self {
Self {
endpoints,
lifecycle,
}
}
pub fn with_stream_endpoints(
stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
lifecycle: impl ModuleLifecycle,
) -> Self {
Self::with_endpoints(Vec::new(), stream_endpoints, lifecycle)
}
pub fn with_event_endpoints(
event_endpoints: Vec<Rc<dyn NativeEventEndpoint>>,
lifecycle: impl ModuleLifecycle,
) -> Self {
Self::with_all_endpoints(Vec::new(), Vec::new(), event_endpoints, lifecycle)
}
pub fn with_all_endpoints(
endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
event_endpoints: Vec<Rc<dyn NativeEventEndpoint>>,
lifecycle: impl ModuleLifecycle,
) -> Self {
Self {
endpoints: NativeEndpointSet::new(endpoints, stream_endpoints, event_endpoints),
lifecycle: Rc::new(lifecycle),
}
}
pub fn endpoints(&self) -> &[Rc<dyn NativeRequestEndpoint>] {
self.endpoints.request()
}
pub fn stream_endpoints(&self) -> &[Rc<dyn NativeStreamEndpoint>] {
self.endpoints.stream()
}
pub fn event_endpoints(&self) -> &[Rc<dyn NativeEventEndpoint>] {
self.endpoints.event()
}
pub fn lifecycle(&self) -> Rc<dyn ModuleLifecycle> {
self.lifecycle.clone()
}
pub(super) fn into_parts(self) -> (NativeEndpointSet, Rc<dyn ModuleLifecycle>) {
(self.endpoints, self.lifecycle)
}
}
#[derive(Clone, Debug)]
pub struct PreparedBinding {
pub(super) consumer_instance: String,
pub(super) provider_instance: String,
pub(super) endpoint: Rc<dyn NativeRequestEndpoint>,
}
#[derive(Clone, Debug)]
pub struct PreparedStreamBinding {
pub(super) consumer_instance: String,
pub(super) provider_instance: String,
pub(super) endpoint: Rc<dyn NativeStreamEndpoint>,
}
#[derive(Clone, Debug)]
pub struct PreparedEventBinding {
pub(super) consumer_instance: String,
pub(super) provider_instance: String,
pub(super) endpoint: Rc<dyn NativeEventEndpoint>,
}
impl PreparedEventBinding {
pub fn new(
consumer_instance: impl Into<String>,
provider_instance: impl Into<String>,
endpoint: Rc<dyn NativeEventEndpoint>,
) -> Self {
Self {
consumer_instance: consumer_instance.into(),
provider_instance: provider_instance.into(),
endpoint,
}
}
pub fn consumer_instance(&self) -> &str {
&self.consumer_instance
}
pub fn provider_instance(&self) -> &str {
&self.provider_instance
}
pub fn endpoint(&self) -> Rc<dyn NativeEventEndpoint> {
self.endpoint.clone()
}
pub(super) fn same_identity(&self, other: &Self) -> bool {
self.consumer_instance == other.consumer_instance
&& self.provider_instance == other.provider_instance
&& self.endpoint.capability_id() == other.endpoint.capability_id()
}
}
impl PreparedStreamBinding {
pub fn new(
consumer_instance: impl Into<String>,
provider_instance: impl Into<String>,
endpoint: Rc<dyn NativeStreamEndpoint>,
) -> Self {
Self {
consumer_instance: consumer_instance.into(),
provider_instance: provider_instance.into(),
endpoint,
}
}
pub fn consumer_instance(&self) -> &str {
&self.consumer_instance
}
pub fn provider_instance(&self) -> &str {
&self.provider_instance
}
pub fn endpoint(&self) -> Rc<dyn NativeStreamEndpoint> {
self.endpoint.clone()
}
pub(super) fn same_identity(&self, other: &Self) -> bool {
self.consumer_instance == other.consumer_instance
&& self.provider_instance == other.provider_instance
&& self.endpoint.capability_id() == other.endpoint.capability_id()
}
}
impl PreparedBinding {
pub fn new(
consumer_instance: impl Into<String>,
provider_instance: impl Into<String>,
endpoint: Rc<dyn NativeRequestEndpoint>,
) -> Self {
Self {
consumer_instance: consumer_instance.into(),
provider_instance: provider_instance.into(),
endpoint,
}
}
pub fn consumer_instance(&self) -> &str {
&self.consumer_instance
}
pub fn provider_instance(&self) -> &str {
&self.provider_instance
}
pub fn endpoint(&self) -> Rc<dyn NativeRequestEndpoint> {
self.endpoint.clone()
}
pub(super) fn same_identity(&self, other: &Self) -> bool {
self.consumer_instance == other.consumer_instance
&& self.provider_instance == other.provider_instance
&& self.endpoint.capability_id() == other.endpoint.capability_id()
}
}
#[derive(Debug)]
pub struct PreparedNativeApp {
pub(super) bindings: Vec<PreparedBinding>,
pub(super) stream_bindings: Vec<PreparedStreamBinding>,
pub(super) event_bindings: Vec<PreparedEventBinding>,
pub(super) generations: BTreeMap<String, PreparedNativeModule>,
}
impl PreparedNativeApp {
pub fn new(
bindings: Vec<PreparedBinding>,
generations: BTreeMap<String, PreparedNativeModule>,
) -> Self {
Self {
bindings,
stream_bindings: Vec::new(),
event_bindings: Vec::new(),
generations,
}
}
pub fn empty() -> Self {
Self::new(Vec::new(), BTreeMap::new())
}
#[must_use]
pub fn with_stream_bindings(mut self, stream_bindings: Vec<PreparedStreamBinding>) -> Self {
self.stream_bindings = stream_bindings;
self
}
#[must_use]
pub fn with_event_bindings(mut self, event_bindings: Vec<PreparedEventBinding>) -> Self {
self.event_bindings = event_bindings;
self
}
pub(super) fn merge(&mut self, other: Self) -> Result<(), RuntimeFailure> {
for binding in other.bindings {
if self
.bindings
.iter()
.any(|existing| existing.same_identity(&binding))
{
return Err(RuntimeFailure::InvalidResolvedPlan {
detail: format!(
"multiple Execution Adapters prepared binding `{}:{}:{}`",
binding.consumer_instance,
binding.endpoint.capability_id(),
binding.provider_instance
),
});
}
self.bindings.push(binding);
}
for binding in other.stream_bindings {
if self
.stream_bindings
.iter()
.any(|existing| existing.same_identity(&binding))
{
return Err(RuntimeFailure::InvalidResolvedPlan {
detail: format!(
"multiple Execution Adapters prepared stream binding `{}:{}:{}`",
binding.consumer_instance,
binding.endpoint.capability_id(),
binding.provider_instance
),
});
}
self.stream_bindings.push(binding);
}
for binding in other.event_bindings {
if self
.event_bindings
.iter()
.any(|existing| existing.same_identity(&binding))
{
return Err(RuntimeFailure::InvalidResolvedPlan {
detail: format!(
"multiple Execution Adapters prepared Event binding `{}:{}:{}`",
binding.consumer_instance,
binding.endpoint.capability_id(),
binding.provider_instance
),
});
}
self.event_bindings.push(binding);
}
for (instance_key, generation) in other.generations {
if self
.generations
.insert(instance_key.clone(), generation)
.is_some()
{
return Err(RuntimeFailure::InvalidResolvedPlan {
detail: format!(
"multiple Execution Adapters prepared Module Instance generation `{instance_key}`"
),
});
}
}
Ok(())
}
}
pub trait ExecutionAdapter: std::fmt::Debug + 'static {
fn execution_class(&self) -> ExecutionClassId;
fn prepare(&self, plan: &ResolvedAppPlan) -> Result<PreparedNativeApp, RuntimeFailure>;
fn recreate(
&self,
_plan: &ResolvedAppPlan,
instance_key: &str,
) -> Result<PreparedNativeModule, RuntimeFailure> {
Err(RuntimeFailure::Internal {
detail: format!("Execution Adapter cannot recreate Module Instance `{instance_key}`"),
})
}
}
pub trait NativeExecutionAdapter: std::fmt::Debug + 'static {
fn prepare(&self, plan: &ResolvedAppPlan) -> Result<PreparedNativeApp, RuntimeFailure>;
fn recreate(
&self,
_plan: &ResolvedAppPlan,
instance_key: &str,
) -> Result<PreparedNativeModule, RuntimeFailure> {
Err(RuntimeFailure::Internal {
detail: format!("Execution Adapter cannot recreate Module Instance `{instance_key}`"),
})
}
}
impl<T: NativeExecutionAdapter> ExecutionAdapter for T {
fn execution_class(&self) -> ExecutionClassId {
ExecutionClassId::native_rust()
}
fn prepare(&self, plan: &ResolvedAppPlan) -> Result<PreparedNativeApp, RuntimeFailure> {
NativeExecutionAdapter::prepare(self, plan)
}
fn recreate(
&self,
plan: &ResolvedAppPlan,
instance_key: &str,
) -> Result<PreparedNativeModule, RuntimeFailure> {
NativeExecutionAdapter::recreate(self, plan, instance_key)
}
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct ExecutionClassSet(BTreeSet<ExecutionClassId>);
impl ExecutionClassSet {
pub fn contains(&self, execution_class: &ExecutionClassId) -> bool {
self.0.contains(execution_class)
}
pub fn iter(&self) -> impl Iterator<Item = &ExecutionClassId> {
self.0.iter()
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ExecutionAdapterCatalogError {
DuplicateExecutionClass { execution_class: String },
}
impl std::fmt::Display for ExecutionAdapterCatalogError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::DuplicateExecutionClass { execution_class } => write!(
formatter,
"multiple Execution Adapters provide class `{execution_class}`"
),
}
}
}
impl std::error::Error for ExecutionAdapterCatalogError {}
#[derive(Debug, Default)]
pub struct ExecutionAdapterCatalog {
pub(super) adapters: BTreeMap<ExecutionClassId, Rc<dyn ExecutionAdapter>>,
}
impl ExecutionAdapterCatalog {
pub fn new() -> Self {
Self::default()
}
pub fn single(adapter: impl ExecutionAdapter) -> Self {
Self::new()
.with_adapter(adapter)
.expect("a new catalog cannot contain a duplicate execution class")
}
pub fn with_adapter(
self,
adapter: impl ExecutionAdapter,
) -> Result<Self, ExecutionAdapterCatalogError> {
self.with_shared_adapter(Rc::new(adapter))
}
pub fn with_shared_adapter(
mut self,
adapter: Rc<dyn ExecutionAdapter>,
) -> Result<Self, ExecutionAdapterCatalogError> {
let execution_class = adapter.execution_class();
if self.adapters.contains_key(&execution_class) {
return Err(ExecutionAdapterCatalogError::DuplicateExecutionClass {
execution_class: execution_class.to_string(),
});
}
self.adapters.insert(execution_class, adapter);
Ok(self)
}
pub fn execution_classes(&self) -> ExecutionClassSet {
ExecutionClassSet(self.adapters.keys().cloned().collect())
}
pub(super) fn adapter(
&self,
execution_class: &ExecutionClassId,
) -> Option<Rc<dyn ExecutionAdapter>> {
self.adapters.get(execution_class).cloned()
}
pub(super) fn prepare(
&self,
plan: &ResolvedAppPlan,
) -> Result<PreparedNativeApp, RuntimeFailure> {
let mut required_classes = BTreeSet::new();
for instance in plan.module_instances() {
if !self.adapters.contains_key(instance.execution_class()) {
return Err(RuntimeFailure::UnavailableExecutionClass {
instance_key: instance.instance_key().to_owned(),
execution_class: instance.execution_class().to_string(),
});
}
required_classes.insert(instance.execution_class().clone());
}
let mut prepared = PreparedNativeApp::empty();
for execution_class in required_classes {
let adapter = self
.adapters
.get(&execution_class)
.expect("required execution classes were validated");
prepared.merge(adapter.prepare(plan)?)?;
}
Ok(prepared)
}
}