#![cfg_attr(docsrs, feature(doc_cfg))]

#![doc = include_str!("../README.md")]

// @@ begin lint list maintained by maint/add_warning @@

#![allow(renamed_and_removed_lints)] // @@REMOVE_WHEN(ci_arti_stable)

#![allow(unknown_lints)] // @@REMOVE_WHEN(ci_arti_nightly)

#![warn(missing_docs)]

#![warn(noop_method_call)]

#![warn(unreachable_pub)]

#![warn(clippy::all)]

#![deny(clippy::await_holding_lock)]

#![deny(clippy::cargo_common_metadata)]

#![deny(clippy::cast_lossless)]

#![deny(clippy::checked_conversions)]

#![allow(clippy::cognitive_complexity)] // See arti#2556

#![deny(clippy::debug_assert_with_mut_call)]

#![deny(clippy::exhaustive_enums)]

#![deny(clippy::exhaustive_structs)]

#![deny(clippy::expl_impl_clone_on_copy)]

#![deny(clippy::fallible_impl_from)]

#![deny(clippy::implicit_clone)]

#![deny(clippy::large_stack_arrays)]

#![warn(clippy::manual_ok_or)]

#![deny(clippy::missing_docs_in_private_items)]

#![warn(clippy::needless_borrow)]

#![warn(clippy::needless_pass_by_value)]

#![warn(clippy::option_option)]

#![warn(clippy::or_fun_call)]

#![deny(clippy::print_stderr)]

#![deny(clippy::print_stdout)]

#![warn(clippy::rc_buffer)]

#![deny(clippy::ref_option_ref)]

#![warn(clippy::semicolon_if_nothing_returned)]

#![warn(clippy::trait_duplication_in_bounds)]

#![deny(clippy::unchecked_time_subtraction)]

#![deny(clippy::unnecessary_wraps)]

#![warn(clippy::unseparated_literal_suffix)]

#![deny(clippy::unwrap_used)]

#![deny(clippy::mod_module_files)]

#![allow(clippy::let_unit_value)] // This can reasonably be done for explicitness

#![allow(clippy::uninlined_format_args)]

#![allow(clippy::significant_drop_in_scrutinee)] // arti/-/merge_requests/588/#note_2812945

#![allow(clippy::result_large_err)] // temporary workaround for arti#587

#![allow(clippy::needless_raw_string_hashes)] // complained-about code is fine, often best

#![allow(clippy::needless_lifetimes)] // See arti#1765

#![allow(mismatched_lifetime_syntaxes)] // temporary workaround for arti#2060

#![allow(clippy::collapsible_if)] // See arti#2342

#![deny(clippy::unused_async)]

#![deny(clippy::string_slice)] // See arti#2571

#![allow(recursion_depth_exceeding_limit)] // arti#2715, rust/issues/159228

//! <!-- @@ end lint list maintained by maint/add_warning @@ -->

pub mod events;

use crate::events::{TorEvent, TorEventKind};

use async_broadcast::{InactiveReceiver, Receiver, Sender, TrySendError};

use futures::StreamExt;

use futures::channel::mpsc;

use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};

use futures::future::Either;

use std::pin::Pin;

use std::sync::OnceLock;

use std::sync::atomic::{AtomicUsize, Ordering};

use std::task::{Context, Poll};

use thiserror::Error;

use tracing::{error, warn};

/// Pointer to an `UnboundedSender`, used to send events into the `EventReactor`.

static EVENT_SENDER: OnceLock<UnboundedSender<TorEvent>> = OnceLock::new();

/// An inactive receiver for the currently active broadcast channel, if there is one.

static CURRENT_RECEIVER: OnceLock<InactiveReceiver<TorEvent>> = OnceLock::new();

/// The number of `TorEventKind`s there are.

const EVENT_KIND_COUNT: usize = 1;

/// An array containing one `AtomicUsize` for each `TorEventKind`, used to track subscriptions.

///

/// When a `TorEventReceiver` subscribes to a `TorEventKind`, it uses its `usize` value to index

/// into this array and increment the associated `AtomicUsize` (and decrements it to unsubscribe).

/// This lets event emitters check whether there are any subscribers, and avoid emitting events

/// if there aren't.

static EVENT_SUBSCRIBERS: [AtomicUsize; EVENT_KIND_COUNT] = [AtomicUsize::new(0); EVENT_KIND_COUNT];

/// The size of the internal broadcast channel used to implement event subscription.

pub static BROADCAST_CAPACITY: usize = 512;

/// A reactor used to forward events to make the event reporting system work.

///

/// # Note

///

