use std::time::Duration;
use super::{
CancellationToken, EventCapability, InvocationContext, ModuleEventDependencyHandle,
NativeAppRuntime, NativeEndpointBinding, NativeEventHandle, NativeRequestHandle,
NativeStreamEndpointBinding, NativeStreamHandle, Rc, RefCell, StreamCapability, Weak,
};
pub trait RequestCapability: 'static {
type Request: 'static;
type Response: 'static;
type DomainError: 'static;
const ID: &'static str;
const DESCRIPTOR_VERSION: &'static str;
}
pub type RequestId = u64;
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RuntimeFailure {
Unavailable { capability: &'static str },
UnknownOperation {
capability: &'static str,
operation: String,
},
AmbiguousBinding {
capability: &'static str,
providers: usize,
},
ProtocolViolation { capability: &'static str },
MissingModuleFactory {
instance: String,
package_id: String,
},
UnavailableExecutionClass {
instance_key: String,
execution_class: String,
},
InvalidResolvedPlan { detail: String },
AdmissionClosed,
ResourceExhausted {
capability: &'static str,
operation: String,
},
DeadlineExceeded { request_id: RequestId },
Cancelled { request_id: RequestId },
Internal { detail: String },
ModuleFailure { detail: String },
ModuleRestartExhausted { instance: String, attempts: usize },
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ModuleLifecyclePhase {
Prepare,
Activate,
Ready,
Deactivate,
}
#[derive(Clone, Debug)]
pub struct ModuleDependency {
pub(super) capability_id: String,
pub(super) provider_instance: String,
pub(super) provider_order: usize,
pub(super) handle: Option<ModuleDependencyHandle>,
pub(super) stream_handle: Option<ModuleStreamDependencyHandle>,
pub(super) event_handle: Option<ModuleEventDependencyHandle>,
}
impl ModuleDependency {
pub(super) fn new(
capability_id: impl Into<String>,
provider_instance: impl Into<String>,
provider_order: usize,
handle: Option<ModuleDependencyHandle>,
stream_handle: Option<ModuleStreamDependencyHandle>,
event_handle: Option<ModuleEventDependencyHandle>,
) -> Self {
Self {
capability_id: capability_id.into(),
provider_instance: provider_instance.into(),
provider_order,
handle,
stream_handle,
event_handle,
}
}
pub fn capability_id(&self) -> &str {
&self.capability_id
}
pub fn provider_instance(&self) -> &str {
&self.provider_instance
}
pub const fn provider_order(&self) -> usize {
self.provider_order
}
pub fn handle(&self) -> Option<ModuleDependencyHandle> {
self.handle.clone()
}
pub fn stream_handle(&self) -> Option<ModuleStreamDependencyHandle> {
self.stream_handle.clone()
}
pub fn event_handle(&self) -> Option<ModuleEventDependencyHandle> {
self.event_handle.clone()
}
}
#[derive(Clone, Debug)]
pub struct ModuleDependencyHandle {
pub(super) binding: NativeEndpointBinding,
pub(super) caller_instance: String,
pub(super) runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
}
#[derive(Clone, Debug)]
pub struct ModuleStreamDependencyHandle {
pub(super) binding: NativeStreamEndpointBinding,
pub(super) caller_instance: String,
pub(super) runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
}
impl ModuleStreamDependencyHandle {
pub fn capability_id(&self) -> &'static str {
self.binding.state.capability_id
}
pub fn descriptor_version(&self) -> &'static str {
self.binding.state.descriptor_version
}
pub fn operations(&self) -> &'static [&'static str] {
self.binding.state.operations
}
pub fn typed<C: StreamCapability>(&self) -> Result<NativeStreamHandle<C>, RuntimeFailure> {
if self.capability_id() != C::ID || self.descriptor_version() != C::DESCRIPTOR_VERSION {
return Err(RuntimeFailure::ProtocolViolation { capability: C::ID });
}
let runtime = self
.runtime
.borrow()
.upgrade()
.ok_or(RuntimeFailure::AdmissionClosed)?;
Ok(NativeStreamHandle::from_endpoints(
std::slice::from_ref(&self.binding),
runtime,
&self.caller_instance,
true,
))
}
}
impl ModuleDependencyHandle {
pub fn capability_id(&self) -> &'static str {
self.binding.state.capability_id
}
pub fn descriptor_version(&self) -> &'static str {
self.binding.state.descriptor_version
}
pub fn operations(&self) -> &'static [&'static str] {
self.binding.state.operations
}
pub fn typed<C: RequestCapability>(&self) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
if self.capability_id() != C::ID || self.descriptor_version() != C::DESCRIPTOR_VERSION {
return Err(RuntimeFailure::ProtocolViolation { capability: C::ID });
}
let runtime = self
.runtime
.borrow()
.upgrade()
.ok_or(RuntimeFailure::AdmissionClosed)?;
Ok(NativeRequestHandle::from_endpoints(
std::slice::from_ref(&self.binding),
runtime,
&self.caller_instance,
true,
))
}
}
#[derive(Clone, Debug, Default)]
pub struct ModuleDependencies {
pub(super) bindings: Vec<ModuleDependency>,
pub(super) caller_instance: String,
pub(super) runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
}
impl ModuleDependencies {
pub(super) fn new(
caller_instance: impl Into<String>,
runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
) -> Self {
Self {
bindings: Vec::new(),
caller_instance: caller_instance.into(),
runtime,
}
}
pub fn bindings(&self) -> &[ModuleDependency] {
&self.bindings
}
pub fn len(&self) -> usize {
self.bindings.len()
}
pub fn is_empty(&self) -> bool {
self.bindings.is_empty()
}
pub fn invocation_context(
&self,
deadline: Option<Duration>,
cancellation: CancellationToken,
) -> Result<InvocationContext, RuntimeFailure> {
let runtime = self
.runtime
.borrow()
.upgrade()
.ok_or(RuntimeFailure::AdmissionClosed)?;
let request_id = runtime.request_ids.get();
runtime.request_ids.set(request_id.saturating_add(1));
Ok(InvocationContext::new(request_id, deadline, cancellation)
.with_caller_instance(self.caller_instance.clone()))
}
pub fn invocation_context_after(
&self,
timeout: Duration,
cancellation: CancellationToken,
) -> Result<InvocationContext, RuntimeFailure> {
let runtime = self
.runtime
.borrow()
.upgrade()
.ok_or(RuntimeFailure::AdmissionClosed)?;
let deadline = (runtime.driver.now)().saturating_add(timeout);
drop(runtime);
self.invocation_context(Some(deadline), cancellation)
}
pub fn one<C: RequestCapability>(&self) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
let handles: Vec<_> = self
.bindings
.iter()
.filter(|binding| binding.capability_id() == C::ID)
.filter_map(ModuleDependency::handle)
.collect();
match handles.as_slice() {
[handle] => handle.typed::<C>(),
[] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
handles => Err(RuntimeFailure::AmbiguousBinding {
capability: C::ID,
providers: handles.len(),
}),
}
}
pub fn optional<C: RequestCapability>(
&self,
) -> Result<Option<NativeRequestHandle<C>>, RuntimeFailure> {
match self
.bindings
.iter()
.filter(|binding| binding.capability_id() == C::ID)
.filter_map(ModuleDependency::handle)
.collect::<Vec<_>>()
.as_slice()
{
[] => Ok(None),
[handle] => handle.typed::<C>().map(Some),
handles => Err(RuntimeFailure::AmbiguousBinding {
capability: C::ID,
providers: handles.len(),
}),
}
}
pub fn many<C: RequestCapability>(
&self,
) -> Result<Vec<NativeRequestHandle<C>>, RuntimeFailure> {
self.bindings
.iter()
.filter(|binding| binding.capability_id() == C::ID)
.filter_map(ModuleDependency::handle)
.map(|handle| handle.typed::<C>())
.collect()
}
pub fn one_stream<C: StreamCapability>(&self) -> Result<NativeStreamHandle<C>, RuntimeFailure> {
let handles: Vec<_> = self
.bindings
.iter()
.filter(|binding| binding.capability_id() == C::ID)
.filter_map(ModuleDependency::stream_handle)
.collect();
match handles.as_slice() {
[handle] => handle.typed::<C>(),
[] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
handles => Err(RuntimeFailure::AmbiguousBinding {
capability: C::ID,
providers: handles.len(),
}),
}
}
pub fn optional_stream<C: StreamCapability>(
&self,
) -> Result<Option<NativeStreamHandle<C>>, RuntimeFailure> {
match self
.bindings
.iter()
.filter(|binding| binding.capability_id() == C::ID)
.filter_map(ModuleDependency::stream_handle)
.collect::<Vec<_>>()
.as_slice()
{
[] => Ok(None),
[handle] => handle.typed::<C>().map(Some),
handles => Err(RuntimeFailure::AmbiguousBinding {
capability: C::ID,
providers: handles.len(),
}),
}
}
pub fn many_stream<C: StreamCapability>(
&self,
) -> Result<Vec<NativeStreamHandle<C>>, RuntimeFailure> {
self.bindings
.iter()
.filter(|binding| binding.capability_id() == C::ID)
.filter_map(ModuleDependency::stream_handle)
.map(|handle| handle.typed::<C>())
.collect()
}
pub fn many_event<C: EventCapability>(&self) -> Result<NativeEventHandle<C>, RuntimeFailure> {
let handles: Vec<_> = self
.bindings
.iter()
.filter(|binding| binding.capability_id() == C::ID)
.filter_map(ModuleDependency::event_handle)
.collect();
if handles.iter().any(|handle| {
handle.capability_id() != C::ID || handle.descriptor_version() != C::DESCRIPTOR_VERSION
}) {
return Err(RuntimeFailure::ProtocolViolation { capability: C::ID });
}
let runtime = self
.runtime
.borrow()
.upgrade()
.ok_or(RuntimeFailure::AdmissionClosed)?;
let endpoints = handles
.iter()
.map(|handle| handle.binding.clone())
.collect::<Vec<_>>();
Ok(NativeEventHandle::from_endpoints(
&endpoints,
runtime,
&self.caller_instance,
true,
))
}
pub fn one_event<C: EventCapability>(&self) -> Result<NativeEventHandle<C>, RuntimeFailure> {
match self
.bindings
.iter()
.filter(|binding| binding.capability_id() == C::ID)
.filter_map(ModuleDependency::event_handle)
.collect::<Vec<_>>()
.as_slice()
{
[handle] => handle.typed::<C>(),
[] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
handles => Err(RuntimeFailure::AmbiguousBinding {
capability: C::ID,
providers: handles.len(),
}),
}
}
pub fn optional_event<C: EventCapability>(
&self,
) -> Result<Option<NativeEventHandle<C>>, RuntimeFailure> {
match self
.bindings
.iter()
.filter(|binding| binding.capability_id() == C::ID)
.filter_map(ModuleDependency::event_handle)
.collect::<Vec<_>>()
.as_slice()
{
[] => Ok(None),
[handle] => handle.typed::<C>().map(Some),
handles => Err(RuntimeFailure::AmbiguousBinding {
capability: C::ID,
providers: handles.len(),
}),
}
}
}