#[path = "plugin_activation.rs"]
mod plugin_activation;
use std::collections::BTreeMap;
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use super::cases::Text;
use super::{Compute, Failed, Lanes, Operations, Owner, Parent, ParentOwner, Request, Response, Slow, request, runtime};
use harmos_runtime::operations::{OperationDeclaration, PluginRegistration, WireDeclaration};
use harmos_runtime::plugin::{
PluginArtifact, PluginBackend, PluginCaller, PluginContributions, PluginDefinition, PluginDefinitions, PluginEndpoint, PluginLifecycle,
PluginManager, PluginScope, PluginTransition, Requirement, Router,
};
use harmos_runtime::{Error, Message, Result, operations::OperationManager};
type Backends = PluginBackend<Resources, Resources, Resources>;
#[harmos_runtime::operation(resources = [SelectedOwner], id = "fixture.selected_provider",
response = String,
visibility = Shared,
policy = super::cases::EchoDeclaration::POLICY
)]
struct SelectedDeclaration {
value: String,
}
struct SelectedOwner(&'static str);
struct SelectedHost {
operations: OperationManager<Lanes>,
scopes: BTreeMap<String, PluginScope>,
callers: BTreeMap<String, PluginCaller>,
}
#[harmos_runtime::operation(resources = [Owner, harmos_runtime::operations::HostContext], codec = manual, id = "test.shared",
response = Response,
visibility = Shared,
policy = Compute::POLICY,
key = |request: &Shared| Compute::key(&request.0),
)]
#[derive(Clone)]
struct Shared(Request);
#[harmos_runtime::operation(resources = [CleanupOwner], id = "fixture.callback_cleanup",
response = String,
policy = super::cases::EchoDeclaration::POLICY
)]
struct CleanupDeclaration {
value: String,
}
#[derive(Default)]
struct CleanupOwner {
started: tokio::sync::Notify,
finalized: AtomicUsize,
attempted: AtomicUsize,
}
#[derive(Default)]
struct Resources {
drop_receipt: Arc<std::sync::atomic::AtomicBool>,
endpoint: Option<PluginEndpoint>,
registration: Option<PluginRegistration>,
}
struct Host {
operations: OperationManager<Lanes>,
owner: Arc<Owner>,
scope: Option<PluginScope>,
fail_start: bool,
skip_operations: bool,
drop_before_commit: bool,
unregister_before_commit: bool,
drop_receipt: Arc<std::sync::atomic::AtomicBool>,
callers: BTreeMap<String, PluginCaller>,
call_during_setup: bool,
cleanup: Option<Arc<CleanupOwner>>,
}
struct Credit {
runtime: harmos_runtime::Runtime<super::Bank>,
manager: PluginManager<Resources, Resources, Resources>,
host: Host,
cleanup: Arc<CleanupOwner>,
target: harmos_runtime::operations::PluginCallback,
echo: harmos_runtime::operations::PluginCallback,
}
struct Running {
runtime: harmos_runtime::Runtime<super::Bank>,
manager: PluginManager<Resources, Resources, Resources>,
host: Host,
gate: Arc<tokio::sync::Notify>,
invocation: harmos_runtime::operations::Running<Compute>,
}
super::in_process!(Shared);
impl harmos_runtime::MemorySize for Shared {
fn memory_size(&self) -> usize {
self.0.memory_size()
}
}
impl From<&str> for SelectedDeclaration {
fn from(value: &str) -> Self {
Self { value: value.to_owned() }
}
}
impl harmos_runtime::Operation for SelectedDeclaration {
type Error = Failed;
async fn run(
self, ctx: harmos_runtime::Context<'_, Self>,
) -> std::result::Result<harmos_runtime::operations::Response<String>, Failed> {
let resources = ctx.resource::<SelectedOwner>();
let result: ::core::result::Result<String, Self::Error> = async move { Ok(format!("{}:{}", resources.0, self.value)) }.await;
result.map(::core::convert::Into::into)
}
}
impl PluginLifecycle for SelectedHost {
type Guest = Resources;
type Sidecar = Resources;
type Embedded = Resources;
fn prepare(&mut self, _: &PluginDefinition) -> Result<Backends> {
Ok(PluginBackend::Embedded(Resources::default()))
}
async fn fence(&mut self, _: PluginTransition, _: &[PluginDefinition]) -> Result<()> {
Ok(())
}
async fn start(
&mut self, definition: &PluginDefinition, router: &Router, _: &harmos_runtime::plugin::PluginLifecycleAuthority<'_>,
backend: &mut Backends,
) -> Result<()> {
let PluginBackend::Embedded(resources) = backend else { unreachable!() };
resources.endpoint = Some(router.enter(&definition.id).unwrap());
let endpoint = resources.endpoint.as_ref().unwrap();
let scope = endpoint.scope().unwrap();
self.scopes.insert(definition.id.clone(), scope.clone());
self.callers.insert(definition.id.clone(), endpoint.caller());
let owner = match definition.id.as_str() {
"first" => SelectedOwner("first"),
"second" => SelectedOwner("second"),
"client" => return Ok(()),
_ => unreachable!("closed fixture owner set"),
};
resources.registration = Some(self.operations.stage_plugin(scope, |bindings| {
bindings.register(harmos_runtime::RegistrationExt::lane(
harmos_runtime::RegistrationBatch::new({
let mut dependencies = harmos_runtime::Dependencies::new();
dependencies.insert(Arc::new(owner)).unwrap();
dependencies
})
.operation::<SelectedDeclaration>(),
Lanes::Open,
))
})?);
Ok(())
}
async fn stop(_: &PluginDefinition, _: &harmos_runtime::plugin::PluginLifecycleAuthority<'_>, backend: &mut Backends) -> Result<()> {
let PluginBackend::Embedded(resources) = backend else { unreachable!() };
if let Some(registration) = &mut resources.registration {
registration.shutdown().await?;
}
resources.registration.take();
resources.endpoint.take();
Ok(())
}
}
impl harmos_runtime::Operation for Shared {
type Error = Failed;
async fn run(self, ctx: harmos_runtime::Context<'_, Self>) -> std::result::Result<harmos_runtime::Response<Response>, Failed> {
let owner = ctx.resource::<Owner>();
let host = ctx.resource::<harmos_runtime::operations::HostContext>();
let context = host.borrow();
if context.attempt() == 0 {
owner.prepared.fetch_add(1, Ordering::SeqCst);
}
owner.compute(&self.0, context).await.map(Into::into)
}
}
struct CleanupGuard(Arc<CleanupOwner>);
impl harmos_runtime::Operation for CleanupDeclaration {
type Error = Failed;
async fn run(self, ctx: harmos_runtime::Context<'_, Self>) -> std::result::Result<harmos_runtime::Response<String>, Self::Error> {
let resources = ctx.resource::<CleanupOwner>();
let _guard = CleanupGuard(resources.clone());
resources.attempted.fetch_add(1, Ordering::SeqCst);
resources.started.notify_one();
if matches!(self.value.as_str(), "block" | "expire") {
std::future::pending::<()>().await;
}
Ok(self.value.into())
}
}
impl Drop for CleanupGuard {
fn drop(&mut self) {
self.0.finalized.fetch_add(1, Ordering::SeqCst);
}
}
impl PluginLifecycle for Host {
type Guest = Resources;
type Sidecar = Resources;
type Embedded = Resources;
fn prepare(&mut self, _definition: &PluginDefinition) -> Result<Backends> {
Ok(PluginBackend::Embedded(Resources {
drop_receipt: self.drop_receipt.clone(),
..Resources::default()
}))
}
async fn fence(&mut self, _transition: PluginTransition, _affected: &[PluginDefinition]) -> Result<()> {
Ok(())
}
async fn start(
&mut self, definition: &PluginDefinition, router: &Router, _authority: &harmos_runtime::plugin::PluginLifecycleAuthority<'_>,
backend: &mut Backends,
) -> Result<()> {
let PluginBackend::Embedded(resources) = backend else { unreachable!() };
resources.endpoint = Some(router.enter(&definition.id).unwrap());
let caller = resources.endpoint.as_ref().unwrap().caller();
self.callers.insert(definition.id.clone(), caller.clone());
if definition.id != "fixture" {
if self.call_during_setup && definition.id == "client" {
assert_eq!(self.operations.caller(&caller)?.call::<Shared>(Shared(request())).await?.value, 7);
}
return Ok(());
}
let scope = resources.endpoint.as_ref().unwrap().scope().unwrap();
self.scope = Some(scope.clone());
if self.skip_operations {
return Ok(());
}
resources.registration = Some(self.operations.stage_plugin(scope, |bindings| {
bindings.register(harmos_runtime::RegistrationExt::lane(
harmos_runtime::RegistrationBatch::new({
let mut dependencies = harmos_runtime::Dependencies::new();
dependencies.insert(self.owner.clone()).unwrap();
dependencies
})
.operation::<Compute>()
.operation::<Slow>(),
Lanes::Open,
))?;
bindings.register(harmos_runtime::RegistrationExt::lane(
harmos_runtime::RegistrationBatch::new({
let mut dependencies = harmos_runtime::Dependencies::new();
dependencies.insert(self.owner.clone()).unwrap();
dependencies.insert(Arc::new(super::cases::Destination("fixture"))).unwrap();
dependencies
})
.operation::<super::cases::EchoDeclaration>()
.operation::<super::cases::PrivateEchoDeclaration>()
.operation::<super::cases::NestedDeclaration>()
.operation::<super::cases::DetachedDeclaration>(),
Lanes::Open,
))?;
if let Some(cleanup) = &self.cleanup {
bindings.register(harmos_runtime::RegistrationExt::lane(
harmos_runtime::RegistrationBatch::new({
let mut dependencies = harmos_runtime::Dependencies::new();
dependencies.insert(cleanup.clone()).unwrap();
dependencies
})
.operation::<CleanupDeclaration>(),
Lanes::Open,
))?;
}
bindings.register(harmos_runtime::RegistrationExt::lane(
harmos_runtime::RegistrationBatch::new({
let mut dependencies = harmos_runtime::Dependencies::new();
dependencies.insert(self.owner.clone()).unwrap();
dependencies
})
.operation::<Shared>(),
Lanes::Serial,
))?;
bindings.register(harmos_runtime::RegistrationExt::lane(
harmos_runtime::RegistrationBatch::new({
let mut dependencies = harmos_runtime::Dependencies::new();
dependencies.insert(Arc::new(ParentOwner)).unwrap();
dependencies
})
.operation::<Parent>(),
Lanes::Serial,
))
})?);
assert!(self.operations.definitions().is_empty());
assert!(matches!(self.operations.definition(Compute::ID), Err(Error::NotReady { .. })));
assert!(matches!(self.operations.submit::<Compute>(request()), Err(Error::NotReady { .. })));
if self.unregister_before_commit {
self.operations.unregister(&harmos_runtime::QualifiedId::qualify("fixture", Compute::ID).unwrap()).await?;
}
if self.drop_before_commit {
resources.registration.take();
}
if self.fail_start {
return Err(Error::InvalidInput {
field: "test activation",
reason: "scripted initialization failure",
});
}
Ok(())
}
async fn stop(
_definition: &PluginDefinition, _authority: &harmos_runtime::plugin::PluginLifecycleAuthority<'_>, backend: &mut Backends,
) -> Result<()> {
let PluginBackend::Embedded(resources) = backend else { unreachable!() };
if !resources.drop_receipt.load(std::sync::atomic::Ordering::SeqCst)
&& let Some(registration) = resources.registration.as_mut()
{
registration.shutdown().await?;
}
resources.registration.take();
resources.endpoint.take();
Ok(())
}
}
impl Credit {
async fn start() -> Self {
let runtime = super::runtime_with_preparation(NonZeroUsize::new(128).unwrap()).await;
let mut manager = PluginManager::new(candidate_with_cleanup()).unwrap();
let mut host = host(runtime.operations.clone());
let cleanup = Arc::new(CleanupOwner::default());
host.cleanup = Some(cleanup.clone());
manager.initialize(&mut host).await.unwrap();
let scope = host.scope.as_ref().unwrap();
let target = runtime.operations.capture_callback(scope, &WireDeclaration::of::<CleanupDeclaration>().contract().unwrap()).unwrap();
let echo = runtime
.operations
.capture_callback(scope, &WireDeclaration::of::<super::cases::EchoDeclaration>().contract().unwrap())
.unwrap();
Self {
runtime,
manager,
host,
cleanup,
target,
echo,
}
}
fn assert_held(&self) {
let encoded = AtomicBool::new(false);
assert!(matches!(
self.echo.submit_with(NonZeroUsize::new(64).unwrap(), |_| {
encoded.store(true, Ordering::SeqCst);
Ok(Text::from("busy").encode())
}),
Err(Error::ResourceExhausted { resource: "operation preparation bytes" })
));
assert!(!encoded.load(Ordering::SeqCst));
}
async fn settle(mut self) {
self.runtime
.operations
.unregister(&harmos_runtime::QualifiedId::qualify("fixture", CleanupDeclaration::ID).unwrap())
.await
.unwrap();
assert_eq!(self.cleanup.finalized.load(Ordering::SeqCst), 1);
let bytes = self.echo.submit_with(NonZeroUsize::new(64).unwrap(), |_| Ok(Text::from("settled").encode())).unwrap().await.unwrap();
assert_eq!(String::decode(&bytes).unwrap(), "settled:1");
self.manager.shutdown(&mut self.host).await.unwrap();
self.runtime.stop().await.unwrap();
}
}
impl Running {
async fn start() -> Self {
let runtime = runtime().await;
let mut manager = PluginManager::new(candidate()).unwrap();
let mut host = host(runtime.operations.clone());
host.drop_receipt.store(true, Ordering::SeqCst);
manager.initialize(&mut host).await.unwrap();
let opened = Arc::new(tokio::sync::Notify::new());
let gate = Arc::new(tokio::sync::Notify::new());
let invocation = runtime
.operation(super::Request {
gate: Some((opened.clone(), gate.clone())),
..request()
})
.start()
.await
.unwrap();
opened.notified().await;
Self { runtime, manager, host, gate, invocation }
}
}
fn selected_candidate(provider: &str, first: bool) -> PluginDefinitions {
let contract = WireDeclaration::of::<SelectedDeclaration>().contract().unwrap();
let owners = if first { vec!["first", "second", "client"] } else { vec!["second", "client"] };
let definitions = owners.into_iter().map(|id| PluginDefinition {
id: id.into(),
version: "1.0.0".parse().unwrap(),
artifact: PluginArtifact::Embedded { revision: "fixture".into() },
requires: vec![],
configuration: b"null".to_vec(),
contributions: if id == "client" {
PluginContributions {
required_contracts: Some(vec![harmos_domain::inventory::contracts::Required {
contract: contract.clone(),
grants: Vec::new(),
}]),
..PluginContributions::default()
}
} else {
PluginContributions {
operations: vec![contract.operation().clone()],
wire_contracts: vec![contract.clone()],
required_contracts: Some(vec![]),
..PluginContributions::default()
}
},
});
PluginDefinitions::new_with_providers(
definitions,
[],
[harmos_runtime::plugin::OperationProvider {
consumer: "client".into(),
operation: harmos_runtime::Identifier::new(SelectedDeclaration::ID).unwrap(),
provider: provider.into(),
}],
)
.unwrap()
}
fn candidate() -> PluginDefinitions {
let mut operations = harmos_runtime::operations::Declarations::of::<Operations>().unwrap();
operations.extend(harmos_runtime::operations::Declarations::of::<super::cases::Operations>().unwrap());
operations.push(harmos_runtime::operations::OperationDefinition::of::<Shared>().unwrap());
operations.push(harmos_runtime::operations::OperationDefinition::of::<Parent>().unwrap());
PluginDefinitions::new(
[PluginDefinition {
id: "fixture".into(),
version: "1.0.0".parse().unwrap(),
artifact: PluginArtifact::Embedded { revision: "fixture".into() },
requires: Vec::new(),
configuration: b"null".to_vec(),
contributions: PluginContributions {
operations,
..PluginContributions::default()
},
}],
[],
)
.unwrap()
}
fn with_clients(client_requires_provider: bool) -> PluginDefinitions {
let provider = candidate().definitions().next().unwrap().clone();
let clients = [
("client", if client_requires_provider { vec!["fixture"] } else { Vec::new() }),
("transitive", vec!["client"]),
("unrelated", Vec::new()),
]
.into_iter()
.map(|(id, requires)| PluginDefinition {
id: id.into(),
version: "1.0.0".parse().unwrap(),
artifact: PluginArtifact::Embedded { revision: "fixture".into() },
requires: requires
.into_iter()
.map(|id| Requirement {
id: id.into(),
version: "*".parse().unwrap(),
})
.collect(),
configuration: b"null".to_vec(),
contributions: PluginContributions::default(),
});
PluginDefinitions::new(std::iter::once(provider).chain(clients), []).unwrap()
}
fn host(operations: OperationManager<Lanes>) -> Host {
Host {
operations,
owner: Arc::new(Owner::default()),
scope: None,
fail_start: false,
skip_operations: false,
drop_before_commit: false,
unregister_before_commit: false,
drop_receipt: Arc::default(),
callers: BTreeMap::new(),
call_during_setup: false,
cleanup: None,
}
}
fn candidate_with_cleanup() -> PluginDefinitions {
let mut definition = candidate().definitions().next().unwrap().clone();
definition.contributions.operations.push(harmos_runtime::operations::OperationDefinition::of::<CleanupDeclaration>().unwrap());
PluginDefinitions::new([definition], []).unwrap()
}