fn ingest(
        &mut self,
        ctx: RpcContext<'_>,
        mut req: IngestRequest,
        sink: UnarySink<IngestResponse>,
    ) {
        if !check_common_name(self.security_mgr.cert_allowed_cn(), &ctx) {
            return;
        }
        let label = "ingest";
        let timer = Instant::now_coarse();

        if self.switcher.get_mode() == SwitchMode::Normal
            && self
                .engine
                .ingest_maybe_slowdown_writes(CF_DEFAULT)
                .expect("cf")
        {
            let err = "too many sst files are ingesting";
            let mut server_is_busy_err = errorpb::ServerIsBusy::default();
            server_is_busy_err.set_reason(err.to_string());
            let mut errorpb = errorpb::Error::default();
            errorpb.set_message(err.to_string());
            errorpb.set_server_is_busy(server_is_busy_err);
            let mut resp = IngestResponse::default();
            resp.set_error(errorpb);
            ctx.spawn(sink.success(resp).map_err(|e| {
                warn!("send rpc failed"; "err" => %e);
            }));
            return;
        }
        // Make ingest command.
        let mut ingest = Request::default();
        ingest.set_cmd_type(CmdType::IngestSst);
        ingest.mut_ingest_sst().set_sst(req.take_sst());
        let mut context = req.take_context();
        let mut header = RaftRequestHeader::default();
        header.set_peer(context.take_peer());
        header.set_region_id(context.get_region_id());
        header.set_region_epoch(context.take_region_epoch());
        let mut cmd = RaftCmdRequest::default();
        cmd.set_header(header);
        cmd.mut_requests().push(ingest);

        let (cb, future) = paired_future_callback();
        if let Err(e) = self.router.send_command(cmd, Callback::Write(cb)) {
            let mut resp = IngestResponse::default();
            resp.set_error(e.into());
            ctx.spawn(sink.success(resp).map_err(|e| {
                warn!("send rpc failed"; "err" => %e);
            }));
            return;
        }

        let ctx_task = async move {
            let res = future.await.map_err(Error::from);
            let res = match res {
                Ok(mut res) => {
                    let mut resp = IngestResponse::default();
                    let mut header = res.response.take_header();
                    if header.has_error() {
                        resp.set_error(header.take_error());
                    }
                    Ok(resp)
                }
                Err(e) => Err(e),
            };
            send_rpc_response!(res, sink, label, timer);
        };
        ctx.spawn(Compat::new(ctx_task.unit_error().boxed()));
    }
/// Ingest the file by sending a raft command to raftstore.
///
/// If the ingestion fails because the region is not found or the epoch does
/// not match, the remaining files will eventually be cleaned up by
/// CleanupSSTWorker.
Ingest the file by sending a raft command to raftstore.
If the ingestion fails because the region is not found or the epoch does
not match, the remaining files will eventually be cleaned up by
CleanupSSTWorker.
Ingest the file by sending a raft command to raftstore . If the ingestion fails because the region is not found or the epoch does not match the remaining files will eventually be cleaned up by CleanupSSTWorker .
fn ingest(
        &mut self,
        ctx: RpcContext<'_>,
        mut req: IngestRequest,
        sink: UnarySink<IngestResponse>,
    ) {
        if !check_common_name(self.security_mgr.cert_allowed_cn(), &ctx) {
            return;
        }
        let label = "ingest";
        let timer = Instant::now_coarse();
        if self.switcher.get_mode() == SwitchMode::Normal
            && self
                .engine
                .ingest_maybe_slowdown_writes(CF_DEFAULT)
                .expect("cf")
        {
            let err = "too many sst files are ingesting";
            let mut server_is_busy_err = errorpb::ServerIsBusy::default();
            server_is_busy_err.set_reason(err.to_string());
            let mut errorpb = errorpb::Error::default();
            errorpb.set_message(err.to_string());
            errorpb.set_server_is_busy(server_is_busy_err);
            let mut resp = IngestResponse::default();
            resp.set_error(errorpb);
            ctx.spawn(sink.success(resp).map_err(|e| {
                warn!("send rpc failed"; "err" => %e);
            }));
            return;
        }
        // Make ingest command.
        let mut ingest = Request::default();
        ingest.set_cmd_type(CmdType::IngestSst);
        ingest.mut_ingest_sst().set_sst(req.take_sst());
        let mut context = req.take_context();
        let mut header = RaftRequestHeader::default();
        header.set_peer(context.take_peer());
        header.set_region_id(context.get_region_id());
        header.set_region_epoch(context.take_region_epoch());
        let mut cmd = RaftCmdRequest::default();
        cmd.set_header(header);
        cmd.mut_requests().push(ingest);
        let (cb, future) = paired_future_callback();
        if let Err(e) = self.router.send_command(cmd, Callback::Write(cb)) {
            let mut resp = IngestResponse::default();
            resp.set_error(e.into());
            ctx.spawn(sink.success(resp).map_err(|e| {
                warn!("send rpc failed"; "err" => %e);
            }));
            return;
        }
        let ctx_task = async move {
            let res = future.await.map_err(Error::from);
            let res = match res {
                Ok(mut res) => {
                    let mut resp = IngestResponse::default();
                    let mut header = res.response.take_header();
                    if header.has_error() {
                        resp.set_error(header.take_error());
                    }
                    Ok(resp)
                }
                Err(e) => Err(e),
            };
            send_rpc_response!(res, sink, label, timer);
        };
        ctx.spawn(Compat::new(ctx_task.unit_error().boxed()));
    }
