cf-mach 0.2.1

Network quality measurement CLI for latency, throughput, packet loss, and responsiveness
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

// Copyright (c) 2023-2024 Cloudflare, Inc.
// Licensed under the BSD-3-Clause license found in the LICENSE file or at https://opensource.org/licenses/BSD-3-Clause

use std::{future::Future, pin::Pin, task::Poll};

use http::Response;
use hyper::body::Incoming;
use tokio::{
    io::{AsyncRead, AsyncWrite},
    sync::oneshot,
};

/// A future that resolves to an http::Response.
pub type ResponseFuture = Pin<Box<dyn Future<Output = hyper::Result<Response<Incoming>>> + Send>>;

/// Create a oneshot channel with a custom implementation to make `Result`
/// handling more ergonomic.
pub fn oneshot_result<T>() -> (oneshot::Sender<crate::nq_core::Result<T>>, OneshotResult<T>) {
    let (tx, rx) = oneshot::channel();

    (tx, OneshotResult { inner: rx })
}

/// An abstraction over a [`oneshot::Receiver`] of a result.
pub struct OneshotResult<T> {
    inner: oneshot::Receiver<crate::nq_core::Result<T>>,
}

impl<T> Future for OneshotResult<T> {
    type Output = crate::nq_core::Result<T>;

    fn poll(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Self::Output> {
        match Pin::new(&mut self.get_mut().inner).poll(cx) {
            Poll::Ready(Ok(t)) => Poll::Ready(t),
            Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
            Poll::Pending => Poll::Pending,
        }
    }
}

/// Trait representing a generic readable and writable byte stream.
/// Mostly copied from rxtx, thanks Ivan for the great abstractions :)
pub trait ByteStream: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static {}

impl<T> ByteStream for T where T: AsyncRead + AsyncWrite + Sync + Send + Unpin + 'static {}