reader.go

package sse

import (
	"bufio"
	"bytes"
	"io"
	"strconv"
	"time"
)

// Reader reads SSE events from an underlying io.Reader.
type Reader struct {
	r *bufio.Reader
}

// NewReader initializes a new SSE reader.
// It wraps the provided io.Reader in a bufio.Reader for efficient line-by-line scanning.
func NewReader(r io.Reader) *Reader {
	reader := &Reader{}
	if br, ok := r.(*bufio.Reader); ok {
		reader.r = br
		return reader
	}

	return &Reader{
		r: bufio.NewReader(r),
	}
}

// Read blocks until a complete Event is parsed from the stream, or an error occurs.
// It handles heartbeats (comments) internally by ignoring them, ensuring the caller
// only receives actionable events.
func (r *Reader) Read() (Event, error) {
	var e Event

	for {
		// Read up to the next newline.
		// ReadBytes is mechanically safer than bufio.Scanner for large SSE payloads.
		line, err := r.r.ReadBytes('\n')
		if err != nil {
			return e, err
		}

		// The SSE spec allows for both \r\n and \n line endings.
		// Strip the trailing newline characters.
		line = bytes.TrimSuffix(line, []byte("\n"))
		line = bytes.TrimSuffix(line, []byte("\r"))

		// An empty line signals the end of the event dispatch.
		if len(line) == 0 {
			// If we received a heartbeat (comment) and hit an empty line,
			// the event will be entirely empty. We shouldn't return this to the caller.
			// We continue the loop to wait for actual data.
			if e.ID == "" && e.Name == "" && e.Data == "" && e.Retry == 0 && len(e.Extensions) == 0 {
				continue
			}

			return e, nil
		}

		// If the line starts with a colon, it is a comment (often used as a heartbeat).
		// The SSE spec dictates these lines must be ignored.
		if line[0] == ':' {
			continue
		}

		// Parse the field name and value according to the SSE spec:
		// "field: value" -> field="field", value="value" (ignoring ONE leading space)
		// "field:" -> field="field", value=""
		// "field" -> field="field", value=""
		var field, value string
		colonIdx := bytes.IndexByte(line, ':')

		if colonIdx != -1 {
			field = string(line[:colonIdx])
			valBytes := line[colonIdx+1:]

			// The spec says: if the value starts with a space, strip exactly one space.
			if len(valBytes) > 0 && valBytes[0] == ' ' {
				valBytes = valBytes[1:]
			}
			value = string(valBytes)
		} else {
			// No colon found; the entire line is the field name.
			field = string(line)
			value = ""
		}

		// Populate the event based on the field name
		switch field {
		case "id":
			e.ID = value
		case "event":
			e.Name = value
		case "retry":
			if ms, err := strconv.Atoi(value); err == nil {
				e.Retry = time.Duration(ms) * time.Millisecond
			}
		case "data":
			// If the data field appears multiple times, append with a newline.
			if e.Data != "" {
				e.Data += "\n" + value
			} else {
				e.Data = value
			}
		default:
			// Unrecognized fields are captured as extensions.
			// Lazy initialization of the map prevents allocations if extensions aren't used.
			if e.Extensions == nil {
				e.Extensions = make(map[string]string)
			}
			e.Extensions[field] = value
		}
	}
}