stream

Documentation

Index

• type MulticastController
  • func NewMulticastController(source *schema.StreamReader, bufferSize int) *MulticastController
  • func (mc *MulticastController) Err() error
  • func (mc *MulticastController) Fork(n int) []*schema.StreamReader
  • func (mc *MulticastController) Stop()
• type StreamWriter
  • func PipeStreamReader() (*schema.StreamReader, *StreamWriter)
  • func (sw *StreamWriter) Close()
  • func (sw *StreamWriter) CloseWithError(err error)
  • func (sw *StreamWriter) Send(msg *schema.Message) error

Types

type MulticastController

type MulticastController struct {
	// contains filtered or unexported fields
}

MulticastController 多播控制器，管理源流到多个子流的分发 提供优雅关闭、错误传播、背压控制等生产级特性

func NewMulticastController

func NewMulticastController(source *schema.StreamReader, bufferSize int) *MulticastController

NewMulticastController 创建多播控制器

func (*MulticastController) Err

func (mc *MulticastController) Err() error

Err 返回多播过程中的错误

func (*MulticastController) Fork

func (mc *MulticastController) Fork(n int) []*schema.StreamReader

Fork 创建 N 个子流，返回可独立读取的 StreamReader 列表 子流通过内部缓冲实现背压隔离，慢消费者不会影响其他消费者

func (*MulticastController) Stop

func (mc *MulticastController) Stop()

Stop 停止多播，关闭所有子流 注意：Stop 不会等待转发协程结束，而是立即关闭所有子流

type StreamWriter

type StreamWriter struct {
	// contains filtered or unexported fields
}

StreamWriter 流式消息写入器 与 StreamReader 配对使用，用于向流中发送消息

func PipeStreamReader

func PipeStreamReader() (*schema.StreamReader, *StreamWriter)

PipeStreamReader 创建一个配对的 StreamReader 和 StreamWriter 类似于 io.Pipe，但用于 Message 流

func (*StreamWriter) Close

func (sw *StreamWriter) Close()

Close 关闭写入器，同时关闭关联的读取器

func (*StreamWriter) CloseWithError

func (sw *StreamWriter) CloseWithError(err error)

CloseWithError 关闭写入器并设置错误

func (*StreamWriter) Send

func (sw *StreamWriter) Send(msg *schema.Message) error

Send 向流中发送一条消息 如果流已关闭或发生错误，返回错误