func (n *NetworkTransport) handleCommand(r *bufio.Reader, dec *codec.Decoder, enc *codec.Encoder) error {
	// Get the rpc type
	rpcType, err := r.ReadByte()
	if err != nil {
		return err
	}

	// Create the RPC object
	respCh := make(chan RPCResponse, 1)
	rpc := RPC{
		RespChan: respCh,
	}

	// Decode the command
	isHeartbeat := false
	switch rpcType {
	case rpcAppendEntries:
		var req AppendEntriesRequest
		if err := dec.Decode(&req); err != nil {
			return err
		}
		rpc.Command = &req

		// Check if this is a heartbeat
		if req.Term != 0 && req.Leader != nil &&
			req.PrevLogEntry == 0 && req.PrevLogTerm == 0 &&
			len(req.Entries) == 0 && req.LeaderCommitIndex == 0 {
			isHeartbeat = true
		}

	case rpcRequestVote:
		var req RequestVoteRequest
		if err := dec.Decode(&req); err != nil {
			return err
		}
		rpc.Command = &req

	case rpcInstallSnapshot:
		var req InstallSnapshotRequest
		if err := dec.Decode(&req); err != nil {
			return err
		}
		rpc.Command = &req
		rpc.Reader = io.LimitReader(r, req.Size)

	default:
		return fmt.Errorf("unknown rpc type %d", rpcType)
	}

	// Check for heartbeat fast-path
	if isHeartbeat {
		n.heartbeatFnLock.Lock()
		fn := n.heartbeatFn
		n.heartbeatFnLock.Unlock()
		if fn != nil {
			fn(rpc)
			goto RESP
		}
	}

	// Dispatch the RPC
	select {
	case n.consumeCh <- rpc:
	case <-n.shutdownCh:
		return ErrTransportShutdown
	}

	// Wait for response
RESP:
	select {
	case resp := <-respCh:
		// Send the error first
		respErr := ""
		if resp.Error != nil {
			respErr = resp.Error.Error()
		}
		if err := enc.Encode(respErr); err != nil {
			return err
		}

		// Send the response
		if err := enc.Encode(resp.Response); err != nil {
			return err
		}
	case <-n.shutdownCh:
		return ErrTransportShutdown
	}
	return nil
}
func (n *NetworkTransport) handleCommand(r *bufio.Reader, dec *codec.Decoder, enc *codec.Encoder) error {
	
	rpcType, err := r.ReadByte()
	if err != nil {
		return err
	}

	
	respCh := make(chan RPCResponse, 1)
	rpc := RPC{
		RespChan: respCh,
	}

	
	isHeartbeat := false
	switch rpcType {
	case rpcAppendEntries:
		var req AppendEntriesRequest
		if err := dec.Decode(&req); err != nil {
			return err
		}
		rpc.Command = &req

		
		if req.Term != 0 && req.Leader != nil &&
			req.PrevLogEntry == 0 && req.PrevLogTerm == 0 &&
			len(req.Entries) == 0 && req.LeaderCommitIndex == 0 {
			isHeartbeat = true
		}

	case rpcRequestVote:
		var req RequestVoteRequest
		if err := dec.Decode(&req); err != nil {
			return err
		}
		rpc.Command = &req

	case rpcInstallSnapshot:
		var req InstallSnapshotRequest
		if err := dec.Decode(&req); err != nil {
			return err
		}
		rpc.Command = &req
		rpc.Reader = io.LimitReader(r, req.Size)

	default:
		return fmt.Errorf("unknown rpc type %d", rpcType)
	}

	
	if isHeartbeat {
		n.heartbeatFnLock.Lock()
		fn := n.heartbeatFn
		n.heartbeatFnLock.Unlock()
		if fn != nil {
			fn(rpc)
			goto RESP
		}
	}

	
	select {
	case n.consumeCh <- rpc:
	case <-n.shutdownCh:
		return ErrTransportShutdown
	}

	
RESP:
	select {
	case resp := <-respCh:
		
		respErr := ""
		if resp.Error != nil {
			respErr = resp.Error.Error()
		}
		if err := enc.Encode(respErr); err != nil {
			return err
		}

		
		if err := enc.Encode(resp.Response); err != nil {
			return err
		}
	case <-n.shutdownCh:
		return ErrTransportShutdown
	}
	return nil
}
func ( n * NetworkTransport ) handleCommand ( r * bufio . Reader , dec * codec . Decoder , enc * codec . Encoder ) error { // Get the rpc type rpcType , err := r . ReadByte ( ) if err != nil { return err } // Create the RPC object respCh := make ( chan RPCResponse , 1 ) rpc := RPC { RespChan : respCh , } // Decode the command isHeartbeat := false switch rpcType { case rpcAppendEntries : var req AppendEntriesRequest if err := dec . Decode ( & req ) ; err != nil { return err } rpc . Command = & req // Check if this is a heartbeat if req . Term != 0 && req . Leader != nil && req . PrevLogEntry == 0 && req . PrevLogTerm == 0 && len ( req . Entries ) == 0 && req . LeaderCommitIndex == 0 { isHeartbeat = true } case rpcRequestVote : var req RequestVoteRequest if err := dec . Decode ( & req ) ; err != nil { return err } rpc . Command = & req case rpcInstallSnapshot : var req InstallSnapshotRequest if err := dec . Decode ( & req ) ; err != nil { return err } rpc . Command = & req rpc . Reader = io . LimitReader ( r , req . Size ) default : return fmt . Errorf ( " " , rpcType ) } // Check for heartbeat fast-path if isHeartbeat { n . heartbeatFnLock . Lock ( ) fn := n . heartbeatFn n . heartbeatFnLock . Unlock ( ) if fn != nil { fn ( rpc ) goto RESP } } // Dispatch the RPC select { case n . consumeCh <- rpc : case <- n . shutdownCh : return ErrTransportShutdown } // Wait for response RESP : select { case resp := <- respCh : // Send the error first respErr := " " if resp . Error != nil { respErr = resp . Error . Error ( ) } if err := enc . Encode ( respErr ) ; err != nil { return err } // Send the response if err := enc . Encode ( resp . Response ) ; err != nil { return err } case <- n . shutdownCh : return ErrTransportShutdown } return nil }
// handleCommand is used to decode and dispatch a single command.
handleCommand is used to decode and dispatch a single command .