StreamReader 是对外暴露的接口，内部有四种具体实现，每种针对一种常见需求：

// schema/stream.go 中的四种 reader 类型
type stream struct { // 基础类型：底层是 channel
items chan streamItem
closed chan struct{}
}
type arrayReader struct { // 数组转流：把 T 包成流
data T
idx int
}
type multiStream struct { // fan‑in：合并多个 StreamReader
readers *StreamReader
}
type withConvert struct { // 类型转换：T → O。
可过滤
src *StreamReader
fn func
}
type cpStreamReader struct { // Copy 产生的子 reader
parent *parentCopy
cur *cpStreamElement
}

外部代码只会看到 *StreamReader;实际实例由工厂函数透明创建。

再看Pipe，创建一对读写端

// 创建一个容量为 cap 的流
reader。writer := schema.Pipe
// 写端
go func {
writer.Send
writer.Send
writer.Close
}
// 读端示例
for {
chunk,err := reader.Recv
if err == io.EOF {
break
}
if err!说起来，= nil {
// 处理错误…break
}
fmt.Print
}
reader.Close // 必须显式关闭

Pipe 内部结构：

type stream struct {
items chan streamItem // 带缓冲的数据通道。容量 = cap 参数
closed chan struct{} // 关闭信号
}
type streamItem struct {
val T
err error // 数据和错误复用同一个通道元素
}

Pain Point 对应：

• #一次性消费：Recv 一次即消费掉，不能回放。
• #显式所有权：If you forget to call .Close,underlying goroutine 会一直阻塞写端。

再看Copy，fan‑out 的懒读链表实现

// 把一个 reader 分裂成 n 个独立的 reader
readers := original.Copy // readers 为 *StreamReader

The implementation uses a **lazy linked list** rar than duplicating channel:

// 链表节点结构体
type cpStre amElement struct {
val T
err error
next *cpStre amElement
once sync.Once // 保证只从原 stream 读取一次
}