<commit_before>package http2

import (
	"bufio"
	"encoding/binary"
	"errors"
	"fmt"
	"io"

	"github.com/nekolunar/http2/hpack"
)

type frameReader struct {
	*bufio.Reader
	bufSize      int
	maxFrameSize uint32

	*hpack.Decoder
	maxHeaderListSize uint32
	pendingHeaders    frameReaderFrom

	payloadLen uint32
	frameType  FrameType
	flags      Flags
	streamID   uint32

	lastPayload io.ReadCloser
	payload     framePayload
}

func newFrameReader(r io.Reader, bufSize int) *frameReader {
	return &frameReader{
		Reader:       bufio.NewReaderSize(r, bufSize),
		bufSize:      bufSize,
		maxFrameSize: defaultMaxFrameSize,
		Decoder:      hpack.NewDecoder(defaultHeaderTableSize),
	}
}

type frameReaderFrom interface {
	Frame
	readFrom(*frameReader) error
}

var frameCtor = map[FrameType]func() frameReaderFrom{
	FrameData:         func() frameReaderFrom { return new(DataFrame) },
	FrameHeaders:      func() frameReaderFrom { return new(HeadersFrame) },
	FramePriority:     func() frameReaderFrom { return new(PriorityFrame) },
	FrameRSTStream:    func() frameReaderFrom { return new(RSTStreamFrame) },
	FrameSettings:     func() frameReaderFrom { return new(SettingsFrame) },
	FramePushPromise:  func() frameReaderFrom { return new(PushPromiseFrame) },
	FramePing:         func() frameReaderFrom { return new(PingFrame) },
	FrameGoAway:       func() frameReaderFrom { return new(GoAwayFrame) },
	FrameWindowUpdate: func() frameReaderFrom { return new(WindowUpdateFrame) },
}

func (r *frameReader) ReadFrame() (Frame, error) {
	if r.lastPayload != nil {
		err := r.lastPayload.Close()
		r.lastPayload = nil

		if err != nil {
			return nil, err
		}
	}

	const frameHeaderLen = 9

again:
	frameHeader, err := r.Peek(frameHeaderLen)
	if err != nil {
		return nil, err
	}

	r.payloadLen = (uint32(frameHeader[0])<<16 | uint32(frameHeader[1])<<8 | uint32(frameHeader[2]))
	r.frameType = FrameType(frameHeader[3])
	r.flags = Flags(frameHeader[4])
	r.streamID = binary.BigEndian.Uint32(frameHeader[5:]) & (1<<31 - 1)

	r.Discard(frameHeaderLen)

	// An endpoint MUST send an error code of FRAME_SIZE_ERROR if a frame
	// exceeds the size defined in SETTINGS_MAX_FRAME_SIZE, exceeds any
	// limit defined for the frame type, or is too small to contain
	// mandatory frame data.  A frame size error in a frame that could alter
	// the state of the entire connection MUST be treated as a connection
	// error (Section 5.4.1); this includes any frame carrying a header
	// block (Section 4.3) (that is, HEADERS, PUSH_PROMISE, and
	// CONTINUATION), SETTINGS, and any frame with a stream identifier of 0.
	if r.payloadLen > r.maxFrameSize {
		return nil, ConnError{
			fmt.Errorf("frame length %d exceeds maximum %d", r.payloadLen, r.maxFrameSize),
			ErrCodeFrameSize,
		}
	}

	// A HEADERS frame without the END_HEADERS flag set MUST be followed
	// by a CONTINUATION frame for the same stream.  A receiver MUST
	// treat the receipt of any other type of frame or a frame on a
	// different stream as a connection error (Section 5.4.1) of type PROTOCOL_ERROR.
	if r.pendingHeaders != nil && r.frameType != FrameContinuation {
		return nil, ConnError{
			fmt.Errorf("received frame of type %s while processing headers", r.frameType),
			ErrCodeProtocol,
		}
	}

	var frame frameReaderFrom

	if r.frameType == FrameContinuation {
		// A CONTINUATION frame MUST be preceded by a HEADERS, PUSH_PROMISE or
		// CONTINUATION frame without the END_HEADERS flag set.  A recipient
		// that observes violation of this rule MUST respond with a connection
		// error (Section 5.4.1) of type PROTOCOL_ERROR.
		if r.pendingHeaders == nil {
			return nil, ConnError{
				fmt.Errorf("received %s frame but not currently processing headers", r.frameType),
				ErrCodeProtocol,
			}
		}

		frame = r.pendingHeaders
	} else {
		if ctor, exists := frameCtor[r.frameType]; exists {
			frame = ctor()
		} else {
			frame = new(UnknownFrame)
		}
	}

	if err = frame.readFrom(r); err != nil {
		// A decoding error in a header block MUST be treated as a connection error
		// (Section 5.4.1) of type COMPRESSION_ERROR.
		if _, ok := err.(hpack.DecodingError); ok {
			return nil, ConnError{err, ErrCodeCompression}
		}

		// SEE 10.5.  Denial-of-Service Considerations
		//     10.5.1.  Limits on Header Block Size
		if err == hpack.ErrHeaderFieldsTooLarge {
			return nil, ConnError{err, ErrCodeEnhanceYourCalm}
		}
	}

	if r.pendingHeaders != nil {
		goto again
	}

	return frame, nil
}

