reqres.go
func (r *reqResReader) recvNextFragment(initial bool) (*readableFragment, error) {
	if r.initialFragment != nil {
		fragment := r.initialFragment
		r.initialFragment = nil
		r.previousFragment = fragment
		return fragment, nil
	}

	// Wait for the appropriate message from the peer
	message := r.messageForFragment(initial)
	frame, err := r.mex.recvPeerFrameOfType(message.messageType())
	if err != nil {
		if err, ok := err.(errorMessage); ok {
			// If we received a serialized error from the other side, then we should go through
			// the normal doneReading path so stats get updated with this error.
			r.err = err.AsSystemError()
			return nil, err
		}

		return nil, r.failed(err)
	}

	// Parse the message and setup the fragment
	fragment, err := parseInboundFragment(r.mex.framePool, frame, message)
	if err != nil {
		return nil, r.failed(err)
	}

	r.previousFragment = fragment
	return fragment, nil
}
func (r *reqResReader) recvNextFragment(initial bool) (*readableFragment, error) {
	if r.initialFragment != nil {
		fragment := r.initialFragment
		r.initialFragment = nil
		r.previousFragment = fragment
		return fragment, nil
	}

	// Wait for the appropriate message from the peer
	message := r.messageForFragment(initial)
	frame, err := r.mex.recvPeerFrameOfType(message.messageType())
	if err != nil {
		if err, ok := err.(errorMessage); ok {
			// If we received a serialized error from the other side, then we should go through
			// the normal doneReading path so stats get updated with this error.
			r.err = err.AsSystemError()
			return nil, err
		}

		return nil, r.failed(err)
	}

	// Parse the message and setup the fragment
	fragment, err := parseInboundFragment(r.mex.framePool, frame, message)
	if err != nil {
		return nil, r.failed(err)
	}

	r.previousFragment = fragment
	return fragment, nil
}
// recvNextFragment receives the next fragment from the underlying message exchange.