| 1 | package provider |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "bytes" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | ) |
| 11 | |
| 12 | // StreamScanner retains the distinction ScanLines normally erases: a complete |
| 13 | // SSE line versus the last unterminated fragment returned when the body closes. |
| 14 | // Only a JSON prefix cut at that boundary is a recoverable stream interruption. |
| 15 | // Malformed complete events remain protocol errors. |
| 16 | type StreamScanner struct { |
| 17 | *bufio.Scanner |
| 18 | unterminated bool |
| 19 | } |
| 20 | |
| 21 | func NewStreamScanner(r io.Reader, maxTokenSize int) *StreamScanner { |
| 22 | s := &StreamScanner{Scanner: bufio.NewScanner(r)} |
| 23 | s.Buffer(make([]byte, 0, 64*1024), maxTokenSize) |
| 24 | s.Split(func(data []byte, atEOF bool) (int, []byte, error) { |
| 25 | advance, token, err := bufio.ScanLines(data, atEOF) |
| 26 | if token != nil { |
| 27 | s.unterminated = atEOF && bytes.IndexByte(data, '\n') < 0 |
| 28 | } |
| 29 | return advance, token, err |
| 30 | }) |
| 31 | return s |
| 32 | } |
| 33 | |
| 34 | func (s *StreamScanner) DecodeError(name, payload string, err error) error { |
| 35 | decoded := StreamDecodeError(name, payload, err) |
| 36 | var syntax *json.SyntaxError |
| 37 | if s.unterminated && errors.As(err, &syntax) && syntax.Offset >= int64(len(payload)) { |
| 38 | return StreamInterrupt(fmt.Errorf("%w: %w", io.ErrUnexpectedEOF, decoded), StreamInterruptPrematureEOF) |
| 39 | } |
| 40 | return decoded |
| 41 | } |
| 42 |