"mio 1.0.4",

default = ["backtrace", "systemd", "structured-logger", "socket2"]

cyphernet = { workspace = true, features = ["tor", "dns", "ed25519", "p2p-ed25519", "noise-framework", "noise_sha2"] }

log = { workspace = true, features = ["kv", "std"] }

mio = { version = "1", features = ["net", "os-poll"] }

socket2 = { version = "0.5.7", features = ["all"], optional = true }

mio = { version = "1", features = ["os-ext"] }

pub mod reactor;

mod controller;

mod listener;

mod session;

mod timer;

mod token;

mod transport;

use std::collections::HashMap;

use std::fmt::{Debug, Display, Formatter};

use std::io::ErrorKind;

use std::sync::Arc;

use std::thread::JoinHandle;

use std::time::Duration;

use std::{io, thread};

use crossbeam_channel::{unbounded, Receiver, TryRecvError};

use localtime::LocalTime;

use mio::event::{Event, Source};

use mio::{Events, Interest, Poll, Waker};

use thiserror::Error;

use timer::Timer;

use token::WAKER;

use crate::wire;

pub(crate) use self::controller::{ControlMessage, Controller};

pub(crate) use listener::Listener;

pub use session::{NoiseSession, ProtocolArtifact, Socks5Session};

pub(crate) use token::{Token, Tokens};

pub(crate) use transport::{SessionEvent, Transport};

const SECONDS_IN_AN_HOUR: u64 = 60 * 60;

/// Maximum amount of time to wait for I/O.

const WAIT_TIMEOUT: Duration = Duration::from_secs(SECONDS_IN_AN_HOUR);

/// A resource which can be managed by the reactor.

pub trait EventHandler {

    /// The type of reactions which this resource may generate upon receiving

    /// I/O from the reactor via [`EventHandler::handle`]. These events are

    /// passed to the reactor [`crate::reactor::ReactionHandler`].

    type Reaction;

    /// Method informing the reactor which types of events this resource is subscribed for.

    fn interests(&self) -> Option<Interest>;

    /// Method called by the reactor when an I/O readiness event

    /// is received for this resource.

    fn handle(&mut self, event: &Event) -> Vec<Self::Reaction>;

}

/// The trait guarantees that the data are either written in full or, in case

/// of an error, none of the data is written. Types implementing the trait must

/// also guarantee that multiple attempts to write do not result in

/// data to be written out of the initial ordering.

pub trait WriteAtomic: std::io::Write {

    /// Atomic non-blocking I/O write operation, which must either write the whole buffer to a

    /// resource without blocking or fail.

    ///

    /// # Panics

    ///

    /// If [`WriteAtomic::write_or_buf`] returns an [`std::io::Error`] of kind

    /// [`ErrorKind::Interrupted`], [`ErrorKind::WouldBlock`], [`ErrorKind::WriteZero`].

    /// In this case, [`WriteAtomic::write_or_buf`] is expected to buffer.

    fn write_atomic(&mut self, buf: &[u8]) -> io::Result<()> {

        use ErrorKind::*;

        if !self.is_ready_to_write() {

            panic!("WriteAtomic::write_atomic was called when the resource is not ready to write");

        }

        let result = self.write_or_buf(buf);

        debug_assert!(

            !matches!(

                result.as_ref().err().map(|err| err.kind()),

                Some(Interrupted | WouldBlock | WriteZero)

            ),

            "WriteAtomic::write_or_buf must handle erros of kind {Interrupted:?}, {WouldBlock:?}, {WriteZero:?} by buffering",

        );

        result

    }

    /// Checks whether resource can be written to without blocking.

    fn is_ready_to_write(&self) -> bool;

    /// Writes to the resource in a non-blocking way, buffering the data if necessary,

    /// or failing with a system-level error.

    ///

    /// This method shouldn't be called directly; call [`WriteAtomic::write_atomic`] instead.

    ///

    /// The method must handle [`std::io::Error`] of kind

    /// [`ErrorKind::Interrupted`], [`ErrorKind::WouldBlock`], [`ErrorKind::WriteZero`].

    /// and buffer the data in such cases.

    fn write_or_buf(&mut self, buf: &[u8]) -> io::Result<()>;

}

/// Reactor errors

#[derive(Error)]

pub enum Error<L: EventHandler, T: EventHandler> {

    #[error("listener {0:?} got disconnected during poll operation")]

    ListenerDisconnect(Token, L),

    #[error("transport {0:?} got disconnected during poll operation")]

    TransportDisconnect(Token, T),

    #[error("registration of a resource has failed: {0}")]

    Poll(io::Error),

    #[error("registration of a resource has failed: {0}")]

    Registration(io::Error),

}

impl<L: EventHandler, T: EventHandler> Debug for Error<L, T> {

    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {

        Display::fmt(self, f)

    }

}

/// Actions which can be provided to the [`Reactor`] by the [`ReactionHandler`].

///

/// Reactor reads actions on each event loop using [`ReactionHandler`] iterator interface.

pub enum Action<L, T> {

    /// Register a new listener resource for the reactor poll.

    ///

    /// Reactor can't instantiate the resource, like bind a network listener.

    /// Reactor only can register already active resource for polling in the event loop.

    RegisterListener(Token, L),

    /// Register a new transport resource for the reactor poll.

    ///

    /// Reactor can't instantiate the resource, like open a file or establish network connection.

    /// Reactor only can register already active resource for polling in the event loop.

    RegisterTransport(Token, T),

    /// Unregister listener resource from the reactor poll and handover it to the [`ReactionHandler`] via

    /// [`ReactionHandler::handover_listener`].

    ///

    /// When the resource is unregistered no action is performed, i.e. the file descriptor is not

    /// closed, listener is not unbound, connections are not closed etc. All these actions must be

    /// handled by the handler upon the handover event.

    #[allow(dead_code)] // For future use

