tor_rtmock/

util.rs

//! Internal utilities for `tor_rtmock`

use derive_deftly::define_derive_deftly;
use futures::channel::mpsc;

define_derive_deftly! {
/// Implements `Runtime` for a struct made of multiple sub-providers
///
/// The type must be a struct containing
/// field(s) which implement `SleepProvider`, `NetProvider`, etc.
///
/// The corresponding fields must be decorated with:
///
///  * `#[deftly(mock(task))]` to indicate the field implementing `Spawn + BlockOn`
///  * `#[deftly(mock(net))]` to indicate the field implementing `NetProvider`
///  * `#[deftly(mock(sleep))]` to indicate the field implementing `SleepProvider`
///     and `CoarseTimeProvider`.
///  * `#[deftly(mock(toplevel))]` to indicate the field implementing `ToplevelBlockOn`
///     unconditionally.
///  * `#[deftly(mock(toplevel_where = "BOUND"))]` to indicate the field implementing
///    `ToplevelBlockOn` only if BOUND is satisfied.
///    For example, `#[deftly(mock(toplevel_where = "R: ToplevelBlockOn"))] runtime: R,`.
// This could perhaps be further reduced:
// ambassador might be able to remove most of the body (although does it do async well?)
    SomeMockRuntime for struct, expect items, beta_deftly:

 $(
  ${when fmeta(mock(task))}

    impl <$tgens> Spawn for $ttype {
        fn spawn_obj(&self, future: FutureObj<'static, ()>) -> Result<(), SpawnError> {
            self.$fname.spawn_obj(future)
        }
    }

    impl <$tgens> Blocking for $ttype {
        type ThreadHandle<T: Send + 'static> = <$ftype as Blocking>::ThreadHandle<T>;

        fn spawn_blocking<F, T>(&self, f: F) -> <$ftype as Blocking>::ThreadHandle<T>
        where
            F: FnOnce() -> T + Send + 'static,
            T: Send + 'static {
            self.$fname.spawn_blocking(f)
        }

        fn reenter_block_on<F>(&self, future: F) -> F::Output
        where
            F: Future,
            F::Output: Send + 'static
        {
            self.$fname.reenter_block_on(future)
        }
    }

 )
 $(
  ${when any(fmeta(mock(toplevel)), fmeta(mock(toplevel_where)))}

    impl <$tgens> ToplevelBlockOn for $ttype
    where ${fmeta(mock(toplevel_where)) as token_stream, default {}}
    {
        fn block_on<F: Future>(&self, future: F) -> F::Output {
            self.$fname.block_on(future)
        }
    }

 )
 $(
  ${when fmeta(mock(net))}

    #[async_trait]
    impl <$tgens> NetStreamProvider for $ttype {
        type Stream = <$ftype as NetStreamProvider>::Stream;
        type Listener = <$ftype as NetStreamProvider>::Listener;
        type ConnectOptions = <$ftype as NetStreamProvider>::ConnectOptions;
        type ListenOptions = <$ftype as NetStreamProvider>::ListenOptions;

        async fn connect(
            &self,
            addr: &SocketAddr,
            options: &Self::ConnectOptions,
        ) -> IoResult<Self::Stream> {
            self.$fname.connect(addr, options).await
        }
        async fn listen(
            &self,
            addr: &SocketAddr,
            options: &Self::ListenOptions,
        ) -> IoResult<Self::Listener> {
            self.$fname.listen(addr, options).await
        }
    }

    #[async_trait]
    impl <$tgens> NetStreamProvider<tor_general_addr::unix::SocketAddr> for $ttype {
        type Stream = FakeStream;
        type Listener = FakeListener<tor_general_addr::unix::SocketAddr>;
        type ConnectOptions = tor_rtcompat::UnixConnectOptions;
        type ListenOptions = tor_rtcompat::UnixListenOptions;

        async fn connect(
            &self,
            _addr: &tor_general_addr::unix::SocketAddr,
            _options: &Self::ConnectOptions,
        ) -> IoResult<Self::Stream> {
            Err(tor_general_addr::unix::NoAfUnixSocketSupport::default().into())
        }
        async fn listen(
            &self,
            _addr: &tor_general_addr::unix::SocketAddr,
            _options: &Self::ListenOptions,
        ) -> IoResult<Self::Listener> {
            Err(tor_general_addr::unix::NoAfUnixSocketSupport::default().into())
        }
    }

    impl <$tgens> TlsProvider<<$ftype as NetStreamProvider>::Stream> for $ttype {
        type Connector = <$ftype as TlsProvider<
            <$ftype as NetStreamProvider>::Stream
            >>::Connector;
        type TlsStream = <$ftype as TlsProvider<
            <$ftype as NetStreamProvider>::Stream
            >>::TlsStream;
        type Acceptor = <$ftype as TlsProvider<
            <$ftype as NetStreamProvider>::Stream
            >>::Acceptor;
        type TlsServerStream = <$ftype as TlsProvider<
            <$ftype as NetStreamProvider>::Stream
            >>::TlsServerStream;

        fn tls_connector(&self) -> Self::Connector {
            self.$fname.tls_connector()
        }
        fn tls_acceptor(&self, settings: tor_rtcompat::tls::TlsAcceptorSettings) -> std::io::Result<Self::Acceptor> {
            self.$fname.tls_acceptor(settings)
        }
        fn supports_keying_material_export(&self) -> bool {
            self.$fname.supports_keying_material_export()
        }
    }

    #[async_trait]
    impl <$tgens> UdpProvider for $ttype {
        type UdpSocket = <$ftype as UdpProvider>::UdpSocket;

        #[inline]
        async fn bind(&self, addr: &SocketAddr) -> IoResult<Self::UdpSocket> {
            self.$fname.bind(addr).await
        }
    }

 )
 $(
  ${when fmeta(mock(sleep))}

    impl <$tgens> SleepProvider for $ttype {
        type SleepFuture = <$ftype as SleepProvider>::SleepFuture;

        fn sleep(&self, dur: Duration) -> Self::SleepFuture {
            self.$fname.sleep(dur)
        }
        fn now(&self) -> Instant {
            self.$fname.now()
        }
        fn wallclock(&self) -> SystemTime {
            self.$fname.wallclock()
        }
        fn block_advance<T: Into<String>>(&self, reason: T) {
            self.$fname.block_advance(reason);
        }
        fn release_advance<T: Into<String>>(&self, reason: T) {
            self.$fname.release_advance(reason);
        }
        fn allow_one_advance(&self, dur: Duration) {
            self.$fname.allow_one_advance(dur);
        }
    }

    impl <$tgens> CoarseTimeProvider for $ttype {
        fn now_coarse(&self) -> CoarseInstant {
            self.$fname.now_coarse()
        }
    }

 )

   // TODO this wants to be assert_impl but it fails at generics
   #[allow(unused)]
   const _: fn() = || {
       fn x(_: impl Runtime) { }
       fn check_impl_runtime<$tgens>(t: $ttype) { x(t) }
   };
}

