Bufio Scanner Streaming NDJSON

package main

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"io"
)

type Event struct {
	Kind string `json:"kind"`
	Data int    `json:"data"`
}

func parseNDJSON(ctx context.Context, r io.Reader) (valid int, rejected int, err error) {
	scanner := bufio.NewScanner(r)

	// 默认缓冲 64KiB，超长行会报 ErrTooLong，按业务上调
	buf := make([]byte, 0, 64*1024)
	scanner.Buffer(buf, 4*1024*1024)

	line := 0
	for scanner.Scan() {
		if err := ctx.Err(); err != nil {
			return valid, rejected, err
		}
		line++

		raw := scanner.Bytes()
		if len(raw) == 0 {
			continue
		}

		var event Event
		if jsonErr := json.Unmarshal(raw, &event); jsonErr != nil {
			rejected++ // 坏行隔离计数，不中断整条流
			continue
		}
		if event.Kind == "" {
			rejected++
			continue
		}
		valid++
	}

	return valid, rejected, scanner.Err()
}

func main() {
	stream := `{"kind":"view","data":1}
{"kind":"click","data":2}
not-json-at-all
{"kind":"purchase","data":3}
`

	valid, rejected, err := parseNDJSON(
		context.Background(),
		stringsReader(stream),
	)
	fmt.Println("valid:", valid, "rejected:", rejected, "err:", err)
}

func stringsReader(s string) io.Reader { return &stringReader{s: s} }

type stringReader struct{ s string }

func (r *stringReader) Read(p []byte) (int, error) {
	if len(r.s) == 0 {
		return 0, io.EOF
	}
	n := copy(p, r.s)
	r.s = r.s[n:]
	return n, nil
}