use parking_lot::{Mutex, RwLock};
use std::fmt;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use tokio::task::JoinHandle;
#[cfg(test)]
mod tests;
use crate::effect::EffectDependencies;
use crate::hook_context::with_hook_context;
use crate::panic_handler::spawn_catch_panic;
pub type DefaultError = Box<dyn std::error::Error + Send + Sync>;
pub type DefaultFutureHandle<T> = FutureHandle<T, DefaultError>;
pub type ProgressCallback = Arc<dyn Fn(f32) + Send + Sync>;
const MAX_CONCURRENT_FUTURES_PER_COMPONENT: usize = 50;
static GLOBAL_ACTIVE_FUTURES: AtomicUsize = AtomicUsize::new(0);
const MAX_GLOBAL_CONCURRENT_FUTURES: usize = 1000;
#[derive(Debug, Default, Clone, PartialEq)]
pub enum FutureState<T, E = Box<dyn std::error::Error + Send + Sync>> {
#[default]
Idle,
Pending,
Resolved(T),
Error(E),
}
impl<T, E> FutureState<T, E> {
pub fn is_idle(&self) -> bool {
matches!(self, FutureState::Idle)
}
pub fn is_pending(&self) -> bool {
matches!(self, FutureState::Pending)
}
pub fn is_resolved(&self) -> bool {
matches!(self, FutureState::Resolved(_))
}
pub fn is_error(&self) -> bool {
matches!(self, FutureState::Error(_))
}
pub fn value(&self) -> Option<&T> {
match self {
FutureState::Resolved(v) => Some(v),
_ => None,
}
}
pub fn error(&self) -> Option<&E> {
match self {
FutureState::Error(error) => Some(error),
_ => None,
}
}
pub fn map<U, F>(self, f: F) -> FutureState<U, E>
where
F: FnOnce(T) -> U,
{
match self {
FutureState::Idle => FutureState::Idle,
FutureState::Pending => FutureState::Pending,
FutureState::Resolved(value) => FutureState::Resolved(f(value)),
FutureState::Error(error) => FutureState::Error(error),
}
}
pub fn map_err<F, G>(self, f: F) -> FutureState<T, G>
where
F: FnOnce(E) -> G,
{
match self {
FutureState::Idle => FutureState::Idle,
FutureState::Pending => FutureState::Pending,
FutureState::Resolved(value) => FutureState::Resolved(value),
FutureState::Error(error) => FutureState::Error(f(error)),
}
}
}
impl<T, E> From<Result<T, E>> for FutureState<T, E> {
fn from(result: Result<T, E>) -> Self {
match result {
Ok(value) => FutureState::Resolved(value),
Err(error) => FutureState::Error(error),
}
}
}
#[derive(Debug, Default)]
pub struct FutureHandle<T, E = Box<dyn std::error::Error + Send + Sync>> {
state: Arc<RwLock<FutureState<T, E>>>,
task_handle: Arc<Mutex<Option<JoinHandle<()>>>>,
}
impl<T, E> FutureHandle<T, E>
where
T: Clone,
E: Clone,
{
fn new() -> Self {
Self {
state: Arc::new(RwLock::new(FutureState::Idle)),
task_handle: Arc::new(Mutex::new(None)),
}
}
pub fn state(&self) -> FutureState<T, E> {
self.state.read().clone()
}
pub fn is_idle(&self) -> bool {
matches!(&*self.state.read(), FutureState::Idle)
}
pub fn is_pending(&self) -> bool {
matches!(&*self.state.read(), FutureState::Pending)
}
pub fn is_resolved(&self) -> bool {
matches!(&*self.state.read(), FutureState::Resolved(_))
}
pub fn is_error(&self) -> bool {
matches!(&*self.state.read(), FutureState::Error(_))
}
pub fn value(&self) -> Option<T> {
match &*self.state.read() {
FutureState::Resolved(value) => Some(value.clone()),
_ => None,
}
}
pub fn error(&self) -> Option<E> {
match &*self.state.read() {
FutureState::Error(error) => Some(error.clone()),
_ => None,
}
}
pub fn cancel(&self) {
if let Some(task_handle) = self.task_handle.lock().take() {
task_handle.abort();
}
}
fn set_state(&self, new_state: FutureState<T, E>) {
*self.state.write() = new_state;
}
fn set_task_handle(&self, handle: JoinHandle<()>) {
*self.task_handle.lock() = Some(handle);
}
}
impl<T, E> Clone for FutureHandle<T, E> {
fn clone(&self) -> Self {
Self {
state: self.state.clone(),
task_handle: self.task_handle.clone(),
}
}
}
impl<T, E> Drop for FutureHandle<T, E> {
fn drop(&mut self) {
if let Some(task_handle) = self.task_handle.lock().take() {
task_handle.abort();
}
}
}
struct FutureHookState<T, E> {
prev_deps: Option<Box<dyn EffectDependencies>>,
handle: FutureHandle<T, E>,
initialized: bool,
active_futures: Arc<AtomicUsize>,
}
impl<T, E> FutureHookState<T, E>
where
T: Clone,
E: Clone,
{
fn new() -> Self {
Self {
prev_deps: None,
handle: FutureHandle::new(),
initialized: false,
active_futures: Arc::new(AtomicUsize::new(0)),
}
}
}
impl<T, E> Drop for FutureHookState<T, E> {
fn drop(&mut self) {
self.prev_deps = None;
if let Some(task_handle) = self.handle.task_handle.lock().take() {
task_handle.abort();
let active_count = self.active_futures.load(Ordering::Relaxed);
if active_count > 0 {
self.active_futures.fetch_sub(1, Ordering::Relaxed);
GLOBAL_ACTIVE_FUTURES.fetch_sub(1, Ordering::Relaxed);
}
}
}
}
#[derive(Debug, Clone)]
pub struct FutureError {
message: String,
}
impl fmt::Display for FutureError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "Future error: {}", self.message)
}
}
impl std::error::Error for FutureError {}
pub fn use_future<Deps, F, Fut, T, E>(
future_factory: F,
deps: impl Into<Option<Deps>>,
) -> FutureHandle<T, E>
where
Deps: EffectDependencies + Clone + PartialEq + 'static,
F: FnOnce() -> Fut + Send + 'static,
Fut: std::future::Future<Output = Result<T, E>> + Send + 'static,
T: Clone + Send + Sync + 'static,
E: Clone + Send + Sync + From<String> + 'static,
{
let deps = deps.into();
with_hook_context(|ctx| {
let hook_index = ctx.next_hook_index();
let mut states = ctx.states.borrow_mut();
let future_state = states
.entry(hook_index)
.or_insert_with(|| Box::new(FutureHookState::<T, E>::new()))
.downcast_mut::<FutureHookState<T, E>>()
.expect("Future state type mismatch");
let should_run = match &deps {
None => {
true
}
Some(current_deps) => {
match &future_state.prev_deps {
None => {
true
}
Some(prev_deps) => {
if prev_deps.deps_hash() != current_deps.deps_hash() {
true
} else {
!current_deps.deps_eq(prev_deps.as_ref())
}
}
}
}
};
if should_run {
let component_futures = future_state.active_futures.load(Ordering::Relaxed);
let global_futures = GLOBAL_ACTIVE_FUTURES.load(Ordering::Relaxed);
if component_futures >= MAX_CONCURRENT_FUTURES_PER_COMPONENT {
future_state.handle.set_state(FutureState::Error(
format!(
"Component future limit exceeded: {}/{}",
component_futures, MAX_CONCURRENT_FUTURES_PER_COMPONENT
)
.into(),
));
return future_state.handle.clone();
}
if global_futures >= MAX_GLOBAL_CONCURRENT_FUTURES {
future_state.handle.set_state(FutureState::Error(
format!(
"Global future limit exceeded: {}/{}",
global_futures, MAX_GLOBAL_CONCURRENT_FUTURES
)
.into(),
));
return future_state.handle.clone();
}
future_state.handle.cancel();
future_state.handle.set_state(FutureState::Pending);
if let Some(current_deps) = &deps {
future_state.prev_deps = Some(current_deps.clone_deps());
} else {
future_state.prev_deps = None;
}
future_state.active_futures.fetch_add(1, Ordering::Relaxed);
GLOBAL_ACTIVE_FUTURES.fetch_add(1, Ordering::Relaxed);
let handle_clone_for_success = future_state.handle.clone();
let handle_clone_for_error = future_state.handle.clone();
let handle_clone_for_panic = future_state.handle.clone();
let active_futures_clone = future_state.active_futures.clone();
let task_handle = tokio::spawn(async move {
let result = spawn_catch_panic(async move {
match future_factory().await {
Ok(value) => {
handle_clone_for_success.set_state(FutureState::Resolved(value));
}
Err(error) => {
handle_clone_for_error.set_state(FutureState::Error(error));
}
}
})
.await;
if let Err(join_error) = result {
let panic_message = if join_error.is_panic() {
format!("Future panicked: {}", join_error)
} else if join_error.is_cancelled() {
"Future was cancelled".to_string()
} else {
format!("Future failed: {}", join_error)
};
handle_clone_for_panic.set_state(FutureState::Error(panic_message.into()));
}
active_futures_clone.fetch_sub(1, Ordering::Relaxed);
GLOBAL_ACTIVE_FUTURES.fetch_sub(1, Ordering::Relaxed);
});
future_state.handle.set_task_handle(task_handle);
future_state.initialized = true;
}
future_state.handle.clone()
})
}