/// Currently, this type is a singleton; there is one event reporting system used for the entire

/// program. This is not stable, and may change in future.

pub struct EventReactor {

    /// A receiver that the reactor uses to learn about incoming events.

    ///

    /// This is unbounded so that event publication doesn't have to be async.

    receiver: UnboundedReceiver<TorEvent>,

    /// A sender that the reactor uses to publish events.

    ///

    /// Events are only sent here if at least one subscriber currently wants them.

    broadcast: Sender<TorEvent>,

}

impl EventReactor {

    /// Initialize the event reporting system, returning a reactor that must be run for it to work,

    /// and a `TorEventReceiver` that can be used to extract events from the system. If the system

    /// has already been initialized, returns `None` instead of a reactor.

    ///

    /// # Warnings

    ///

    /// The returned reactor *must* be run with `EventReactor::run`, in a background async task.

    /// If it is not, the event system might consume unbounded amounts of memory.

        } else {

        }

    /// Get a `TorEventReceiver` to receive events from, assuming an `EventReactor` is already

    /// running somewhere. (If it isn't, returns `None`.)

    ///

    /// As noted in the type-level documentation, this function might not always work this way.

    /// Run the event forwarding reactor.

    ///

    /// You *must* call this function once a reactor is created.

                    // If the channel is full, do a blocking broadcast to wait for it to be

                    // not full, and log a warning about receivers lagging behind.

                }

            }

        }

        // It shouldn't be possible to get here, since we have globals keeping the channels

        // open. Still, if we somehow do, log an error about it.

}

/// An error encountered when trying to receive a `TorEvent`.

#[derive(Clone, Debug, Error)]

#[non_exhaustive]

pub enum ReceiverError {

    /// The receiver isn't subscribed to anything, so wouldn't ever return any events.

    #[error("No event subscriptions")]

    NoSubscriptions,

    /// The internal broadcast channel was closed, which shouldn't ever happen.

    #[error("Internal event broadcast channel closed")]

    ChannelClosed,

}

/// A receiver for `TorEvent`s emitted by other users of this crate.

///

/// To use this type, first subscribe to some kinds of event by calling

/// `TorEventReceiver::subscribe`. Then, consume events using the implementation of

/// `futures::stream::Stream`.

///

/// # Warning

///

/// Once interest in events has been signalled with `subscribe`, events must be continuously

/// read from the receiver in order to avoid excessive memory consumption.

#[derive(Clone, Debug)]

pub struct TorEventReceiver {

    /// If no events have been subscribed to yet, this is an `InactiveReceiver`; otherwise,

    /// it's a `Receiver`.

    inner: Either<Receiver<TorEvent>, InactiveReceiver<TorEvent>>,

    /// Whether we're subscribed to each event kind (if `subscribed[kind]` is true, we're

    /// subscribed to `kind`).

    subscribed: [bool; EVENT_KIND_COUNT],

}

impl futures::stream::Stream for TorEventReceiver {

    type Item = TorEvent;

                        // loop, since we weren't subscribed to that event

                    }

                }

            },

            Either::Right(_) => {

            }

        }

}

impl TorEventReceiver {

    /// Create a `TorEventReceiver` from an `InactiveReceiver` handle.

    /// Subscribe to a given kind of `TorEvent`.

    ///

    /// After calling this function, `TorEventReceiver::recv` will emit events of that kind.

    /// This function is idempotent (subscribing twice has the same effect as doing so once).

        // FIXME(eta): cloning is ungood, but hard to avoid

    /// Unsubscribe from a given kind of `TorEvent`.

    ///

    /// After calling this function, `TorEventReceiver::recv` will no longer emit events of that

    /// kind.

    /// This function is idempotent (unsubscribing twice has the same effect as doing so once).

        // If we're now not subscribed to anything, deactivate our channel.

            // FIXME(eta): cloning is ungood, but hard to avoid

}

impl Drop for TorEventReceiver {

            // FIXME(eta): duplicates logic from Self::unsubscribe, because it's not possible

            //             to go from a `usize` to a `TorEventKind`

        }

}

/// Returns a boolean indicating whether the event `kind` has any subscribers (as in,

/// whether `TorEventReceiver::subscribe` has been called with that event kind).

///

/// This is useful to avoid doing work to generate events that might be computationally expensive

/// to generate.

/// Broadcast the given `TorEvent` to any interested subscribers.

///

/// As an optimization, does nothing if the event has no subscribers (`event_has_subscribers`

/// returns false). (also does nothing if the event subsystem hasn't been initialized yet)

///

/// This function isn't intended for use outside Arti crates (as in, library consumers of Arti