    UnregisterListener(Token),

    /// Unregister transport resource from the reactor poll and handover it to the [`ReactionHandler`] via

    /// [`ReactionHandler::handover_transport`].

    ///

    /// When the resource is unregistered no action is performed, i.e. the file descriptor is not

    /// closed, listener is not unbound, connections are not closed etc. All these actions must be

    /// handled by the handler upon the handover event.

    UnregisterTransport(Token),

    /// Write the data to one of the transport resources using [`io::Write`].

    Send(Token, Vec<u8>),

    /// Set a new timer for a given duration from this moment.

    ///

    /// When the timer elapses, the reactor will timeout from poll and call

    /// [`ReactionHandler::timer_reacted`].

    SetTimer(Duration),

}

impl<L: EventHandler, T: EventHandler> Display for Action<L, T> {

    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {

        match self {

            Action::RegisterListener(token, _listener) => f

                .debug_struct("RegisterListener")

                .field("token", token)

                .field("listener", &"<omitted>")

                .finish(),

            Action::RegisterTransport(token, _transport) => f

                .debug_struct("RegisterTransport")

                .field("token", token)

                .field("transport", &"<omitted>")

                .finish(),

            Action::UnregisterListener(token) => f

                .debug_struct("UnregisterListener")

                .field("token", token)

                .finish(),

            Action::UnregisterTransport(token) => f

                .debug_struct("UnregisterTransport")

                .field("token", token)

                .finish(),

            Action::Send(token, _data) => f

                .debug_struct("Send")

                .field("token", token)

                .field("data", &"<omitted>")

                .finish(),

            Action::SetTimer(duration) => f

                .debug_struct("SetTimer")

                .field("duration", duration)

                .finish(),

        }

    }

}

/// A service which handles reactions to the events generated in the [`Reactor`].

pub trait ReactionHandler: Send + Iterator<Item = Action<Self::Listener, Self::Transport>> {

    /// Type for a listener resource.

    ///

    /// Listener resources are resources which may spawn more resources and can't be written to. A

    /// typical example of a listener resource is a [`std::net::TcpListener`], however this may also

    /// be a special form of a peripheral device or something else.

    type Listener: EventHandler + Source + Send;

    /// Type for a transport resource.

    ///

    /// Transport is a "full" resource which can be read from - and written to. Usual files, network

    /// connections, database connections etc are all fall into this category.

    type Transport: EventHandler + Source + Send + WriteAtomic;

    /// Method called by the reactor on the start of each event loop once the poll has returned.

    fn tick(&mut self, time: localtime::LocalTime);

    /// Method called by the reactor when a previously set timeout is fired.

    ///

    /// Related: [`Action::SetTimer`].

    fn timer_reacted(&mut self);

    /// Method called by the reactor upon a reaction to an I/O event on a listener resource.

    ///

    /// Since listener doesn't support writing, it can be only a read event (indicating that a new

    /// resource can be spawned from the listener).

    fn listener_reacted(

        &mut self,

        token: Token,

        reaction: <Self::Listener as EventHandler>::Reaction,

        time: localtime::LocalTime,

    );

    /// Method called by the reactor upon a reaction to an I/O event on a transport resource.

    fn transport_reacted(

        &mut self,

        token: Token,

        reaction: <Self::Transport as EventHandler>::Reaction,

        time: localtime::LocalTime,

    );

    /// Method called by the reactor when a given resource was successfully registered

    /// for given token.

    ///

    /// The token will be used later in [`ReactionHandler::listener_reacted`]

    /// and [`ReactionHandler::handover_listener`] calls to the handler.

    fn listener_registered(&mut self, token: Token, listener: &Self::Listener);

    /// Method called by the reactor when a given resource was successfully registered

    /// for given token.

    ///

    /// The token will be used later in [`ReactionHandler::transport_reacted`],

    /// [`ReactionHandler::handover_transport`] calls to the handler.

    fn transport_registered(&mut self, token: Token, transport: &Self::Transport);

    /// Method called by the reactor when a command is received for the

    /// [`ReactionHandler`].

    ///

    /// The commands are sent via `Controller` from outside of the reactor, including other

    /// threads.

    fn handle_command(&mut self, cmd: wire::Control);

    /// Method called by the reactor on any kind of error during the event loop, including errors of

    /// the poll syscall or I/O errors returned as a part of the poll result events.

    ///

    /// See [`enum@Error`] for the details on errors which may happen.

    fn handle_error(&mut self, err: Error<Self::Listener, Self::Transport>);

    /// Method called by the reactor upon receiving [`Action::UnregisterListener`].

    ///

    /// Passes the listener resource to the [`ReactionHandler`] when it is already not a part of the reactor

    /// poll. From this point of time it is safe to send the resource to other threads (like

    /// workers) or close the resource.

    fn handover_listener(&mut self, token: Token, listener: Self::Listener);

    /// Method called by the reactor upon receiving [`Action::UnregisterTransport`].

    ///

    /// Passes the transport resource to the [`ReactionHandler`] when it is already not a part of the

    /// reactor poll. From this point of time it is safe to send the resource to other threads

    /// (like workers) or close the resource.

    fn handover_transport(&mut self, token: Token, transport: Self::Transport);

}

/// High-level reactor API wrapping reactor [`Runtime`] into a thread and providing basic thread

/// management for it.

///

/// Apps running the [`Reactor`] can interface it and a [`ReactionHandler`] via use of the `Controller`

/// API.

pub struct Reactor {

    thread: JoinHandle<()>,

    controller: Controller,

}

impl Reactor {

    /// Creates new reactor and a service exposing the [`ReactionHandler`] to

    /// the reactor.

    ///

    /// The service is sent to the newly created reactor thread which runs the

    /// reactor [`Runtime`].

    pub fn new<H>(service: H, thread_name: String) -> Result<Self, io::Error>

    where

