pub fn propose<T: Transport, C>(
        &mut self,
        ctx: &mut PollContext<EK, ER, T, C>,
        cb: Callback<EK::Snapshot>,
        req: RaftCmdRequest,
        mut err_resp: RaftCmdResponse,
    ) -> bool {
        if self.pending_remove {
            return false;
        }

        ctx.raft_metrics.propose.all += 1;

        let req_admin_cmd_type = if !req.has_admin_request() {
            None
        } else {
            Some(req.get_admin_request().get_cmd_type())
        };
        let is_urgent = is_request_urgent(&req);

        let policy = self.inspect(&req);
        let res = match policy {
            Ok(RequestPolicy::ReadLocal) => {
                self.read_local(ctx, req, cb);
                return false;
            }
            Ok(RequestPolicy::ReadIndex) => return self.read_index(ctx, req, err_resp, cb),
            Ok(RequestPolicy::ProposeNormal) => self.propose_normal(ctx, req),
            Ok(RequestPolicy::ProposeTransferLeader) => {
                return self.propose_transfer_leader(ctx, req, cb);
            }
            Ok(RequestPolicy::ProposeConfChange) => self.propose_conf_change(ctx, &req),
            Err(e) => Err(e),
        };

        match res {
            Err(e) => {
                cmd_resp::bind_error(&mut err_resp, e);
                cb.invoke_with_response(err_resp);
                false
            }
            Ok(Either::Right(idx)) => {
                if !cb.is_none() {
                    self.cmd_epoch_checker.attach_to_conflict_cmd(idx, cb);
                }
                false
            }
            Ok(Either::Left(idx)) => {
                if is_urgent {
                    self.last_urgent_proposal_idx = idx;
                    // Eager flush to make urgent proposal be applied on all nodes as soon as
                    // possible.
                    self.raft_group.skip_bcast_commit(false);
                }
                self.should_wake_up = true;
                let p = Proposal {
                    is_conf_change: req_admin_cmd_type == Some(AdminCmdType::ChangePeer),
                    index: idx,
                    term: self.term(),
                    cb,
                    renew_lease_time: None,
                };
                if let Some(cmd_type) = req_admin_cmd_type {
                    self.cmd_epoch_checker
                        .post_propose(cmd_type, idx, self.term());
                }
                self.post_propose(ctx, p);
                true
            }
        }
    }
/// Propose a request.
///
/// Return true means the request has been proposed successfully.
Propose a request.
Return true means the request has been proposed successfully.
Propose a request . Return true means the request has been proposed successfully .
pub fn propose<T: Transport, C>(
        &mut self,
        ctx: &mut PollContext<EK, ER, T, C>,
        cb: Callback<EK::Snapshot>,
        req: RaftCmdRequest,
        mut err_resp: RaftCmdResponse,
    ) -> bool {
        if self.pending_remove {
            return false;
        }
        ctx.raft_metrics.propose.all += 1;
        let req_admin_cmd_type = if !req.has_admin_request() {
            None
        } else {
            Some(req.get_admin_request().get_cmd_type())
        };
        let is_urgent = is_request_urgent(&req);
        let policy = self.inspect(&req);
        let res = match policy {
            Ok(RequestPolicy::ReadLocal) => {
                self.read_local(ctx, req, cb);
                return false;
            }
            Ok(RequestPolicy::ReadIndex) => return self.read_index(ctx, req, err_resp, cb),
            Ok(RequestPolicy::ProposeNormal) => self.propose_normal(ctx, req),
            Ok(RequestPolicy::ProposeTransferLeader) => {
                return self.propose_transfer_leader(ctx, req, cb);
            }
            Ok(RequestPolicy::ProposeConfChange) => self.propose_conf_change(ctx, &req),
            Err(e) => Err(e),
        };
        match res {
            Err(e) => {
                cmd_resp::bind_error(&mut err_resp, e);
                cb.invoke_with_response(err_resp);
                false
            }
            Ok(Either::Right(idx)) => {
                if !cb.is_none() {
                    self.cmd_epoch_checker.attach_to_conflict_cmd(idx, cb);
                }
                false
            }
            Ok(Either::Left(idx)) => {
                if is_urgent {
                    self.last_urgent_proposal_idx = idx;
                    // Eager flush to make urgent proposal be applied on all nodes as soon as
                    // possible.
                    self.raft_group.skip_bcast_commit(false);
                }
                self.should_wake_up = true;
                let p = Proposal {
                    is_conf_change: req_admin_cmd_type == Some(AdminCmdType::ChangePeer),
                    index: idx,
                    term: self.term(),
                    cb,
                    renew_lease_time: None,
                };
                if let Some(cmd_type) = req_admin_cmd_type {
                    self.cmd_epoch_checker
                        .post_propose(cmd_type, idx, self.term());
                }
                self.post_propose(ctx, p);
                true
            }
        }
    }
pub fn propose < T : Transport , C > ( & mut self , ctx : & mut PollContext < EK , ER , T , C > , cb : Callback < EK :: Snapshot > , req : RaftCmdRequest , mut err_resp : RaftCmdResponse , ) -> bool { if self . pending_remove { return false ; } ctx . raft_metrics . propose . all += 1 ; let req_admin_cmd_type = if ! req . has_admin_request ( ) { None } else { Some ( req . get_admin_request ( ) . get_cmd_type ( ) ) } ; let is_urgent = is_request_urgent ( & req ) ; let policy = self . inspect ( & req ) ; let res = match policy { Ok ( RequestPolicy :: ReadLocal ) => { self . read_local ( ctx , req , cb ) ; return false ; } Ok ( RequestPolicy :: ReadIndex ) => return self . read_index ( ctx , req , err_resp , cb ) , Ok ( RequestPolicy :: ProposeNormal ) => self . propose_normal ( ctx , req ) , Ok ( RequestPolicy :: ProposeTransferLeader ) => { return self . propose_transfer_leader ( ctx , req , cb ) ; } Ok ( RequestPolicy :: ProposeConfChange ) => self . propose_conf_change ( ctx , & req ) , Err ( e ) => Err ( e ) , } ; match res { Err ( e ) => { cmd_resp :: bind_error ( & mut err_resp , e ) ; cb . invoke_with_response ( err_resp ) ; false } Ok ( Either :: Right ( idx ) ) => { if ! cb . is_none ( ) { self . cmd_epoch_checker . attach_to_conflict_cmd ( idx , cb ) ; } false } Ok ( Either :: Left ( idx ) ) => { if is_urgent { self . last_urgent_proposal_idx = idx ; // Eager flush to make urgent proposal be applied on all nodes as soon as // possible. self . raft_group . skip_bcast_commit ( false ) ; } self . should_wake_up = true ; let p = Proposal { is_conf_change : req_admin_cmd_type == Some ( AdminCmdType :: ChangePeer ) , index : idx , term : self . term ( ) , cb , renew_lease_time : None , } ; if let Some ( cmd_type ) = req_admin_cmd_type { self . cmd_epoch_checker . post_propose ( cmd_type , idx , self . term ( ) ) ; } self . post_propose ( ctx , p ) ; true } } }
Propose a request.
Propose a request .
&mut self
ctx
&mut PollContext<EK, ER, T, C>
cb
Callback<EK::Snapshot>
req
RaftCmdRequest
err_resp
RaftCmdResponse