Repository: tikv/tikv
File: components/cdc/src/service.rs (Rust)

Code before:
/// Service implements the `ChangeData` service.
///
/// It's a front-end of the CDC service, schedules requests to the `Endpoint`.
#[derive(Clone)]
pub struct Service {
    scheduler: Scheduler<Task>,
    memory_quota: Arc<MemoryQuota>,
    pool: Arc<Runtime>,
}

impl Service {
    /// Create a ChangeData service.
    ///
    /// It requires a scheduler of an `Endpoint` in order to schedule tasks.
    pub fn new(scheduler: Scheduler<Task>, memory_quota: Arc<MemoryQuota>) -> Service {
        let pool = Arc::new(
            tokio::runtime::Builder::new_multi_thread()
                .worker_threads(1)
                .thread_name("cdc-watchdog")
                .enable_all()
                .build()
                .unwrap(),
        );
        
        Service {
            scheduler,
            memory_quota,
            pool,
        }
    }

    // Parse HTTP/2 headers. Only for `Self::event_feed_v2`.
    fn parse_headers(ctx: &RpcContext<'_>) -> Result<EventFeedHeaders, String> {
        let mut header = EventFeedHeaders::default();
        let metadata = ctx.request_headers();
        for i in 0..metadata.len() {
            let (key, value) = metadata.get(i).unwrap();
            if key == EventFeedHeaders::FEATURES_KEY {
                header.features = EventFeedHeaders::parse_features(value)?;
            }
        }
        Ok(header)
    }

    fn parse_version_from_request_header(
        request: &ChangeDataRequest,
        peer: &str,
    ) -> semver::Version {
        let version_field = request.get_header().get_ticdc_version();
        semver::Version::parse(version_field).unwrap_or_else(|e| {
            warn!(

Review comment (suggestion):
This seems quite similar to using `ctx.spawn`, especially since it's a multi-thread runtime with only one worker thread.

Code after:
            return Err(unknowns.join(","));
        }
        Ok(features)
    }
}

/// Service implements the `ChangeData` service.
///
/// It's a front-end of the CDC service, schedules requests to the `Endpoint`.
#[derive(Clone)]
pub struct Service {
    scheduler: Scheduler<Task>,
    memory_quota: Arc<MemoryQuota>,
    pool: Arc<Worker>,
}

impl Service {
    /// Create a ChangeData service.
    ///
    /// It requires a scheduler of an `Endpoint` in order to schedule tasks.
    pub fn new(
        scheduler: Scheduler<Task>,
        memory_quota: Arc<MemoryQuota>,
        pool: Arc<Worker>,
    ) -> Service {
        Service {
            scheduler,
            memory_quota,
            pool,
        }
    }

    // Parse HTTP/2 headers. Only for `Self::event_feed_v2`.
    fn parse_headers(ctx: &RpcContext<'_>) -> Result<EventFeedHeaders, String> {
        let mut header = EventFeedHeaders::default();
        let metadata = ctx.request_headers();
        for i in 0..metadata.len() {
            let (key, value) = metadata.get(i).unwrap();
            if key == EventFeedHeaders::FEATURES_KEY {
                header.features = EventFeedHeaders::parse_features(value)?;
            }
        }
        Ok(header)
    }

    fn parse_version_from_request_header(
        request: &ChangeDataRequest,
        peer: &str,
    ) -> semver::Version {
        let version_field = request.get_header().get_ticdc_version();
        semver::Version::parse(version_field).unwrap_or_else(|e| {