(r *Reader[T])
| 538 | } |
| 539 | |
| 540 | func readerToStream[T any](r *Reader[T]) *stream[T] { |
| 541 | switch r.typ { |
| 542 | case readerTypeStream: |
| 543 | return r.st |
| 544 | case readerTypeArray: |
| 545 | return r.ar.toStream() |
| 546 | default: |
| 547 | // 通过读取和发送转换其他类型 |
| 548 | s := newStream[T](5) |
| 549 | go func() { |
| 550 | defer s.closeSend() |
| 551 | for { |
| 552 | v, err := r.Recv() |
| 553 | if errors.Is(err, io.EOF) { |
| 554 | return |
| 555 | } |
| 556 | if s.send(v, err) { |
| 557 | return |
| 558 | } |
| 559 | } |
| 560 | }() |
| 561 | return s |
| 562 | } |
| 563 | } |
| 564 | |
| 565 | // Transform 将类型 T 的流转换为类型 U |
| 566 | // 返回 ErrSkip 可跳过某个值 |