moq-dir 0.3.4

Media over QUIC
Documentation 
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
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

use anyhow::Context;
use clap::Parser;
use futures::{stream::FuturesUnordered, StreamExt};

use std::net;

use moq_native_ietf::{quic, tls};

mod listing;
mod listings;
mod session;

pub use listing::*;
pub use listings::*;
pub use session::*;

#[derive(Clone, clap::Parser)]
pub struct Cli {
    /// Listen for UDP packets on the given address.
    #[arg(long, default_value = "[::]:443")]
    pub bind: net::SocketAddr,

    /// The TLS configuration.
    #[command(flatten)]
    pub tls: tls::Args,

    /// Aggregate all announcements received with this namespace prefix.
    /// The list of announcements that match are available as tracks, ending with /.
    ///
    /// ex. ANNOUNCE namespace=public/meeting/12342/alice
    /// ex. TRACK    namespace=public/ name=meeting/12342/ payload=alice
    ///
    /// Any announcements that don't match are ignored.
    #[arg(long, default_value = ".")]
    pub namespace: String,
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    env_logger::init();

    // Disable tracing so we don't get a bunch of Quinn spam.
    let tracer = tracing_subscriber::FmtSubscriber::builder()
        .with_max_level(tracing::Level::WARN)
        .finish();
    tracing::subscriber::set_global_default(tracer).unwrap();

    let cli = Cli::parse();
    let tls = cli.tls.load()?;

    let quic = quic::Endpoint::new(quic::Config {
        bind: cli.bind,
        tls,
    })?;
    let mut quic = quic.server.context("missing server certificate")?;

    let listings = Listings::new(cli.namespace);

    let mut tasks = FuturesUnordered::new();

    log::info!("listening on {}", quic.local_addr()?);

    loop {
        tokio::select! {
            res = quic.accept() => {
                let session = res.context("failed to accept QUIC connection")?;
                let session = Session::new(session, listings.clone());

                tasks.push(async move {
                    if let Err(err) = session.run().await {
                        log::warn!("session terminated: {}", err);
                    }
                });
            },
            res = tasks.next(), if !tasks.is_empty() => res.unwrap(),
        }
    }
}