fn ingest(&self, ctx: RpcContext, mut req: IngestRequest, sink: UnarySink<IngestResponse>) {
        let label = "ingest";
        let timer = Instant::now_coarse();

        // Make ingest command.
        let mut ingest = Request::new();
        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::new();
        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::new();
        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)) {
            return send_rpc_error(ctx, sink, e);
        }

        ctx.spawn(
            future
                .map_err(Error::from)
                .then(|res| match res {
                    Ok(mut res) => {
                        let mut resp = IngestResponse::new();
                        let mut header = res.response.take_header();
                        if header.has_error() {
                            resp.set_error(header.take_error());
                        }
                        future::ok(resp)
                    }
                    Err(e) => future::err(e),
                })
                .then(move |res| send_rpc_response!(res, sink, label, timer)),
        )
    }
/// 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(&self, ctx: RpcContext, mut req: IngestRequest, sink: UnarySink<IngestResponse>) {
        let label = "ingest";
        let timer = Instant::now_coarse();
        // Make ingest command.
        let mut ingest = Request::new();
        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::new();
        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::new();
        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)) {
            return send_rpc_error(ctx, sink, e);
        }
        ctx.spawn(
            future
                .map_err(Error::from)
                .then(|res| match res {
                    Ok(mut res) => {
                        let mut resp = IngestResponse::new();
                        let mut header = res.response.take_header();
                        if header.has_error() {
                            resp.set_error(header.take_error());
                        }
                        future::ok(resp)
                    }
                    Err(e) => future::err(e),
                })
                .then(move |res| send_rpc_response!(res, sink, label, timer)),
        )
    }
fn ingest ( & self , ctx : RpcContext , mut req : IngestRequest , sink : UnarySink < IngestResponse > ) { let label = " " ; let timer = Instant :: now_coarse ( ) ; // Make ingest command. let mut ingest = Request :: new ( ) ; 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 :: new ( ) ; 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 :: new ( ) ; 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 ) ) { return send_rpc_error ( ctx , sink , e ) ; } ctx . spawn ( future . map_err ( Error :: from ) . then ( | res | match res { Ok ( mut res ) => { let mut resp = IngestResponse :: new ( ) ; let mut header = res . response . take_header ( ) ; if header . has_error ( ) { resp . set_error ( header . take_error ( ) ) ; } future :: ok ( resp ) } Err ( e ) => future :: err ( e ) , } ) . then ( move | res | send_rpc_response ! ( res sink label timer ) ) , ) }
Ingest the file by sending a raft command to raftstore.
Ingest the file by sending a raft command to raftstore .
&self
ctx
RpcContext
req
IngestRequest
sink
UnarySink<IngestResponse>