async fn monitor(
        port: String,
        route: &'static RpcServerFilter,
        http_codec: Arc<FusenHttpCodec>,
        handler_context: Arc<HandlerContext>,
        mut shutdown: Shutdown,
        shutdown_complete_tx: mpsc::Sender<()>,
    ) -> crate::Result<()> {
        let notify_shutdown = broadcast::channel(1).0;
        let listener = TcpListener::bind(&format!("0.0.0.0:{}", port)).await?;
        let mut builder = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new());
        builder.http2().max_concurrent_streams(None);
        builder.http1().keep_alive(true);
        let builder = Arc::new(builder);
        loop {
            let tcp_stream = tokio::select! {
                _ = shutdown.recv() => {
                    drop(notify_shutdown);
                    drop(shutdown_complete_tx);
                    return Ok(());
                },
                res = listener.accept() => res
            };
            match tcp_stream {
                Ok(stream) => {
                    let stream_handler = StreamHandler {
                        builder: builder.clone(),
                        tcp_stream: stream.0,
                        route,
                        http_codec: http_codec.clone(),
                        handler_context: handler_context.clone(),
                        shutdown: notify_shutdown.subscribe(),
                        _shutdown_complete: shutdown_complete_tx.clone(),
                    };
                    debug!("socket stream connect, addr: {:?}", stream.1);
                    tokio::spawn(stream_handler.run_http());
                }
                Err(err) => error!("tcp connect, err: {:?}", err),
            }
        }
    }