futu_grpc/server/
startup.rs

1use std::sync::Arc;
2
3use tokio::sync::watch;
4
5use futu_auth::{KeyStore, RuntimeCounters};
6use futu_server::router::RequestRouter;
7
8use crate::proto::futu_open_d_server::FutuOpenDServer;
9
10use super::{FutuGrpcService, GRPC_MAX_MESSAGE_SIZE_BYTES, GrpcPushBroadcaster};
11
12/// 构建未鉴权 gRPC 服务（测试专用）。
13#[cfg(test)]
14#[allow(dead_code)]
15pub(super) fn build_service(
16    router: Arc<RequestRouter>,
17    push_broadcaster: Arc<GrpcPushBroadcaster>,
18) -> FutuOpenDServer<FutuGrpcService> {
19    apply_grpc_message_limits(FutuOpenDServer::new(FutuGrpcService::new(
20        router,
21        push_broadcaster,
22    )))
23}
24
25/// 构建 gRPC 服务（带 KeyStore 鉴权 + 共享限额 counters）
26pub fn build_service_with_auth(
27    router: Arc<RequestRouter>,
28    push_broadcaster: Arc<GrpcPushBroadcaster>,
29    key_store: Arc<KeyStore>,
30    counters: Arc<RuntimeCounters>,
31) -> FutuOpenDServer<FutuGrpcService> {
32    apply_grpc_message_limits(FutuOpenDServer::new(FutuGrpcService::with_auth(
33        router,
34        push_broadcaster,
35        key_store,
36        counters,
37    )))
38}
39
40fn apply_grpc_message_limits(
41    service: FutuOpenDServer<FutuGrpcService>,
42) -> FutuOpenDServer<FutuGrpcService> {
43    service
44        .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE_BYTES)
45        .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE_BYTES)
46}
47
48/// 启动 gRPC 服务
49pub async fn start(
50    listen_addr: &str,
51    router: Arc<RequestRouter>,
52    push_broadcaster: Arc<GrpcPushBroadcaster>,
53) -> Result<(), Box<dyn std::error::Error>> {
54    start_with_auth(
55        listen_addr,
56        router,
57        push_broadcaster,
58        Arc::new(KeyStore::empty()),
59        Arc::new(RuntimeCounters::new()),
60    )
61    .await
62}
63
64/// 启动 gRPC 服务（带 KeyStore 鉴权 + 共享限额 counters）
65pub async fn start_with_auth(
66    listen_addr: &str,
67    router: Arc<RequestRouter>,
68    push_broadcaster: Arc<GrpcPushBroadcaster>,
69    key_store: Arc<KeyStore>,
70    counters: Arc<RuntimeCounters>,
71) -> Result<(), Box<dyn std::error::Error>> {
72    let addr = listen_addr
73        .parse()
74        .map_err(|e| format!("invalid addr: {e}"))?;
75    if !key_store.is_configured() {
76        tracing::warn!(
77            "gRPC server running WITHOUT API key auth (legacy mode); \
78             all RPCs are open. Pass --grpc-keys-file to enable scope-based auth."
79        );
80    }
81    let service = build_service_with_auth(router, push_broadcaster, key_store, counters);
82    tracing::info!(addr = %listen_addr, "gRPC 服务已启动");
83    tonic::transport::Server::builder()
84        .add_service(service)
85        .serve(addr)
86        .await
87        .map_err(|error| {
88            std::io::Error::other(futu_server::bind_hint::bind_error_message(
89                "gRPC",
90                "--grpc-port",
91                listen_addr,
92                error,
93            ))
94        })?;
95    Ok(())
96}
97
98/// 启动 gRPC 服务（带 KeyStore 鉴权 + 共享限额 counters），并支持 daemon
99/// 统一 shutdown 信号。
100pub async fn start_with_auth_until_shutdown(
101    listen_addr: &str,
102    router: Arc<RequestRouter>,
103    push_broadcaster: Arc<GrpcPushBroadcaster>,
104    key_store: Arc<KeyStore>,
105    counters: Arc<RuntimeCounters>,
106    shutdown_rx: watch::Receiver<bool>,
107) -> Result<(), Box<dyn std::error::Error>> {
108    let addr = listen_addr
109        .parse()
110        .map_err(|e| format!("invalid addr: {e}"))?;
111    if !key_store.is_configured() {
112        tracing::warn!(
113            "gRPC server running WITHOUT API key auth (legacy mode); \
114             all RPCs are open. Pass --grpc-keys-file to enable scope-based auth."
115        );
116    }
117    let service = build_service_with_auth(router, push_broadcaster, key_store, counters);
118    tracing::info!(addr = %listen_addr, "gRPC 服务已启动");
119    tonic::transport::Server::builder()
120        .add_service(service)
121        .serve_with_shutdown(addr, grpc_shutdown_requested(shutdown_rx))
122        .await
123        .map_err(|error| {
124            std::io::Error::other(futu_server::bind_hint::bind_error_message(
125                "gRPC",
126                "--grpc-port",
127                listen_addr,
128                error,
129            ))
130        })?;
131    Ok(())
132}
133
134async fn grpc_shutdown_requested(mut shutdown_rx: watch::Receiver<bool>) {
135    loop {
136        if *shutdown_rx.borrow() {
137            tracing::info!("gRPC server stopped by shutdown signal");
138            return;
139        }
140        if shutdown_rx.changed().await.is_err() {
141            tracing::info!("gRPC server stopped after shutdown sender dropped");
142            return;
143        }
144    }
145}
futu_grpc/server/startup.rs

futu_grpc/server/
startup.rs