fn ingest ( & mut self , ctx : RpcContext < ' _ > , mut req : IngestRequest , sink : UnarySink < IngestResponse > , ) { if ! check_common_name ( self . security_mgr . cert_allowed_cn ( ) , & ctx ) { return ; } let label = " " ; let timer = Instant :: now_coarse ( ) ; if self . switcher . get_mode ( ) == SwitchMode :: Normal && self . engine . ingest_maybe_slowdown_writes ( CF_DEFAULT ) . expect ( " " ) { let err = " " ; let mut server_is_busy_err = errorpb :: ServerIsBusy :: default ( ) ; server_is_busy_err . set_reason ( err . to_string ( ) ) ; let mut errorpb = errorpb :: Error :: default ( ) ; errorpb . set_message ( err . to_string ( ) ) ; errorpb . set_server_is_busy ( server_is_busy_err ) ; let mut resp = IngestResponse :: default ( ) ; resp . set_error ( errorpb ) ; ctx . spawn ( sink . success ( resp ) . map_err ( | e | { warn ! ( " " " " e ) ; } ) ) ; return ; } // Make ingest command. let mut ingest = Request :: default ( ) ; ingest . set_cmd_type ( CmdType :: IngestSst ) ; ingest . mut_ingest_sst ( ) . set_sst ( req . take_sst ( ) ) ; let mut context = req . take_context ( ) ; let mut header = RaftRequestHeader :: default ( ) ; header . set_peer ( context . take_peer ( ) ) ; header . set_region_id ( context . get_region_id ( ) ) ; header . set_region_epoch ( context . take_region_epoch ( ) ) ; let mut cmd = RaftCmdRequest :: default ( ) ; cmd . set_header ( header ) ; cmd . mut_requests ( ) . push ( ingest ) ; let ( cb , future ) = paired_future_callback ( ) ; if let Err ( e ) = self . router . send_command ( cmd , Callback :: Write ( cb ) ) { let mut resp = IngestResponse :: default ( ) ; resp . set_error ( e . into ( ) ) ; ctx . spawn ( sink . success ( resp ) . map_err ( | e | { warn ! ( " " " " e ) ; } ) ) ; return ; } let ctx_task = async move { let res = future . await . map_err ( Error :: from ) ; let res = match res { Ok ( mut res ) => { let mut resp = IngestResponse :: default ( ) ; let mut header = res . response . take_header ( ) ; if header . has_error ( ) { resp . set_error ( header . take_error ( ) ) ; } Ok ( resp ) } Err ( e ) => Err ( e ) , } ; send_rpc_response ! ( res sink label timer ) ; } ; ctx . spawn ( Compat :: new ( ctx_task . unit_error ( ) . boxed ( ) ) ) ; }
Ingest the file by sending a raft command to raftstore.
Ingest the file by sending a raft command to raftstore .
&mut self
ctx
RpcContext<'_>
req
IngestRequest
sink
UnarySink<IngestResponse>