Go 1.24 range-over-func로 스트리밍 API 파이프라인 리팩토링하기 — 채널·콜백·이터레이터 패턴 비교

// iter 패키지 정의 (Go 1.23+)
type Seq[V any]     func(yield func(V) bool)
type Seq2[K, V any] func(yield func(K, V) bool)

func Integers(start, end int) iter.Seq[int] {
    return func(yield func(int) bool) {
        for i := start; i < end; i++ {
            if !yield(i) { // 소비자가 break하면 false 반환
                return     // 여기서 정리 로직 실행
            }
        }
    }
}
 
for n := range Integers(1, 100) {
    if n > 10 {
        break // 안전: yield에 false가 자동 전달됨
    }
    fmt.Println(n)
}

next, stop := iter.Pull(someIterator)
defer stop() // 반드시! 누락 시 백킹 코루틴 영구 대기
 
for {
    val, ok := next()
    if !ok {
        break
    }
    // val 처리
}

// Go 1.24 strings 패키지
for line := range strings.Lines(bigText) {
    process(line)
}
 
for field := range strings.FieldsSeq(csvLine) {
    columns = append(columns, field)
}
 
// slices 패키지 (Go 1.23+)
for i, v := range slices.All(mySlice) {
    fmt.Println(i, v)
}

func StreamUsers(ctx context.Context, db *sql.DB) <-chan User {
    ch := make(chan User)
    go func() {
        defer close(ch)
        rows, err := db.QueryContext(ctx, "SELECT * FROM users")
        if err != nil {
            return // 에러를 전달할 방법이 없음 (brevity — 실제로는 별도 에러 채널 필요)
        }
        defer rows.Close()
        for rows.Next() {
            var u User
            rows.Scan(&u.ID, &u.Name, &u.Email)
            select {
            case ch <- u:
            case <-ctx.Done(): // 이게 없으면 break 시 고루틴 영구 대기
                return
            }
        }
    }()
    return ch
}

func StreamUsers(ctx context.Context, db *sql.DB) iter.Seq2[User, error] {
    return func(yield func(User, error) bool) {
        rows, err := db.QueryContext(ctx, "SELECT * FROM users")
        if err != nil {
            yield(User{}, err)
            return
        }
        defer rows.Close()
 
        for rows.Next() {
            var u User
            if err := rows.Scan(&u.ID, &u.Name, &u.Email); err != nil {
                // 스캔 오류: 해당 row를 오류로 전달하고 다음 row로 계속 진행.
                // 일부 row 손상을 허용하는 배치 처리를 가정한 것입니다.
                // 단 하나의 오류도 허용하지 않는다면 yield 후 return으로 변경하세요.
                if !yield(User{}, err) {
                    return
                }
                continue
            }
            if !yield(u, nil) {
                return // 소비자가 break하면 즉시 정리
            }
        }
        if err := rows.Err(); err != nil {
            yield(User{}, err)
        }
    }
}
 
// 소비자
for u, err := range StreamUsers(ctx, db) {
    if err != nil {
        log.Printf("오류: %v", err)
        continue
    }
    if u.ID > 1000 {
        break // 안전: rows.Close()가 defer로 호출됨. ctx 없이도 누수 없음.
    }
    process(u)
}

func Map[In, Out any](seq iter.Seq[In], fn func(In) Out) iter.Seq[Out] {
    return func(yield func(Out) bool) {
        for v := range seq {
            if !yield(fn(v)) {
                return
            }
        }
    }
}
 
func Filter[V any](seq iter.Seq[V], pred func(V) bool) iter.Seq[V] {
    return func(yield func(V) bool) {
        for v := range seq {
            if pred(v) {
                if !yield(v) {
                    return
                }
            }
        }
    }
}

type Chunk struct {
    Text string
    Done bool
}
 
func StreamAIResponse(ctx context.Context, prompt string) iter.Seq[Chunk] {
    return func(yield func(Chunk) bool) {
        resp, err := callAIAPI(ctx, prompt)
        if err != nil {
            return
        }
        defer resp.Body.Close()
 
        scanner := bufio.NewScanner(resp.Body)
        for scanner.Scan() {
            chunk, ok := parseSSELine(scanner.Text())
            if !ok {
                continue
            }
            if !yield(chunk) {
                return // 소비자가 멈추면 즉시 resp.Body.Close
            }
            if chunk.Done {
                return
            }
        }
    }
}
 
func ProcessStream(ctx context.Context, prompt string) iter.Seq[string] {
    raw := StreamAIResponse(ctx, prompt)
    nonEmpty := Filter(raw, func(c Chunk) bool { return c.Text != "" })
    return Map(nonEmpty, func(c Chunk) string { return c.Text })
}
 