        H: 'static + ReactionHandler,

    {

        let builder = thread::Builder::new().name(thread_name);

        let (sender, receiver) = unbounded();

        let poll = Poll::new()?;

        let controller = Controller::new(sender, Arc::new(Waker::new(poll.registry(), WAKER)?));

        log::debug!(target: "reactor-controller", "Initializing reactor thread...");

        let thread = builder.spawn(move || {

            let runtime = Runtime {

                service,

                poll,

                receiver,

                listeners: HashMap::new(),

                transports: HashMap::new(),

                timeouts: Timer::new(),

            };

            log::info!(target: "reactor", "Entering reactor event loop");

            runtime.run();

        })?;

        // Waking up to consume actions which were provided by the service on launch

        controller.wake()?;

        Ok(Self { thread, controller })

    }

    /// Provides a `Controller` that can be used to send events to

    /// [`ReactionHandler`] via self.

    pub fn controller(&self) -> Controller {

        self.controller.clone()

    }

    /// Joins the reactor thread.

    pub fn join(self) -> thread::Result<()> {

        self.thread.join()

    }

}

/// Internal [`Reactor`] runtime which is run in a dedicated thread.

///

/// This runtime structure *does not* spawn a thread and is *blocking*.

/// It implements the actual reactor event loop.

pub struct Runtime<H: ReactionHandler> {

    service: H,

    poll: Poll,

    receiver: Receiver<ControlMessage>,

    listeners: HashMap<Token, H::Listener>,

    transports: HashMap<Token, H::Transport>,

    timeouts: Timer,

}

impl<H: ReactionHandler> Runtime<H> {

    fn register_interests(&mut self) -> io::Result<()> {

        let registry = self.poll.registry();

        for (id, res) in self.listeners.iter_mut() {

            match res.interests() {

                None => registry.deregister(res)?,

                Some(interests) => registry.reregister(res, *id, interests)?,

            };

        }

        for (id, res) in self.transports.iter_mut() {

            match res.interests() {

                None => registry.deregister(res)?,

                Some(interests) => registry.reregister(res, *id, interests)?,

            };

        }

        Ok(())

    }

    fn run(mut self) {

        loop {

            let before_poll = LocalTime::now();

            let timeout = self

                .timeouts

                .next_expiring_from(before_poll)

                .unwrap_or(WAIT_TIMEOUT);

            self.register_interests()

                .expect("registering interests must work to ensure correct operation");

            log::trace!(target: "reactor", "Polling with timeout {timeout:?}");

            let mut events = Events::with_capacity(1024);

            // Blocking

            let res = self.poll.poll(&mut events, Some(timeout));

            let now = LocalTime::now();

            self.service.tick(now);

            // The way this is currently used basically ignores which keys have

            // timed out. So as long as *something* timed out, we wake the service.

            let timers_fired = self.timeouts.remove_expired_by(now);

            if timers_fired > 0 {

                log::trace!(target: "reactor", "Timer has fired");

                self.service.timer_reacted();

            }

            if let Err(err) = res {

                log::error!(target: "reactor", "Error during polling: {err}");

                self.service.handle_error(Error::Poll(err));

            }

            let awoken = self.handle_events(now, events);

            // Process the commands only if we awoken by the waker.

            if awoken {

                loop {

                    match self.receiver.try_recv() {

                        Err(TryRecvError::Empty) => break,

                        Err(TryRecvError::Disconnected) => {

                            panic!("control channel disconnected unexpectedly")

                        }

                        Ok(ControlMessage::Shutdown) => return self.handle_shutdown(),

                        Ok(ControlMessage::Command(cmd)) => self.service.handle_command(*cmd),

                    }

                }

            }

            self.handle_actions(now);

        }

    }

    /// # Returns

    ///

    /// Whether one of the events was originated from the waker.

    fn handle_events(&mut self, time: LocalTime, events: Events) -> bool {

        let mut awoken = false;

        for event in events.into_iter() {

            let id = event.token();

            if id == WAKER {

                log::trace!(target: "reactor", "Awoken by the controller");

                awoken = true;

            } else if self.listeners.contains_key(&id) {

                log::trace!(target: "reactor", event:debug; "From listener");

                if !event.is_error() {

                    let listener = self.listeners.get_mut(&id).expect("resource disappeared");

                    listener

                        .handle(event)

                        .into_iter()

                        .for_each(|service_event| {

                            self.service.listener_reacted(id, service_event, time);

                        });

                } else {

                    let listener = self

                        .unregister_listener(id)

                        .expect("listener has disappeared");

                    self.service

                        .handle_error(Error::ListenerDisconnect(id, listener));

                }

            } else if self.transports.contains_key(&id) {

                log::trace!(target: "reactor", event:debug; "From transport");

                if !event.is_error() {

                    let transport = self.transports.get_mut(&id).expect("resource disappeared");

                    transport

                        .handle(event)

                        .into_iter()

                        .for_each(|service_event| {

                            self.service.transport_reacted(id, service_event, time);

                        });

                } else {

                    let transport = self

                        .unregister_transport(id)

                        .expect("transport has disappeared");

                    self.service

                        .handle_error(Error::TransportDisconnect(id, transport));

                }

            } else {

                panic!("token in poll which is not a known waker, listener or transport")

            }

        }

        awoken

    }

    fn handle_actions(&mut self, time: LocalTime) {

        while let Some(action) = self.service.next() {

            log::trace!(target: "reactor", "Handling action {action} from the service");

            // Deadlock may happen here if the service will generate events over and over

            // in the handle_* calls we may never get out of this loop

            if let Err(err) = self.handle_action(action, time) {

                log::error!(target: "reactor", "Error: {err}");

                self.service.handle_error(err);

            }

        }

    }

