Function Decode

internal/sse/decode.go:34–106

Decode reads one SSE stream from body until EOF, context cancellation, or a handler error. Decode closes body only when ctx is canceled so a blocked Read can unblock and observe the cancellation.

(ctx context.Context, body io.ReadCloser, handler Handler)

// handler error. Decode closes body only when ctx is canceled so a blocked Read
// can unblock and observe the cancellation.
func Decode(ctx context.Context, body io.ReadCloser, handler Handler) error {
	if ctx == nil {
		return fmt.Errorf("sse: context is required")
	}
	if readerIsNil(body) {
		return fmt.Errorf("sse: body is required")
	}
	if handler == nil {
		return fmt.Errorf("sse: handler is required")
	}

	cancelDone, cancelCloseErr := closeReaderOnCancel(ctx, body)
	defer close(cancelDone)

	scanner := bufio.NewScanner(body)
	scanner.Buffer(make([]byte, 0, 64*1024), maxLineBytes)

	event := Event{}
	dataBuffer := make([]byte, 0, 256)
	emit := func() (bool, error) {
		if event.ID == "" && event.Event == "" && len(dataBuffer) == 0 {
			return false, nil
		}
		if len(dataBuffer) > 0 {
			event.Data = append(json.RawMessage(nil), dataBuffer...)
		}
		err := handler(event)
		event = Event{}
		dataBuffer = dataBuffer[:0]
		if errors.Is(err, ErrStop) {
			return true, nil
		}
		return false, err
	}

	for scanner.Scan() {
		if err := ctx.Err(); err != nil {
			return decodeContextError(err, cancelCloseErr)
		}

		shouldEmit, err := decodeLine(scanner.Text(), &event, &dataBuffer)
		if err != nil {
			return err
		}
		if shouldEmit {
			stop, err := emit()
			if err != nil {
				return err
			}
			if stop {
				return nil
			}
		}
	}

	if err := scanner.Err(); err != nil {
		if ctxErr := ctx.Err(); ctxErr != nil {
			return decodeContextError(ctxErr, cancelCloseErr)