hermesmqd 0.3.1

hermesmq server daemon: a Raft-replicated message queue
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

use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;

use clap::Parser;
use hermesmq_core::{build_raft, serve_clients, serve_http, serve_peer, RedbStore};
use tokio::net::TcpListener;

#[derive(Parser, Debug)]
#[command(name = "hermesmqd", version, about = "Raft-replicated message queue")]
struct Cli {
    #[arg(long, env = "HERMESMQ_NODE_ID", default_value_t = 1)]
    node_id: u64,

    #[arg(long, env = "HERMESMQ_DATA_DIR", default_value = "data")]
    data_dir: PathBuf,

    #[arg(long, env = "HERMESMQ_CLIENT_ADDR", default_value = "127.0.0.1:7600")]
    client_addr: SocketAddr,

    #[arg(long, env = "HERMESMQ_PEER_ADDR", default_value = "127.0.0.1:7700")]
    peer_addr: SocketAddr,

    #[arg(long, env = "HERMESMQ_METRICS_ADDR", default_value = "127.0.0.1:9600")]
    metrics_addr: SocketAddr,

    #[arg(
        long,
        env = "HERMESMQ_METRICS_ENABLED",
        default_value_t = true,
        action = clap::ArgAction::Set
    )]
    metrics_enabled: bool,
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .init();

    let cli = Cli::parse();

    std::fs::create_dir_all(&cli.data_dir)?;
    let db_path = cli.data_dir.join("hermesmq.redb");
    let db = Arc::new(RedbStore::open(&db_path)?);

    let (raft, sm) = build_raft(cli.node_id, db)
        .await
        .map_err(|e| anyhow::anyhow!(e))?;

    let peer_listener = TcpListener::bind(cli.peer_addr).await?;
    let client_listener = TcpListener::bind(cli.client_addr).await?;
    let metrics_listener = TcpListener::bind(cli.metrics_addr).await?;
    tracing::info!(
        node_id = cli.node_id,
        peer = %cli.peer_addr,
        client = %cli.client_addr,
        metrics = %cli.metrics_addr,
        metrics_enabled = cli.metrics_enabled,
        data_dir = %cli.data_dir.display(),
        "hermesmqd listening (waiting for client bootstrap)"
    );

    tokio::spawn(serve_peer(raft.clone(), peer_listener));
    tokio::spawn(serve_clients(raft.clone(), sm.clone(), client_listener));
    tokio::spawn(serve_http(raft.clone(), sm, metrics_listener, cli.metrics_enabled));

    tokio::signal::ctrl_c().await?;
    tracing::info!("shutdown signal received; stopping raft node");
    raft.shutdown().await?;
    Ok(())
}