    fn handle_action(

        &mut self,

        action: Action<H::Listener, H::Transport>,

        time: LocalTime,

    ) -> Result<(), Error<H::Listener, H::Transport>> {

        match action {

            Action::RegisterListener(token, mut listener) => {

                log::debug!(target: "reactor", token=token.0; "Registering listener");

                self.poll

                    .registry()

                    .register(&mut listener, token, Interest::READABLE)

                    .map_err(Error::Registration)?;

                self.listeners.insert(token, listener);

                self.service

                    .listener_registered(token, &self.listeners[&token]);

            }

            Action::RegisterTransport(token, mut transport) => {

                log::debug!(target: "reactor", token=token.0; "Registering transport");

                self.poll

                    .registry()

                    .register(&mut transport, token, Interest::READABLE)

                    .map_err(Error::Registration)?;

                self.transports.insert(token, transport);

                self.service

                    .transport_registered(token, &self.transports[&token]);

            }

            Action::UnregisterListener(token) => {

                let Some(listener) = self.unregister_listener(token) else {

                    return Ok(());

                };

                log::debug!(target: "reactor", token=token.0; "Handing over listener");

                self.service.handover_listener(token, listener);

            }

            Action::UnregisterTransport(token) => {

                let Some(transport) = self.unregister_transport(token) else {

                    return Ok(());

                };

                log::debug!(target: "reactor", token=token.0; "Handing over transport");

                self.service.handover_transport(token, transport);

            }

            Action::Send(token, data) => {

                log::trace!(target: "reactor", "Sending {} bytes to {token:?}", data.len());

                if let Some(transport) = self.transports.get_mut(&token) {

                    if let Err(e) = transport.write_atomic(&data) {

                        log::error!(target: "reactor", "Fatal error writing to transport {token:?}, disconnecting. Error details: {e:?}");

                        if let Some(transport) = self.unregister_transport(token) {

                            return Err(Error::TransportDisconnect(token, transport));

                        }

                    }

                } else {

                    log::error!(target: "reactor", "Transport {token:?} is not in the reactor");

                }

            }

            Action::SetTimer(duration) => {

                log::trace!(target: "reactor", "Adding timer {duration:?} from now");

                self.timeouts.set_timeout(duration, time);

            }

        }

        Ok(())

    }

    fn handle_shutdown(self) {

        log::info!(target: "reactor", "Shutdown");

    }

    fn unregister_listener(&mut self, token: Token) -> Option<H::Listener> {

        let Some(mut source) = self.listeners.remove(&token) else {

            log::warn!(target: "reactor", token=token.0; "Unregistering non-registered listener");

            return None;

        };

        if let Err(err) = self.poll.registry().deregister(&mut source) {

            log::warn!(target: "reactor", token=token.0; "Failed to deregister listener from mio: {err}");

        }

        Some(source)

    }

    fn unregister_transport(&mut self, token: Token) -> Option<H::Transport> {

        let Some(mut source) = self.transports.remove(&token) else {

            log::warn!(target: "reactor", token=token.0; "Unregistering non-registered transport");

            return None;

        };

        if let Err(err) = self.poll.registry().deregister(&mut source) {

            log::warn!(target: "reactor", token=token.0; "Failed to deregister transport from mio: {err}");

        }

        Some(source)

    }

}

use crossbeam_channel::Sender;

use mio::Waker;

use std::io;

use std::io::ErrorKind;

use std::sync::Arc;

use crate::wire;

/// A command which may be sent to the [`super::ReactionHandler`] from outside of the [`super::Reactor`],

/// including other threads.

///

/// The handler object is owned by the reactor runtime and executes always in the context of the

/// reactor runtime thread. Thus, if other (micro)services within the app needs to communicate

/// to the handler they have to use this data type, which usually is an enumeration for a set of

/// commands supported by the handler.

pub enum ControlMessage {

    Command(Box<wire::Control>),

    Shutdown,

}

/// Used by the [`crate::reactor::Reactor`] to inform the

/// [`crate::reactor::ReactionHandler`] about

/// incoming commands, sent via this [`Controller`].

#[derive(Clone)]

pub struct Controller {

    sender: Sender<ControlMessage>,

    waker: Arc<Waker>,

}

impl Controller {

    pub fn new(sender: Sender<ControlMessage>, waker: Arc<Waker>) -> Self {

        Self { sender, waker }

    }

    pub fn wake(&self) -> io::Result<()> {

        log::trace!(target: "reactor::controller", "Wakening the reactor");

        self.waker.wake()

    }

    pub fn cmd(&self, command: wire::Control) -> io::Result<()> {

        log::trace!(target: "reactor::controller", "Sending command {command:?} to the reactor");

        self.sender

            .send(ControlMessage::Command(Box::new(command)))

            .map_err(|_| ErrorKind::BrokenPipe)?;

        self.wake()

    }

    pub fn shutdown(self) -> Result<(), Self> {

        log::info!(target: "reactor::controller", "Initiating reactor shutdown...");

        let res1 = self.sender.send(ControlMessage::Shutdown);

        let res2 = self.wake();

        res1.or(res2).map_err(|_| self)

    }

}

use mio::event::{Event, Source};

use mio::net::{TcpListener, TcpStream};

use mio::{Interest, Registry, Token};

use std::io::Result;

use std::net::SocketAddr;

use std::time::Duration;

use crate::reactor::EventHandler;

/// A reactor-manageable TCP listener which can

/// be aware of additional encryption, authentication and other forms of

/// transport-layer protocols which will be automatically injected into accepted

/// connections.

#[derive(Debug)]

pub struct Listener(TcpListener);

impl Source for Listener {

    fn register(&mut self, registry: &Registry, token: Token, interests: Interest) -> Result<()> {

        self.0.register(registry, token, interests)

    }

    fn reregister(&mut self, registry: &Registry, token: Token, interests: Interest) -> Result<()> {

        self.0.reregister(registry, token, interests)

    }

    fn deregister(&mut self, registry: &Registry) -> Result<()> {

        self.0.deregister(registry)

    }

}