/// Prelude that must be imported to derive
/// [`SomeMockRuntime`](derive_deftly_template_SomeMockRuntime)
//
// This could have been part of the expansion of `impl_runtime!`,
// but it seems rather too exciting for a macro to import things as a side gig.
//
// Arguably this ought to be an internal crate::prelude instead.
// But crate-internal preludes are controversial within the Arti team.  -Diziet
//
// For macro visibility reasons, this must come *lexically after* the macro,
// to allow it to refer to the macro in the doc comment.
pub(crate) mod impl_runtime_prelude {
    pub(crate) use async_trait::async_trait;
    pub(crate) use derive_deftly::Deftly;
    pub(crate) use futures::Future;
    pub(crate) use futures::task::{FutureObj, Spawn, SpawnError};
    pub(crate) use std::io::Result as IoResult;
    pub(crate) use std::net::SocketAddr;
    pub(crate) use tor_rtcompat::{
        Blocking, CoarseInstant, CoarseTimeProvider, NetStreamProvider, Runtime, SleepProvider,
        TlsProvider, ToplevelBlockOn, UdpProvider, unimpl::FakeListener, unimpl::FakeStream,
    };
    pub(crate) use web_time_compat::{Duration, Instant, SystemTime, SystemTimeExt};
}

/// Wrapper for `futures::channel::mpsc::channel` that embodies the `#[allow]`
///
/// We don't care about mq tracking in this test crate.
///
/// Exactly like `tor_async_utils::mpsc_channel_no_memquota`,
/// but we can't use that here for crate hierarchy reasons.
#[allow(clippy::disallowed_methods)]
pub(crate) fn mpsc_channel<T>(buffer: usize) -> (mpsc::Sender<T>, mpsc::Receiver<T>) {
    mpsc::channel(buffer)
}