refactor(server): simplify ServerConfig and enhance InternalRouter structure

Modified app/src/event.rs +2 −8 View app/src/event.rs

@@ -86,14 +86,8 @@

8686 let cancel = CancellationToken::new();

8787 let (shutdown_tx, shutdown_rx) = watch::channel(false);

8888

89 let server_config = internal::ServerConfig::new(

90 cfg.rpc.listen_address.clone(),

91 cfg.rpc.rpc_key.clone(),

92 vec![

93 "internal.v1.EventIngressService",

94 "internal.v1.SubscriptionService",

95 ],

96 );

89 let server_config =

90 internal::ServerConfig::new(cfg.rpc.listen_address.clone(), cfg.rpc.rpc_key.clone());

9791 let router = internal::server::router(&server_config)

9892 .map_err(|error| anyhow!("building the internal rpc server: {error}"))?;

9993 let gate = server_config.gate();

Modified app/src/main.rs +1 −11 View app/src/main.rs

@@ -384,17 +384,7 @@

384384 "internal.rpc.rpc_key must be configured when internal.rpc.listen_address is set"

385385 ));

386386 }

387 let server_config = internal::ServerConfig::new(

388 listen,

389 cfg.internal.rpc.rpc_key.clone(),

390 vec![

391 "internal.v1.IdentityService",

392 "internal.v1.ChatAuthorizationService",

393 "internal.v1.IssueAuthorizationService",

394 "internal.v1.PullAuthorizationService",

395 "internal.v1.SpaceAuthorizationService",

396 ],

397 );

387 let server_config = internal::ServerConfig::new(listen, cfg.internal.rpc.rpc_key.clone());

398388 let router = internal::server::router(&server_config).map_err(|error| anyhow!("{error}"))?;

399389 let gate = server_config.gate();

400390 let router = auth::rpc::register(router, &deps.auth, gate.clone());

Modified lib/infra/internal/src/server.rs +50 −16 View lib/infra/internal/src/server.rs

@@ -1,7 +1,13 @@

1use std::convert::Infallible;

12use std::net::SocketAddr;

23

4use axum::response::IntoResponse;

5use http::Request;

6use tonic::body::Body;

7use tonic::server::NamedService;

38use tonic::transport::Server;

49use tonic::transport::server::Router;

10use tower::Service;

511use tower::layer::util::{Identity, Stack};

612

713use crate::error::InternalError;

@@ -11,27 +17,50 @@

1117

1218type Layer = Stack<oltp::GrpcRequestLog, Identity>;

1319

14/// 统一装配后的内部 RPC 路由：health + reflection + 日志层已就绪，

15/// 业务 crate 只需继续 `add_service`。

16pub type InternalRouter = Router<Layer>;

20enum RouterState {

21 Pending(Server<Layer>),

22 Ready(Router<Layer>),

23}

24

25/// 统一装配后的内部 RPC 路由：记录业务 crate 注册的 service 名，

26/// health / reflection 在 [`serve`] 时据此生成，避免手工清单漂移。

27pub struct InternalRouter {

28 state: RouterState,

29 services: Vec<&'static str>,

30}

31

32impl InternalRouter {

33 pub fn add_service<S>(mut self, service: S) -> Self

34 where

35 S: Service<Request<Body>, Error = Infallible>

36 + NamedService

37 + Clone

38 + Send

39 + Sync

40 + 'static,

41 S::Response: IntoResponse,

42 S::Future: Send + 'static,

43 {

44 self.services.push(S::NAME);

45 self.state = match self.state {

46 RouterState::Pending(mut server) => RouterState::Ready(server.add_service(service)),

47 RouterState::Ready(router) => RouterState::Ready(router.add_service(service)),

48 };

49 self

50 }

51}

1752

1853#[derive(Debug, Clone)]

1954pub struct ServerConfig {

2055 pub listen_address: String,

2156 pub rpc_key: String,

22 pub service_names: Vec<&'static str>,

2357}

2458

2559impl ServerConfig {

26 pub fn new(

27 listen_address: impl Into<String>,

28 rpc_key: impl Into<String>,

29 service_names: Vec<&'static str>,

30 ) -> Self {

60 pub fn new(listen_address: impl Into<String>, rpc_key: impl Into<String>) -> Self {

3161 Self {

3262 listen_address: listen_address.into(),

3363 rpc_key: rpc_key.into(),

34 service_names,

3564 }

3665 }

3766

@@ -51,12 +80,10 @@

5180 "internal rpc key must not be blank".to_string(),

5281 ));

5382 }

54 let health = HealthService::new(&config.service_names).server();

55 let reflection = ReflectionService::new(&config.service_names)?.server();

56 Ok(Server::builder()

57 .layer(oltp::GrpcRequestLog)

58 .add_service(health)

59 .add_service(reflection))

83 Ok(InternalRouter {

84 state: RouterState::Pending(Server::builder().layer(oltp::GrpcRequestLog)),

85 services: Vec::new(),

86 })

6087}

6188

6289pub fn parse_address(listen: &str) -> Result<SocketAddr, InternalError> {

@@ -71,6 +98,13 @@

7198 listen: &str,

7299 shutdown: impl std::future::Future<Output = ()> + Send + 'static,

73100) -> Result<(), InternalError> {

101 let InternalRouter { state, services } = router;

102 let health = HealthService::new(&services).server();

103 let reflection = ReflectionService::new(&services)?.server();

104 let router = match state {

105 RouterState::Pending(mut server) => server.add_service(health).add_service(reflection),

106 RouterState::Ready(router) => router.add_service(health).add_service(reflection),

107 };

74108 let address = parse_address(listen)?;

75109 router

76110 .serve_with_shutdown(address, shutdown)