impl Listener {

    pub fn bind(addr: SocketAddr) -> Result<Self> {

        Ok(Self(TcpListener::bind(addr)?))

    }

    /// Returns the local [`std::net::SocketAddr`] on which self accepts

    /// connections.

    pub fn local_addr(&self) -> std::net::SocketAddr {

        self.0.local_addr().expect("TCP listener has local address")

    }

    fn accept(&mut self) -> Result<(TcpStream, SocketAddr)> {

        /// Maximum time to wait when reading from a socket.

        const READ_TIMEOUT: Duration = Duration::from_secs(6);

        /// Maximum time to wait when writing to a socket.

        const WRITE_TIMEOUT: Duration = Duration::from_secs(3);

        let (stream, peer) = self.0.accept()?;

        let stream = std::net::TcpStream::from(stream);

        stream.set_read_timeout(Some(READ_TIMEOUT))?;

        stream.set_write_timeout(Some(WRITE_TIMEOUT))?;

        stream.set_nonblocking(true)?;

        Ok((TcpStream::from_std(stream), peer))

    }

}

impl EventHandler for Listener {

    type Reaction = Result<(TcpStream, SocketAddr)>;

    fn interests(&self) -> Option<Interest> {

        Some(Interest::READABLE)

    }

    fn handle(&mut self, event: &Event) -> Vec<Self::Reaction> {

        if !event.is_readable() {

            return vec![];

        }

        vec![self.accept()]

    }

}

use std::error;

use std::fmt::{Debug, Display};

use std::io;

use std::io::{Read, Write};

use std::net::{Shutdown, SocketAddr};

use cyphernet::encrypt::noise::NoiseState;

use cyphernet::proxy::socks5;

use mio::event::Source;

use mio::net::TcpStream;

use mio::{Interest, Registry, Token};

pub type NoiseSession<E, D, S> = Protocol<NoiseState<E, D>, S>;

pub type Socks5Session<S> = Protocol<socks5::Socks5, S>;

pub trait Session: Send + Read + Write {

    type Inner: Session;

    type Artifact: Display;

    fn is_established(&self) -> bool {

        self.artifact().is_some()

    }

    fn run_handshake(&mut self) -> io::Result<()> {

        Ok(())

    }

    fn display(&self) -> String {

        self.artifact()

            .map(|artifact| artifact.to_string())

            .unwrap_or_else(|| "<no-id>".to_string())

    }

    fn artifact(&self) -> Option<Self::Artifact>;

    fn stream(&mut self) -> &mut TcpStream;

    fn disconnect(self) -> io::Result<()>;

}

pub trait StateMachine: Sized + Send {

    const NAME: &'static str;

    type Artifact;

    type Error: error::Error + Send + Sync + 'static;

    fn next_read_len(&self) -> usize;

    fn advance(&mut self, input: &[u8]) -> Result<Vec<u8>, Self::Error>;

    fn artifact(&self) -> Option<Self::Artifact>;

    // Blocking

    fn run_handshake<RW>(&mut self, stream: &mut RW) -> io::Result<()>

    where

        RW: Read + Write,

    {

        let mut input = vec![];

        while !self.is_complete() {

            let act = self.advance(&input).map_err(|err| {

                log::error!(target: Self::NAME, "Handshake failure: {err}");

                io::Error::other(err)

            })?;

            if !act.is_empty() {

                log::trace!(target: Self::NAME, "Sending handshake act {act:02x?}");

                stream.write_all(&act)?;

            }

            if !self.is_complete() {

                input = vec![0u8; self.next_read_len()];

                stream.read_exact(&mut input)?;

                log::trace!(target: Self::NAME, "Receiving handshake act {input:02x?}");

            }

        }

        log::debug!(target: Self::NAME, "Handshake protocol {} successfully completed", Self::NAME);

        Ok(())

    }

    fn is_complete(&self) -> bool {

        self.artifact().is_some()

    }

}

#[derive(Clone, Eq, PartialEq, Hash, Debug)]

pub struct ProtocolArtifact<M: StateMachine, S: Session> {

    pub(crate) session: S::Artifact,

    pub(crate) state: M::Artifact,

}

impl<M: StateMachine, S: Session> Display for ProtocolArtifact<M, S> {

    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

        f.debug_struct("ProtocolArtifact")

            .field("session", &"<omitted>")

            .field("state", &"<omitted>")

            .finish()

    }

}

#[derive(Copy, Clone, Eq, PartialEq)]

pub struct Protocol<M: StateMachine, S: Session> {

    pub(crate) state: M,

    pub(crate) session: S,

}

impl<M: StateMachine, S: Session> Protocol<M, S> {

    pub fn new(session: S, state_machine: M) -> Self {

        Self {

            state: state_machine,

            session,

        }

    }

}

impl<M: StateMachine, S: Session> io::Read for Protocol<M, S> {

    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {

        log::trace!(target: M::NAME, "Reading event");

        if self.state.is_complete() || !self.session.is_established() {

            log::trace!(target: M::NAME, "Passing reading to inner not yet established session");

            return self.session.read(buf);

        }

        let len = self.state.next_read_len();

        let mut input = vec![0u8; len];

        self.session.read_exact(&mut input)?;

        log::trace!(target: M::NAME, "Received handshake act: {input:02x?}");

        if !input.is_empty() {

            let output = self.state.advance(&input).map_err(|err| {

                log::error!(target: M::NAME, "Handshake failure: {err}");

                io::Error::other(err)

            })?;

            if !output.is_empty() {

                log::trace!(target: M::NAME, "Sending handshake act on read: {output:02x?}");

                self.session.write_all(&output)?;

            }

        }

        Ok(0)

    }

}

