timerwheel 0.1.0

// Copyright © 2026-present The Timerwheel Authors. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::fmt;
use std::panic::{self, AssertUnwindSafe};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use crate::error::{Error, Result};
use crate::executor::metric_sink::NoopMetricSink;
use crate::executor::metrics::ExecutorMetricsInner;
use crate::executor::{self, BoxTask, ExecutorMetrics};
use crate::policy::{BackpressurePolicy, ExpiredTaskPolicy};
use crate::timer::{Timeout, TimerMetrics};

/// Executor adapter backed by the current Tokio runtime handle.
pub struct Executor {
    handle: tokio::runtime::Handle,
    closed: AtomicBool,
    metrics: Arc<ExecutorMetricsInner>,
}

impl Executor {
    /// Creates an executor from the currently-entered Tokio runtime handle.
    pub fn current() -> Result<Self> {
        let handle = tokio::runtime::Handle::try_current().map_err(|_| Error::Closed)?;
        Ok(Self {
            handle,
            closed: AtomicBool::new(false),
            metrics: Arc::new(ExecutorMetricsInner::new(0, Arc::new(NoopMetricSink))),
        })
    }
}

impl executor::Executor for Executor {
    fn try_execute(&self, task: BoxTask) -> std::result::Result<(), executor::RejectedTask> {
        if self.closed.load(Ordering::Acquire) {
            return Err(executor::RejectedTask::new(Error::Closed, task));
        }

        let metrics = Arc::clone(&self.metrics);
        self.metrics.submitted();
        self.handle.spawn(async move {
            let result = panic::catch_unwind(AssertUnwindSafe(|| {
                task.run();
            }));
            if result.is_err() {
                metrics.panicked();
            }
            metrics.completed();
        });
        Ok(())
    }

    fn shutdown(&self) -> Result<()> {
        self.closed.store(true, Ordering::Release);
        Ok(())
    }

    fn metrics(&self) -> ExecutorMetrics {
        self.metrics.snapshot()
    }
}

impl fmt::Debug for Executor {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Executor")
            .field("closed", &self.closed.load(Ordering::Acquire))
            .field("metrics", &self.metrics.snapshot())
            .finish_non_exhaustive()
    }
}

#[derive(Debug)]
/// Timer facade that uses the Tokio executor adapter by default.
pub struct Timer {
    inner: crate::Timer,
}

impl Timer {
    /// Returns a builder for a Tokio-backed timer facade.
    pub fn builder() -> TimerBuilder {
        TimerBuilder::default()
    }

    /// Schedules a closure to run after the given delay.
    pub fn schedule<F>(&self, delay: Duration, task: F) -> Result<Timeout>
    where
        F: FnOnce() + Send + 'static,
    {
        self.inner.schedule(delay, task)
    }

    /// Stops the timer scheduler.
    pub fn shutdown(&self) -> Result<()> {
        self.inner.shutdown()
    }

    /// Returns an immutable metrics snapshot.
    pub fn metrics(&self) -> TimerMetrics {
        self.inner.metrics()
    }

    /// Returns the number of pending timeouts currently owned by the timer.
    pub fn len(&self) -> usize {
        self.inner.len()
    }

    /// Returns true when the timer has no pending timeouts.
    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }
}

#[derive(Clone, Default)]
/// Builder for the Tokio timer facade.
pub struct TimerBuilder {
    inner: crate::TimerBuilder,
}

impl TimerBuilder {
    /// Sets the timer-wheel tick resolution.
    pub fn tick(mut self, tick: Duration) -> Self {
        self.inner = self.inner.tick(tick);
        self
    }

    /// Sets the number of buckets in each timer-wheel level.
    pub fn bucket_count(mut self, bucket_count: usize) -> Self {
        self.inner = self.inner.bucket_count(bucket_count);
        self
    }

    /// Sets the bounded scheduler command queue capacity.
    pub fn command_capacity(mut self, command_capacity: usize) -> Self {
        self.inner = self.inner.command_capacity(command_capacity);
        self
    }

    /// Sets the maximum number of pending timeouts accepted by the timer.
    pub fn max_pending(mut self, max_pending: usize) -> Self {
        self.inner = self.inner.max_pending(max_pending);
        self
    }

    /// Sets the policy used when the scheduler command queue is full.
    pub fn backpressure_policy(mut self, policy: BackpressurePolicy) -> Self {
        self.inner = self.inner.backpressure_policy(policy);
        self
    }

    /// Sets the policy used when the Tokio executor rejects an expired task.
    pub fn expired_task_policy(mut self, policy: ExpiredTaskPolicy) -> Self {
        self.inner = self.inner.expired_task_policy(policy);
        self
    }

    /// Sets the retry delay for expired tasks rejected by a saturated executor.
    pub fn expired_task_retry(mut self, retry: Duration) -> Self {
        self.inner = self.inner.expired_task_retry(retry);
        self
    }

    /// Builds the timer facade using the current Tokio runtime handle.
    pub fn build(self) -> Result<Timer> {
        let executor: Arc<dyn executor::Executor> = Arc::new(Executor::current()?);
        Ok(Timer {
            inner: self.inner.executor(executor).build()?,
        })
    }
}

impl fmt::Debug for TimerBuilder {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("TimerBuilder").finish_non_exhaustive()
    }
}