type framePayload struct {
	r *frameReader
	n int
	p uint8
}

func (p *framePayload) Read(dst []byte) (n int, err error) {
	if len(dst) == 0 {
		return 0, nil
	}
	if p.n <= 0 {
		return 0, io.EOF
	}
	if len(dst) > p.n {
		dst = dst[:p.n]
	}
	n, err = p.r.Read(dst)
	p.n -= n
	return<commit_msg>frameReader now correctly return an error<commit_after>package http2

import (
	"bufio"
	"encoding/binary"
	"errors"
	"fmt"
	"io"

	"github.com/nekolunar/http2/hpack"
)

type frameReader struct {
	*bufio.Reader
	bufSize      int
	maxFrameSize uint32

	*hpack.Decoder
	maxHeaderListSize uint32
	pendingHeaders    frameReaderFrom

	payloadLen uint32
	frameType  FrameType
	flags      Flags
	streamID   uint32

	lastPayload io.ReadCloser
	payload     framePayload
}

func newFrameReader(r io.Reader, bufSize int) *frameReader {
	return &frameReader{
		Reader:       bufio.NewReaderSize(r, bufSize),
		bufSize:      bufSize,
		maxFrameSize: defaultMaxFrameSize,
		Decoder:      hpack.NewDecoder(defaultHeaderTableSize),
	}
}

type frameReaderFrom interface {
	Frame
	readFrom(*frameReader) error
}

var frameCtor = map[FrameType]func() frameReaderFrom{
	FrameData:         func() frameReaderFrom { return new(DataFrame) },
	FrameHeaders:      func() frameReaderFrom { return new(HeadersFrame) },
	FramePriority:     func() frameReaderFrom { return new(PriorityFrame) },
	FrameRSTStream:    func() frameReaderFrom { return new(RSTStreamFrame) },
	FrameSettings:     func() frameReaderFrom { return new(SettingsFrame) },
	FramePushPromise:  func() frameReaderFrom { return new(PushPromiseFrame) },
	FramePing:         func() frameReaderFrom { return new(PingFrame) },
	FrameGoAway:       func() frameReaderFrom { return new(GoAwayFrame) },
	FrameWindowUpdate: func() frameReaderFrom { return new(WindowUpdateFrame) },
}

func (r *frameReader) ReadFrame() (Frame, error) {
	if r.lastPayload != nil {
		err := r.lastPayload.Close()
		r.lastPayload = nil

		if err != nil {
			return nil, err
		}
	}

	const frameHeaderLen = 9

again:
	frameHeader, err := r.Peek(frameHeaderLen)
	if err != nil {
		return nil, err
	}

	r.payloadLen = (uint32(frameHeader[0])<<16 | uint32(frameHeader[1])<<8 | uint32(frameHeader[2]))
	r.frameType = FrameType(frameHeader[3])
	r.flags = Flags(frameHeader[4])
	r.streamID = binary.BigEndian.Uint32(frameHeader[5:]) & (1<<31 - 1)

	r.Discard(frameHeaderLen)

	// An endpoint MUST send an error code of FRAME_SIZE_ERROR if a frame
	// exceeds the size defined in SETTINGS_MAX_FRAME_SIZE, exceeds any
	// limit defined for the frame type, or is too small to contain
	// mandatory frame data.  A frame size error in a frame that could alter
	// the state of the entire connection MUST be treated as a connection
	// error (Section 5.4.1); this includes any frame carrying a header
	// block (Section 4.3) (that is, HEADERS, PUSH_PROMISE, and
	// CONTINUATION), SETTINGS, and any frame with a stream identifier of 0.
	if r.payloadLen > r.maxFrameSize {
		return nil, ConnError{
			fmt.Errorf("frame length %d exceeds maximum %d", r.payloadLen, r.maxFrameSize),
			ErrCodeFrameSize,
		}
	}

	// A HEADERS frame without the END_HEADERS flag set MUST be followed
	// by a CONTINUATION frame for the same stream.  A receiver MUST
	// treat the receipt of any other type of frame or a frame on a
	// different stream as a connection error (Section 5.4.1) of type PROTOCOL_ERROR.
	if r.pendingHeaders != nil && r.frameType != FrameContinuation {
		return nil, ConnError{
			fmt.Errorf("received frame of type %s while processing headers", r.frameType),
			ErrCodeProtocol,
		}
	}

	var frame frameReaderFrom

	if r.frameType == FrameContinuation {
		// A CONTINUATION frame MUST be preceded by a HEADERS, PUSH_PROMISE or
		// CONTINUATION frame without the END_HEADERS flag set.  A recipient
		// that observes violation of this rule MUST respond with a connection
		// error (Section 5.4.1) of type PROTOCOL_ERROR.
		if r.pendingHeaders == nil {
			return nil, ConnError{
				fmt.Errorf("received %s frame but not currently processing headers", r.frameType),
				ErrCodeProtocol,
			}
		}

		frame = r.pendingHeaders
	} else {
		if ctor, exists := frameCtor[r.frameType]; exists {
			frame = ctor()
		} else {
			frame = new(UnknownFrame)
		}
	}

	if err = frame.readFrom(r); err != nil {
		// SEE 10.5.  Denial-of-Service Considerations
		//     10.5.1.  Limits on Header Block Size
		if err == hpack.ErrHeaderFieldsTooLarge {
			return nil, ConnError{err, ErrCodeEnhanceYourCalm}
		}

		switch err.(type) {
		case hpack.DecodingError:
			// A decoding error in a header block MUST be treated as a connection error
			// (Section 5.4.1) of type COMPRESSION_ERROR.
			return nil, ConnError{err, ErrCodeCompression}
		default:
			return nil, err
		}
	}

	if r.pendingHeaders != nil {
		goto again
	}

	return frame, nil
}

type framePayload struct {
	r *frameReader
	n int
	p uint8
}

func (p *framePayload) Read(dst []byte) (n int, err error) {
	if len(dst) == 0 {
		return 0, nil
	}
	if p.n <= 0 {
		return 0, io.EOF
	}
	if len(dst) > p.n {
		dst = dst[:p.n]
	}
	n, err = p.r.Read(dst)
	p.n -= n
	return
{'fraction_non_alphanumeric': 0.10213255279113527, 'fraction_numerical': 0.012335354380096175, 'mean_word_length': 3.6106024096385543, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2661684', 'n_tokens_mistral': 3424, 'n_tokens_neox': 3149, 'n_words': 982}