impl<M: StateMachine, S: Session> Write for Protocol<M, S> {

    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {

        log::trace!(target: M::NAME, "Writing event (state_complete={}, session_established={})", self.state.is_complete(), self.session.is_established());

        if self.state.is_complete() || !self.session.is_established() {

            log::trace!(target: M::NAME, "Passing writing to inner session");

            return self.session.write(buf);

        }

        if self.state.next_read_len() == 0 {

            log::trace!(target: M::NAME, "Starting handshake protocol");

            let act = self.state.advance(&[]).map_err(|err| {

                log::error!(target: M::NAME, "Handshake failure: {err}");

                io::Error::other(err)

            })?;

            if !act.is_empty() {

                log::trace!(target: M::NAME, "Sending handshake act on write: {act:02x?}");

                self.session.write_all(&act)?;

            } else {

                log::trace!(target: M::NAME, "Handshake complete, passing data to inner session");

                return self.session.write(buf);

            }

        }

        if buf.is_empty() {

            Ok(0)

        } else {

            Err(io::ErrorKind::Interrupted.into())

        }

    }

    fn flush(&mut self) -> io::Result<()> {

        self.session.flush()

    }

}

impl<M: StateMachine, S: Session> Session for Protocol<M, S> {

    type Inner = S;

    type Artifact = ProtocolArtifact<M, S>;

    fn run_handshake(&mut self) -> io::Result<()> {

        log::debug!(target: M::NAME, "Starting handshake protocol {}", M::NAME);

        if !self.session.is_established() {

            self.session.run_handshake()?;

        }

        self.state.run_handshake(self.session.stream())

    }

    fn artifact(&self) -> Option<Self::Artifact> {

        Some(ProtocolArtifact {

            session: self.session.artifact()?,

            state: self.state.artifact()?,

        })

    }

    fn stream(&mut self) -> &mut TcpStream {

        self.session.stream()

    }

    fn disconnect(self) -> io::Result<()> {

        self.session.disconnect()

    }

}

impl<M: StateMachine, S: Session + Source> Source for Protocol<M, S> {

    fn register(

        &mut self,

        registry: &Registry,

        token: Token,

        interests: Interest,

    ) -> io::Result<()> {

        self.session.register(registry, token, interests)

    }

    fn reregister(

        &mut self,

        registry: &Registry,

        token: Token,

        interests: Interest,

    ) -> io::Result<()> {

        self.session.reregister(registry, token, interests)

    }

    fn deregister(&mut self, registry: &Registry) -> io::Result<()> {

        self.session.deregister(registry)

    }

}

impl Session for TcpStream {

    type Inner = Self;

    type Artifact = SocketAddr;

    fn artifact(&self) -> Option<Self::Artifact> {

        self.peer_addr().ok()

    }

    fn stream(&mut self) -> &mut TcpStream {

        self

    }

    fn disconnect(self) -> io::Result<()> {

        self.shutdown(Shutdown::Both)

    }

}

mod impl_noise {

    use cyphernet::encrypt::noise::{error::NoiseError as Error, NoiseState as Noise};

    use cyphernet::{Digest, Ecdh};

    use super::*;

    #[derive(Copy, Clone, Eq, PartialEq, Hash, Debug)]

    pub struct NoiseArtifact<E: Ecdh, D: Digest> {

        pub handshake_hash: D::Output,

        pub remote_static_key: Option<E::Pk>,

    }

    impl<E: Ecdh, D: Digest> StateMachine for Noise<E, D> {

        const NAME: &'static str = "noise";

        type Artifact = NoiseArtifact<E, D>;

        type Error = Error;

        fn next_read_len(&self) -> usize {

            self.next_read_len()

        }

        fn advance(&mut self, input: &[u8]) -> Result<Vec<u8>, Self::Error> {

            self.advance(input)

        }

        fn artifact(&self) -> Option<Self::Artifact> {

            self.get_handshake_hash().map(|hh| NoiseArtifact {

                handshake_hash: hh,

                remote_static_key: self.get_remote_static_key(),

            })

        }

    }

}

mod impl_socks5 {

    use cyphernet::addr::{Host as _, HostName, NetAddr};

    use cyphernet::proxy::socks5::{Error, Socks5};

    use super::*;

    impl StateMachine for Socks5 {

        const NAME: &'static str = "socks5";

        type Artifact = NetAddr<HostName>;

        type Error = Error;

        fn next_read_len(&self) -> usize {

            self.next_read_len()

        }

        fn advance(&mut self, input: &[u8]) -> Result<Vec<u8>, Self::Error> {

            self.advance(input)

        }

        fn artifact(&self) -> Option<Self::Artifact> {

            match self {

                Socks5::Initial(addr, false) if !addr.requires_proxy() => Some(addr.clone()),

                Socks5::Active(addr) => Some(addr.clone()),

                _ => None,

            }

        }

    }

}

use std::collections::BTreeSet;

use std::time::Duration;

use localtime::{LocalDuration, LocalTime};

/// Manages timers and triggers timeouts.

#[derive(Debug, Default)]

pub struct Timer {

    /// Timeouts are durations since the UNIX epoch.

    timeouts: BTreeSet<localtime::LocalTime>,

}

impl Timer {

    /// Create a new timer containing no timeouts.

    pub fn new() -> Self {

        Self {

            timeouts: BTreeSet::new(),

        }

    }

    /// Return the number of timeouts being tracked.

    #[cfg(test)]

    pub fn count(&self) -> usize {

        self.timeouts.len()

    }

    /// Check whether there are timeouts being tracked.

    #[cfg(test)]

    pub fn has_timeouts(&self) -> bool {

        !self.timeouts.is_empty()

    }

    /// Register a new timeout relative to a certain point in time.

    pub fn set_timeout(&mut self, timeout: Duration, after: LocalTime) {

        let time = after + LocalDuration::from_millis(timeout.as_millis());

        self.timeouts.insert(time);

    }

    /// Get the first timeout expiring right at or after certain moment of time.

