refactor(managed): remove legacy runtime #52
No files matched your search
@@ -7,8 +7,8 @@ use thiserror::Error;
|
|||||||
use tokio::runtime::{Builder, Runtime};
|
use tokio::runtime::{Builder, Runtime};
|
||||||
|
|
||||||
use super::{
|
use super::{
|
||||||
ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedConfigError, ManagedOutcome,
|
ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask, ManagedConfigError,
|
||||||
ManagedTargetConfig,
|
ManagedEventSubscription, ManagedOutcome, ManagedSubscribeError, ManagedTargetConfig,
|
||||||
};
|
};
|
||||||
|
|
||||||
const BLOCKING_RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(100);
|
const BLOCKING_RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(100);
|
||||||
@@ -56,6 +56,10 @@ pub enum BlockingManagedStartError {
|
|||||||
#[error(transparent)]
|
#[error(transparent)]
|
||||||
Config(#[from] ManagedConfigError),
|
Config(#[from] ManagedConfigError),
|
||||||
|
|
||||||
|
/// The managed event subscription could not be created before startup.
|
||||||
|
#[error(transparent)]
|
||||||
|
Subscribe(#[from] ManagedSubscribeError),
|
||||||
|
|
||||||
/// The dedicated Tokio runtime could not be created.
|
/// The dedicated Tokio runtime could not be created.
|
||||||
#[error("failed to create blocking managed Tokio runtime: {source}")]
|
#[error("failed to create blocking managed Tokio runtime: {source}")]
|
||||||
Runtime {
|
Runtime {
|
||||||
@@ -98,6 +102,29 @@ impl BlockingManagedClient {
|
|||||||
targets: Vec<ManagedTargetConfig>,
|
targets: Vec<ManagedTargetConfig>,
|
||||||
) -> Result<Self, BlockingManagedStartError> {
|
) -> Result<Self, BlockingManagedStartError> {
|
||||||
let (task, handle) = ManagedClient::task(config, targets)?;
|
let (task, handle) = ManagedClient::task(config, targets)?;
|
||||||
|
Self::start_task(task, handle)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Validate configuration, subscribe to presentation events, and start a
|
||||||
|
/// dedicated managed worker.
|
||||||
|
///
|
||||||
|
/// The returned subscription exists before the managed task receives its
|
||||||
|
/// first poll, so it cannot miss startup events because of subscription
|
||||||
|
/// ordering.
|
||||||
|
pub fn start_with_subscription(
|
||||||
|
config: ManagedClientConfig,
|
||||||
|
targets: Vec<ManagedTargetConfig>,
|
||||||
|
) -> Result<(Self, ManagedEventSubscription), BlockingManagedStartError> {
|
||||||
|
let (task, handle) = ManagedClient::task(config, targets)?;
|
||||||
|
let subscription = handle.subscribe()?;
|
||||||
|
let owner = Self::start_task(task, handle)?;
|
||||||
|
Ok((owner, subscription))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn start_task(
|
||||||
|
task: ManagedClientTask,
|
||||||
|
handle: ManagedClientHandle,
|
||||||
|
) -> Result<Self, BlockingManagedStartError> {
|
||||||
let runtime = Builder::new_current_thread()
|
let runtime = Builder::new_current_thread()
|
||||||
.enable_io()
|
.enable_io()
|
||||||
.enable_time()
|
.enable_time()
|
||||||
|
|||||||
@@ -14,7 +14,8 @@ use tokio::{runtime::Builder, time::timeout};
|
|||||||
use super::{
|
use super::{
|
||||||
blocking::{join_worker, WorkerRuntime},
|
blocking::{join_worker, WorkerRuntime},
|
||||||
BlockingManagedClient, BlockingManagedJoinError, ManagedClientConfig, ManagedCompletionPolicy,
|
BlockingManagedClient, BlockingManagedJoinError, ManagedClientConfig, ManagedCompletionPolicy,
|
||||||
ManagedEndReason, ManagedLifecycle, ManagedPacing, ManagedTargetConfig, ManagedTargetEndReason,
|
ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedPacing, ManagedTargetConfig,
|
||||||
|
ManagedTargetEndReason,
|
||||||
};
|
};
|
||||||
use crate::ClientConfig;
|
use crate::ClientConfig;
|
||||||
|
|
||||||
@@ -183,6 +184,35 @@ fn starts_without_caller_runtime_and_finishes_naturally() {
|
|||||||
server.finish();
|
server.finish();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn subscribed_start_captures_events_from_a_short_run() {
|
||||||
|
let server = EchoServer::start();
|
||||||
|
let (owner, mut events) = BlockingManagedClient::start_with_subscription(
|
||||||
|
config(ManagedCompletionPolicy::FinishWhenQuiescent),
|
||||||
|
vec![target(&server)],
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
server.wait_probe();
|
||||||
|
let outcome = owner.join().unwrap();
|
||||||
|
assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete);
|
||||||
|
|
||||||
|
let mut saw_started = false;
|
||||||
|
let mut saw_finished = false;
|
||||||
|
let mut saw_completed = false;
|
||||||
|
while let Ok(event) = events.try_recv() {
|
||||||
|
match event {
|
||||||
|
ManagedEvent::Started => saw_started = true,
|
||||||
|
ManagedEvent::TargetFinished { .. } => saw_finished = true,
|
||||||
|
ManagedEvent::Completed { .. } => saw_completed = true,
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assert!(saw_started && saw_finished && saw_completed);
|
||||||
|
server.wait_close();
|
||||||
|
server.finish();
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn external_handle_stop_ends_with_graceful_close() {
|
fn external_handle_stop_ends_with_graceful_close() {
|
||||||
let server = EchoServer::start();
|
let server = EchoServer::start();
|
||||||
|
|||||||
@@ -173,6 +173,9 @@ pub enum ManagedEvent {
|
|||||||
/// Receiving half of the lossy managed event stream.
|
/// Receiving half of the lossy managed event stream.
|
||||||
pub type ManagedEventSubscription = tokio::sync::broadcast::Receiver<ManagedEvent>;
|
pub type ManagedEventSubscription = tokio::sync::broadcast::Receiver<ManagedEvent>;
|
||||||
|
|
||||||
|
/// Immediate result of attempting to receive a managed presentation event.
|
||||||
|
pub type ManagedEventTryRecvError = tokio::sync::broadcast::error::TryRecvError;
|
||||||
|
|
||||||
/// Final authoritative outcome returned by the task.
|
/// Final authoritative outcome returned by the task.
|
||||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||||
pub struct ManagedOutcome {
|
pub struct ManagedOutcome {
|
||||||
|
|||||||
Reference in new issue
Block a user