func handler(w http.ResponseWriter, r *http.Request) {
    w.Header().Set("Content-Type", "text/event-stream")
 
    for text := range ProcessStream(r.Context(), r.URL.Query().Get("prompt")) {
        fmt.Fprintf(w, "data: %s\n\n", text)
        if f, ok := w.(http.Flusher); ok {
            f.Flush()
        }
    }
}

// 잘못된 코드: 같은 msg 포인터에 RecvMsg로 계속 덮어씀
// 소비자가 이전 값을 참조하고 있어도 다음 호출이 그 메모리를 덮어씀
func RecvAllBad[T any](stream grpc.ClientStream, msg *T) iter.Seq2[*T, error] {
    return func(yield func(*T, error) bool) {
        for {
            if err := stream.RecvMsg(msg); err != nil {
                if err != io.EOF {
                    yield(nil, err)
                }
                return
            }
            if !yield(msg, nil) { // 위험: 다음 RecvMsg가 이 포인터 내용을 덮어씀
                return
            }
        }
    }
}

func RecvAll[T any](recv func() (*T, error)) iter.Seq2[T, error] {
    return func(yield func(T, error) bool) {
        for {
            msg, err := recv()
            if err == io.EOF {
                return
            }
            if err != nil {
                var zero T
                yield(zero, err)
                return
            }
            if !yield(*msg, nil) { // 값 복사로 yield — 소비자가 안전하게 참조 가능
                return
            }
        }
    }
}
 
// 사용 예
for resp, err := range RecvAll(stream.Recv) {
    if err != nil {
        return fmt.Errorf("스트림 오류: %w", err)
    }
    process(resp)
}

func MergeStreams[T any](a, b iter.Seq2[T, error]) iter.Seq2[T, error] {
    return func(yield func(T, error) bool) {
        nextA, stopA := iter.Pull2(a)
        defer stopA()
        nextB, stopB := iter.Pull2(b)
        defer stopB()
 
        for {
            vA, errA, okA := nextA()
            if okA && !yield(vA, errA) {
                return
            }
            vB, errB, okB := nextB()
            if okB && !yield(vB, errB) {
                return
            }
            if !okA && !okB {
                return
            }
        }
    }
}

func ParseAccessLog(r io.Reader) iter.Seq2[LogEntry, error] {
    return func(yield func(LogEntry, error) bool) {
        scanner := bufio.NewScanner(r)
        for scanner.Scan() {
            line := strings.TrimSpace(scanner.Text())
            if line == "" {
                continue
            }
 
            // Go 1.24: strings.FieldsSeq는 이터레이터 반환 — slices.Collect로 구체화
            fields := slices.Collect(strings.FieldsSeq(line))
            if len(fields) < 4 {
                continue
            }
 
            entry, err := parseFields(fields)
            if !yield(entry, err) {
                return
            }
        }
        if err := scanner.Err(); err != nil {
            yield(LogEntry{}, err)
        }
    }
}
 
// 느린 요청만 집계
var slowRequests []LogEntry
for entry, err := range ParseAccessLog(logFile) {
    if err != nil {
        log.Println("파싱 오류:", err)
        continue
    }
    if entry.Latency > 500*time.Millisecond {
        slowRequests = append(slowRequests, entry)
    }
}

// 잘못된 코드 — 컴파일 에러 없음, 버그 있음
func BadIterator() iter.Seq[int] {
    return func(yield func(int) bool) {
        for i := 0; i < 1_000_000; i++ {
            yield(i) // 반환값 무시! 소비자가 break해도 루프가 계속 돔
        }
    }
}
 
// 올바른 코드
func GoodIterator() iter.Seq[int] {
    return func(yield func(int) bool) {
        for i := 0; i < 1_000_000; i++ {
            if !yield(i) {
                return
            }
        }
    }
}

// 잘못된 코드 — 백킹 코루틴 영구 대기
next, stop := iter.Pull(someSeq)
for {
    v, ok := next()
    if !ok { break }
    process(v)
}
 
// 올바른 코드
next, stop := iter.Pull(someSeq)
defer stop() // 선언 즉시 defer

func SafeIterator(data []string) iter.Seq2[int, error] {
    return func(yield func(int, error) bool) {
        for _, s := range data {
            n, err := strconv.Atoi(s)
            if err != nil {
                if !yield(0, err) {
                    return
                }
                continue
            }
            if !yield(n, nil) {
                return
            }
        }
    }
}

// 잘못된 코드 — data race 발생
next, stop := iter.Pull(seq)
defer stop()
for range workers {
    go func() {
        v, ok := next() // 동시 호출 금지!
        process(v)
    }()
}