hrelayer 0.1.1

hrelayer是http协议转发器
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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

use crate::app_state::AppState;
use crate::configuration::{Args, Configuration};
use anyhow::Error;
use clap::Parser;
use detcd::Service;
use detcd::registrar::Registrar;
use run::{http1, http2};
use std::time::Duration;
use tokio::sync::watch;
use tokio::time::sleep;

mod app_state;
mod configuration;
mod run;
mod trace;
mod mailbox;
mod util;
mod extract;
mod backend;

#[tokio::main]
async fn main() {
    let configuration = match Configuration::build(&Args::parse()) {
        Ok(c) => c,
        Err(e) => {
            println!("{:#}", e);
            return;
        }
    };

    println!("{:?}", configuration);

    #[cfg(debug_assertions)]
    let to_console = true;

    #[cfg(not(debug_assertions))]
    let to_console = false;

    // 初始化日志
    let _guard = trace::init(&configuration, false, to_console);

    if let Err(e) = run(configuration).await {
        tracing::error!("{:#}", e);
    }

    tracing::info!("server exit!!!!");
}

async fn run(configuration: Configuration) -> Result<(), Error> {
    let (quit_tx, quit_rx) = watch::channel(false);

    // 创建app数据
    let mut app_state = AppState::build(&configuration, quit_rx).await?;

    // 注册服务自己
    let mut registrar = Registrar::<Service>::from((
        app_state.etcd_client.clone(),
        configuration.service.new_register_service(),
    ));

    registrar
        .register()
        .await
        .map_err(|e| Error::new(e).context("register service error"))?;

    // 设置状态
    app_state
        .set_id(registrar.service().key.id.unwrap())
        .set_namespace(configuration.service.namespace.clone())
        .set_running(registrar.status().registered());

    if configuration.server.is_http1() {
        http1::run(app_state.clone(), &configuration).await?;
    } else {
        http2::run(app_state.clone(), &configuration).await?;
    }

    loop {
        tokio::select! {
            _ = tokio::signal::ctrl_c() => {
                let _ = quit_tx.send(true);
                tracing::info!("waiting to exit gracefully");
                break;
            }
            status = registrar.changed() => {
                app_state.set_running(status.registered());
            }
        }
    }

    drop(registrar);
    // wait for seconds
    sleep(Duration::from_secs(5)).await;
    tracing::info!("exit ok");
    Ok(())
}