    /// Returns [`None`] if there are no timeouts.

    pub fn next_expiring_from(&self, time: impl Into<LocalTime>) -> Option<Duration> {

        let time = time.into();

        let last = *self.timeouts.first()?;

        Some(if last >= time {

            Duration::from_millis(last.as_millis() - time.as_millis())

        } else {

            Duration::from_secs(0)

        })

    }

    /// Removes timeouts which expire by a certain moment of time (inclusive),

    /// returning total number of timeouts which were removed.

    pub fn remove_expired_by(&mut self, time: LocalTime) -> usize {

        // Since `split_off` returns everything *after* the given key, including the key,

        // if a timer is set for exactly the given time, it would remain in the "after"

        // set of unexpired keys. This isn't what we want, therefore we add `1` to the

        // given time value so that it is put in the "before" set that gets expired

        // and overwritten.

        let at = time + LocalDuration::from_millis(1);

        let unexpired = self.timeouts.split_off(&at);

        let fired = self.timeouts.len();

        self.timeouts = unexpired;

        fired

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn test_wake_exact() {

        let mut tm = Timer::new();

        let now = LocalTime::now();

        tm.set_timeout(Duration::from_secs(8), now);

        tm.set_timeout(Duration::from_secs(9), now);

        tm.set_timeout(Duration::from_secs(10), now);

        assert_eq!(tm.remove_expired_by(now + LocalDuration::from_secs(9)), 2);

        assert_eq!(tm.count(), 1);

    }

    #[test]

    fn test_wake() {

        let mut tm = Timer::new();

        let now = LocalTime::now();

        tm.set_timeout(Duration::from_secs(8), now);

        tm.set_timeout(Duration::from_secs(16), now);

        tm.set_timeout(Duration::from_secs(64), now);

        tm.set_timeout(Duration::from_secs(72), now);

        assert_eq!(tm.remove_expired_by(now), 0);

        assert_eq!(tm.count(), 4);

        assert_eq!(tm.remove_expired_by(now + LocalDuration::from_secs(9)), 1);

        assert_eq!(tm.count(), 3, "one timeout has expired");

        assert_eq!(tm.remove_expired_by(now + LocalDuration::from_secs(66)), 2);

        assert_eq!(tm.count(), 1, "another two timeouts have expired");

        assert_eq!(tm.remove_expired_by(now + LocalDuration::from_secs(96)), 1);

        assert!(!tm.has_timeouts(), "all timeouts have expired");

    }

    #[test]

    fn test_next() {

        let mut tm = Timer::new();

        let mut now = LocalTime::now();

        tm.set_timeout(Duration::from_secs(3), now);

        assert_eq!(tm.next_expiring_from(now), Some(Duration::from_secs(3)));

        now = now + LocalDuration::from_secs(2);

        assert_eq!(tm.next_expiring_from(now), Some(Duration::from_secs(1)));

        now = now + LocalDuration::from_secs(1);

        assert_eq!(tm.next_expiring_from(now), Some(Duration::from_secs(0)));

        now = now + LocalDuration::from_secs(1);

        assert_eq!(tm.next_expiring_from(now), Some(Duration::from_secs(0)));

        assert_eq!(tm.remove_expired_by(now), 1);

        assert_eq!(tm.count(), 0);

        assert_eq!(tm.next_expiring_from(now), None);

    }

}

pub use mio::Token;

pub const WAKER: Token = Token(0);

#[derive(Clone, Debug)]

pub struct Tokens {

    initial: usize,

    current: usize,

}

impl Tokens {

    pub fn new(initial: usize) -> Self {

        Tokens {

            initial,

            current: initial,

        }

    }

    #[inline]

    pub fn advance(&mut self) -> Token {

        let current = self.current;

        self.current = {

            let candidate = current.wrapping_add(1);

            if candidate == usize::MIN {

                // If we overflowed, reset to the initial value.

                // The range of `usize` is so large that likely

                // a few years have passed since the early tokens

                // were used.

                log::info!(target = "reactor"; "Tokens wrapped.");

                self.initial

            } else {

                candidate

            }

        };

        Token(current)

    }

}

impl Default for Tokens {

    fn default() -> Self {

        Tokens::new(1)

    }

}

use std::collections::VecDeque;

use std::fmt::{Debug, Display, Formatter};

use std::io::Write;

use std::{fmt, io};

use mio::event::{Event, Source};

use mio::{Interest, Registry, Token};

use radicle::node::Link;

use crate::reactor::session::Session;

use crate::reactor::{EventHandler, WriteAtomic};

const READ_BUFFER_SIZE: usize = u16::MAX as usize;

/// An event happening for a [`Transport`] network transport and delivered to

/// a [`ReactionHandler`].

///

/// [`ReactionHandler`]: crate::reactor::ReactionHandler

pub enum SessionEvent<S: Session> {

    Established(S::Artifact),

    Data(Vec<u8>),

    Terminated(io::Error),

}

/// A state of [`Transport`] network transport.

#[derive(Clone, Copy, Ord, PartialOrd, Eq, PartialEq, Hash, Debug)]

pub enum TransportState {

    /// The transport is initiated, but the connection has not been established yet.

    /// This happens only for outgoing connections due to the use of

    /// non-blocking calls to `connect`. The state changes once

    /// we receive the first notification on a `write` event on this resource

    /// from the reactor.

    Init,

    /// The connection is established, but the session handshake is still in

    /// progress. This happens while encryption handshake, authentication and

    /// other protocols injected into the session haven't completed yet.

    Handshake,

    /// The session is active. All handshakes have completed.

    Active,

    /// Session was terminated (for an unspecified reason, e.g. local shutdown,

    /// remote orderly shutdown, connectivity issue, dropped connections,

    /// encryption, or authentication problem etc.

    /// Reading and writing from the resource in

    /// this state will result in an error ([`io::Error`]).