/// shouldn't broadcast events!).

#[cfg(test)]

mod test {

    // @@ begin test lint list maintained by maint/add_warning @@

    #![allow(clippy::bool_assert_comparison)]

    #![allow(clippy::clone_on_copy)]

    #![allow(clippy::dbg_macro)]

    #![allow(clippy::mixed_attributes_style)]

    #![allow(clippy::print_stderr)]

    #![allow(clippy::print_stdout)]

    #![allow(clippy::single_char_pattern)]

    #![allow(clippy::unwrap_used)]

    #![allow(clippy::unchecked_time_subtraction)]

    #![allow(clippy::useless_vec)]

    #![allow(clippy::needless_pass_by_value)]

    #![allow(clippy::string_slice)] // See arti#2571

    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->

    use crate::{

        EventReactor, StreamExt, TorEvent, TorEventKind, broadcast, event_has_subscribers,

    };

    use std::sync::{Mutex, MutexGuard, OnceLock};

    use std::time::Duration;

    use tokio::runtime::Runtime;

    // HACK(eta): these tests need to run effectively singlethreaded, since they mutate global

    //            state. They *also* need to share the same tokio runtime, which the

    //            #[tokio::test] thing doesn't do (it makes a new runtime per test), because of

    //            the need to have a background singleton EventReactor.

    //

    //            To hack around this, we just have a global runtime protected by a mutex!

    static TEST_MUTEX: OnceLock<Mutex<Runtime>> = OnceLock::new();

    /// Locks the mutex, and makes sure the event reactor is initialized.

    fn test_setup() -> MutexGuard<'static, Runtime> {

        let mutex = TEST_MUTEX.get_or_init(|| Mutex::new(Runtime::new().unwrap()));

        let runtime = mutex

            .lock()

            .expect("mutex poisoned, probably by other failing tests");

        if let Some(reactor) = EventReactor::new() {

            runtime.handle().spawn(reactor.run());

        }

        runtime

    }

    #[test]

    fn subscriptions() {

        let rt = test_setup();

        rt.block_on(async move {

            // shouldn't have any subscribers at the start

            assert!(!event_has_subscribers(TorEventKind::Empty));

            let mut rx = EventReactor::receiver().unwrap();

            // creating a receiver shouldn't result in any subscriptions

            assert!(!event_has_subscribers(TorEventKind::Empty));

            rx.subscribe(TorEventKind::Empty);

            // subscription should work

            assert!(event_has_subscribers(TorEventKind::Empty));

            rx.unsubscribe(TorEventKind::Empty);

            // unsubscribing should work

            assert!(!event_has_subscribers(TorEventKind::Empty));

            // subscription should be idempotent

            rx.subscribe(TorEventKind::Empty);

            rx.subscribe(TorEventKind::Empty);

            rx.subscribe(TorEventKind::Empty);

            assert!(event_has_subscribers(TorEventKind::Empty));

            rx.unsubscribe(TorEventKind::Empty);

            assert!(!event_has_subscribers(TorEventKind::Empty));

            rx.subscribe(TorEventKind::Empty);

            assert!(event_has_subscribers(TorEventKind::Empty));

            std::mem::drop(rx);

            // dropping the receiver should auto-unsubscribe

            assert!(!event_has_subscribers(TorEventKind::Empty));

        });

    }

    #[test]

    fn empty_recv() {

        let rt = test_setup();

        rt.block_on(async move {

            let mut rx = EventReactor::receiver().unwrap();

            // attempting to read from a receiver with no subscriptions should return None

            let result = rx.next().await;

            assert!(result.is_none());

        });

    }

    #[test]

    fn receives_events() {

        let rt = test_setup();

        rt.block_on(async move {

            let mut rx = EventReactor::receiver().unwrap();

            rx.subscribe(TorEventKind::Empty);

            // HACK(eta): give the event reactor time to run

            tokio::time::sleep(Duration::from_millis(100)).await;

            broadcast(TorEvent::Empty);

            let result = rx.next().await;

            assert_eq!(result, Some(TorEvent::Empty));

        });

    }

    #[test]

    fn does_not_send_to_no_subscribers() {

        let rt = test_setup();

        rt.block_on(async move {

            // this event should just get dropped on the floor, because no subscribers exist

            broadcast(TorEvent::Empty);

            let mut rx = EventReactor::receiver().unwrap();

            rx.subscribe(TorEventKind::Empty);

            // this shouldn't have an event to receive now

            let result = tokio::time::timeout(Duration::from_millis(100), rx.next()).await;

            assert!(result.is_err());

        });

    }

}