// decodeAppendEntriesRequest decodes AppendEntriesRequest data.
// Returns true when successful
965522
func decodeAppendEntriesRequest(
	b []byte,
	args *raft.AppendEntriesRequest,
) bool {
	if len(b) < 40 {
		return false
	}
	args.Term = binary.LittleEndian.Uint64(b[0:8])
	args.PrevLogEntry = binary.LittleEndian.Uint64(b[8:16])
	args.PrevLogTerm = binary.LittleEndian.Uint64(b[16:24])
	args.LeaderCommitIndex = binary.LittleEndian.Uint64(b[24:32])
	args.Leader = make([]byte, int(binary.LittleEndian.Uint64(b[32:40])))
	b = b[40:]
	if len(b) < len(args.Leader) {
		return false
	}
	copy(args.Leader, b[:len(args.Leader)])
	b = b[len(args.Leader):]
	if len(b) < 8 {
		return false
	}
	args.Entries = make([]*raft.Log, int(binary.LittleEndian.Uint64(b)))
	b = b[8:]
	for i := 0; i < len(args.Entries); i++ {
		if len(b) < 25 {
			return false
		}
		args.Entries[i] = &raft.Log{}
		args.Entries[i].Index = binary.LittleEndian.Uint64(b[0:8])
		args.Entries[i].Term = binary.LittleEndian.Uint64(b[8:16])
		args.Entries[i].Type = raft.LogType(b[16])
		args.Entries[i].Data = make([]byte,
			int(binary.LittleEndian.Uint64(b[17:25])))
		b = b[25:]
		if len(b) < len(args.Entries[i].Data) {
			return false
		}
		copy(args.Entries[i].Data, b[:len(args.Entries[i].Data)])
		b = b[len(args.Entries[i].Data):]
	}
	return len(b) == 0
}
32855
func NopRequestDecoder(ctx context.Context, r *http.Request) (interface{}, error) {
	return nil, nil
}
390156
func NopRequestDecoder(_ context.Context, _ *nats.Msg) (interface{}, error) {
	return nil, nil
}
524709
func newAppendEntriesRequest(term uint64, prevLogIndex uint64, prevLogTerm uint64,
	commitIndex uint64, leaderName string, entries []*LogEntry) *AppendEntriesRequest {
	pbEntries := make([]*protobuf.LogEntry, len(entries))

	for i := range entries {
		pbEntries[i] = entries[i].pb
	}

	return &AppendEntriesRequest{
		Term:         term,
		PrevLogIndex: prevLogIndex,
		PrevLogTerm:  prevLogTerm,
		CommitIndex:  commitIndex,
		LeaderName:   leaderName,
		Entries:      pbEntries,
	}
}
1282413
func (p *Peer) sendAppendEntriesRequest(req *AppendEntriesRequest) {
	tracef("peer.append.send: %s->%s [prevLog:%v length: %v]\n",
		p.server.Name(), p.Name, req.PrevLogIndex, len(req.Entries))

	resp := p.server.Transporter().SendAppendEntriesRequest(p.server, p, req)
	if resp == nil {
		p.server.DispatchEvent(newEvent(HeartbeatIntervalEventType, p, nil))
		debugln("peer.append.timeout: ", p.server.Name(), "->", p.Name)
		return
	}
	traceln("peer.append.resp: ", p.server.Name(), "<-", p.Name)

	p.setLastActivity(time.Now())
	// If successful then update the previous log index.
	p.Lock()
	if resp.Success() {
		if len(req.Entries) > 0 {
			p.prevLogIndex = req.Entries[len(req.Entries)-1].GetIndex()

			// if peer append a log entry from the current term
			// we set append to true
			if req.Entries[len(req.Entries)-1].GetTerm() == p.server.currentTerm {
				resp.append = true
			}
		}
		traceln("peer.append.resp.success: ", p.Name, "; idx =", p.prevLogIndex)
		// If it was unsuccessful then decrement the previous log index and
		// we'll try again next time.
	} else {
		if resp.Term() > p.server.Term() {
			// this happens when there is a new leader comes up that this *leader* has not
			// known yet.
			// this server can know until the new leader send a ae with higher term
			// or this server finish processing this response.
			debugln("peer.append.resp.not.update: new.leader.found")
		} else if resp.Term() == req.Term && resp.CommitIndex() >= p.prevLogIndex {
			// we may miss a response from peer
			// so maybe the peer has committed the logs we just sent
			// but we did not receive the successful reply and did not increase
			// the prevLogIndex

			// peer failed to truncate the log and sent a fail reply at this time
			// we just need to update peer's prevLog index to commitIndex

			p.prevLogIndex = resp.CommitIndex()
			debugln("peer.append.resp.update: ", p.Name, "; idx =", p.prevLogIndex)

		} else if p.prevLogIndex > 0 {
			// Decrement the previous log index down until we find a match. Don't
			// let it go below where the peer's commit index is though. That's a
			// problem.
			p.prevLogIndex--
			// if it not enough, we directly decrease to the index of the
			if p.prevLogIndex > resp.Index() {
				p.prevLogIndex = resp.Index()
			}

			debugln("peer.append.resp.decrement: ", p.Name, "; idx =", p.prevLogIndex)
		}
	}
	p.Unlock()

	// Attach the peer to resp, thus server can know where it comes from
	resp.peer = p.Name
	// Send response to server for processing.
	p.server.sendAsync(resp)
}
51627
func DecodeRequest(w http.ResponseWriter, r *http.Request, req interface{}) (err error) {
	if err = json.NewDecoder(r.Body).Decode(req); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
	}
	return
}
933712
func (req *ReferenceUpdateRequest) Decode(r io.Reader) error {
	var rc io.ReadCloser
	var ok bool
	rc, ok = r.(io.ReadCloser)
	if !ok {
		rc = ioutil.NopCloser(r)
	}

	d := &updReqDecoder{r: rc, s: pktline.NewScanner(r)}
	return d.Decode(req)
}
1198401
func (cc corsCodec) NewRequest(req *http.Request) rpc.CodecRequest {
	return corsCodecRequest{cc.corsDomains, cc.baseCodec.NewRequest(req)}
}
494738
func (c *client) GetEntries(key string) ([]string, error) {
	resp, err := c.kv.Get(c.ctx, key, clientv3.WithPrefix())
	if err != nil {
		return nil, err
	}

	entries := make([]string, len(resp.Kvs))
	for i, kv := range resp.Kvs {
		entries[i] = string(kv.Value)
	}

	return entries, nil
}
881625
func encodeGRPCConcatRequest(_ context.Context, request interface{}) (interface{}, error) {
	req := request.(addendpoint.ConcatRequest)
	return &pb.ConcatRequest{A: req.A, B: req.B}, nil
}
113153
func decodeOtherRefs(p *advRefsDecoder) decoderStateFn {
	if ok := p.nextLine(); !ok {
		return nil
	}

	if bytes.HasPrefix(p.line, shallow) {
		return decodeShallow
	}

	if len(p.line) == 0 {
		return nil
	}

	saveTo := p.data.References
	if bytes.HasSuffix(p.line, peeled) {
		p.line = bytes.TrimSuffix(p.line, peeled)
		saveTo = p.data.Peeled
	}

	ref, hash, err := readRef(p.line)
	if err != nil {
		p.error("%s", err)
		return nil
	}
	saveTo[ref] = hash

	return decodeOtherRefs
}
868156
func GobDecodeEntry(b []byte) (entry Entry, err error) {
	dec := gob.NewDecoder(bytes.NewBuffer(b))
	err = dec.Decode(&entry)
	return
}
200915
func (e *StorageEntry) DecodeJSON(out interface{}) error {
	return jsonutil.DecodeJSON(e.Value, out)
}
658056
func newAppendEntriesResponse(term uint64, success bool, index uint64, commitIndex uint64) *AppendEntriesResponse {
	pb := &protobuf.AppendEntriesResponse{
		Term:        proto.Uint64(term),
		Index:       proto.Uint64(index),
		Success:     proto.Bool(success),
		CommitIndex: proto.Uint64(commitIndex),
	}

	return &AppendEntriesResponse{
		pb: pb,
	}
}
33914
func (service *Service) DecodeRequest(req *http.Request, v interface{}) error {
	body, contentType := req.Body, req.Header.Get("Content-Type")
	defer body.Close()

	if err := service.Decoder.Decode(v, body, contentType); err != nil {
		return fmt.Errorf("failed to decode request body with content type %#v: %s", contentType, err)
	}

	return nil
}
18383
func newCodecRequest(r *http.Request, encoder rpc.Encoder, errorMapper func(error) error) rpc.CodecRequest {
	// Decode the request body and check if RPC method is valid.
	req := new(serverRequest)
	err := json.NewDecoder(r.Body).Decode(req)

	if err != nil {
		err = &Error{
			Code:    E_PARSE,
			Message: err.Error(),
			Data:    req,
		}
	} else if req.Version != Version {
		err = &Error{
			Code:    E_INVALID_REQ,
			Message: "jsonrpc must be " + Version,
			Data:    req,
		}
	}

	r.Body.Close()
	return &CodecRequest{request: req, err: err, encoder: encoder, errorMapper: errorMapper}
}
911467
func (r *RequestParams) GetEntry(key string) memstore.Entry {
	entry, _ := r.Store.GetEntry(key)
	return entry
}