    Terminated,

}

/// Transport is an adaptor around a specific [`Session`] (implementing

/// session management, including optional handshake, encoding, etc.) to be used

/// as a transport resource in a [`crate::reactor::Reactor`].

#[derive(Debug)]

pub struct Transport<S: Session> {

    state: TransportState,

    session: S,

    link_direction: Link,

    write_intent: bool,

    read_buffer: Box<[u8; READ_BUFFER_SIZE]>,

    write_buffer: VecDeque<u8>,

}

impl<S: Session + Source> Source for Transport<S> {

    fn register(

        &mut self,

        registry: &Registry,

        token: Token,

        interests: Interest,

    ) -> io::Result<()> {

        self.session.register(registry, token, interests)

    }

    fn reregister(

        &mut self,

        registry: &Registry,

        token: Token,

        interests: Interest,

    ) -> io::Result<()> {

        self.session.reregister(registry, token, interests)

    }

    fn deregister(&mut self, registry: &Registry) -> io::Result<()> {

        self.session.deregister(registry)

    }

}

impl<S: Session> Display for Transport<S> {

    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {

        match self.session.artifact() {

            None => f

                .debug_struct("Transport")

                .field("state", &self.state)

                .field("link_direction", &self.link_direction)

                .field("write_intent", &self.write_intent)

                .finish(),

            Some(id) => Display::fmt(&id, f),

        }

    }

}

impl<S: Session> Transport<S> {

    /// Constructs reactor-managed resource around an existing [`Session`].

    ///

    /// Must not be called for connections created in a non-blocking mode!

    ///

    /// # Errors

    ///

    /// If a session can be put into a non-blocking mode.

    pub fn with_session(session: S, link_direction: Link) -> io::Result<Self> {

        let state = if session.is_established() {

            // If we are disconnected, we will get instantly updated from the

            // reactor and the state will change automatically

            TransportState::Active

        } else {

            TransportState::Handshake

        };

        Ok(Self {

            state,

            session,

            link_direction,

            write_intent: true,

            read_buffer: Box::new([0u8; READ_BUFFER_SIZE]),

            write_buffer: VecDeque::new(),

        })

    }

    pub fn display(&self) -> impl Display {

        self.session.display()

    }

    fn terminate(&mut self, reason: io::Error) -> SessionEvent<S> {

        log::trace!(target: "transport", "Terminating session {self} due to {reason:?}");

        self.state = TransportState::Terminated;

        SessionEvent::Terminated(reason)

    }

    fn handle_io(&mut self, interest: Interest) -> Option<SessionEvent<S>> {

        if self.state == TransportState::Terminated {

            log::warn!(target: "transport", "Transport {self} is terminated, ignoring I/O event");

            return None;

        }

        let mut force_write_intent = false;

        if self.state == TransportState::Init {

            log::debug!(target: "transport", "Transport {self} is connected, initializing handshake");

            force_write_intent = true;

            self.state = TransportState::Handshake;

        } else if self.state == TransportState::Handshake {

            debug_assert!(!self.session.is_established());

            log::trace!(target: "transport", "Transport {self} got I/O while in handshake mode");

        }

        let resp = match interest {

            Interest::READABLE => self.handle_readable(),

            Interest::WRITABLE => self.handle_writable(),

            _ => unreachable!(),

        };

        if force_write_intent {

            self.write_intent = true;

        } else if self.state == TransportState::Handshake {

            // During handshake, after each read we need to write and then wait

            self.write_intent = interest == Interest::READABLE;

        }

        if matches!(&resp, Some(SessionEvent::Terminated(e)) if e.kind() == io::ErrorKind::ConnectionReset)

            && self.state != TransportState::Handshake

        {

            log::debug!(target: "transport", "Peer {self} has reset the connection");

            self.state = TransportState::Terminated;

            resp

        } else if self.session.is_established() && self.state == TransportState::Handshake {

            log::debug!(target: "transport", "Handshake with {self} is complete");

            // We just got connected; may need to send output

            self.write_intent = true;

            self.state = TransportState::Active;

            Some(SessionEvent::Established(

                self.session.artifact().expect("session is established"),

            ))

        } else {

            resp

        }

    }

    fn handle_writable(&mut self) -> Option<SessionEvent<S>> {

        if !self.session.is_established() {

            let _ = self.session.write(&[]);

            self.write_intent = true;

            return None;

        }

        match self.flush() {

            Ok(_) => None,

            // In this case, the write could not complete. Leave `needs_flush` set

            // to be notified when the socket is ready to write again.

            Err(err)

                if matches!(

                    err.kind(),

                    io::ErrorKind::WouldBlock

                        | io::ErrorKind::WriteZero

                        | io::ErrorKind::OutOfMemory

                        | io::ErrorKind::Interrupted

                ) =>

            {

                log::warn!(target: "transport", "Resource {} was not able to consume any data even though it has announced its write readiness", self.display());

                self.write_intent = true;

                None

            }

            Err(err) => Some(self.terminate(err)),

        }

    }

    fn handle_readable(&mut self) -> Option<SessionEvent<S>> {

        // Since `poll`, which this reactor is based on, is *level-triggered*,

        // we will be notified again if there is still data to be read on the socket.

        // Hence, there is no use in putting this socket read in a loop, as the second

        // invocation would likely block.

        match self.session.read(self.read_buffer.as_mut()) {

            Ok(0) if !self.session.is_established() => None,

            Ok(0) => Some(SessionEvent::Terminated(

                io::ErrorKind::ConnectionReset.into(),

            )),

            Ok(len) => Some(SessionEvent::Data(self.read_buffer[..len].to_vec())),

            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {

                // This should not happen, since this function is only called

                // when there's data on the socket. We leave it here in case external

                // conditions change.

                log::warn!(target: "transport",

                    "WOULD_BLOCK on resource which had read intent - probably normal thing to happen"

                );

                None