streams

Documentation

Index

• Variables
• type BufferedStream
  • func NewBufferedStream() *BufferedStream
  • func (s *BufferedStream) ConnectionEvents() <-chan ConnectionEvent
  • func (s *BufferedStream) Join(ctx context.Context, conn Connection) error
  • func (s *BufferedStream) Start(_ context.Context) error
  • func (s *BufferedStream) Stop(ctx context.Context) error
• type Connection
• type ConnectionEvent
• type ConnectionEventType
• type ConnectionWithLog
  • func (conn ConnectionWithLog) Close(ctx context.Context) error
  • func (conn ConnectionWithLog) Receive(ctx context.Context) ([]byte, error)
  • func (conn ConnectionWithLog) Send(ctx context.Context, data []byte) error
• type GRPCStreamClientConnection
  • func NewGRPCStreamClientConnection(name string, client streamv1grpc.Streams_ConnectStreamClient) *GRPCStreamClientConnection
  • func (conn *GRPCStreamClientConnection) Close(_ context.Context) error
  • func (conn *GRPCStreamClientConnection) Name() string
  • func (conn *GRPCStreamClientConnection) Receive(_ context.Context) ([]byte, error)
  • func (conn *GRPCStreamClientConnection) Send(_ context.Context, data []byte) error
• type GRPCStreamServerConnection
  • func NewGRPCStreamServerConnection(name string, server streamv1grpc.Streams_ConnectStreamServer) *GRPCStreamServerConnection
  • func (conn *GRPCStreamServerConnection) Close(_ context.Context) error
  • func (conn *GRPCStreamServerConnection) Done() <-chan struct{}
  • func (conn *GRPCStreamServerConnection) Name() string
  • func (conn *GRPCStreamServerConnection) Receive(_ context.Context) ([]byte, error)
  • func (conn *GRPCStreamServerConnection) Send(_ context.Context, data []byte) error
• type InMemoryManager
  • func NewInMemoryManager() *InMemoryManager
  • func (mgr *InMemoryManager) CreateStream(ctx context.Context, ins *StreamInstance) (*StreamInstance, error)
  • func (mgr *InMemoryManager) DeleteStream(ctx context.Context, uid metav1.UID) error
  • func (mgr *InMemoryManager) GetStream(_ context.Context, uid metav1.UID) (*StreamInstance, error)
  • func (mgr *InMemoryManager) ListStreams(_ context.Context) ([]*StreamInstance, error)
• type Manager
• type Stream
• type StreamInstance
  • func (ins *StreamInstance) Clone() *StreamInstance
• type WebSocketConnection
  • func NewWebSocketConnection(name string, conn *websocket.Conn) *WebSocketConnection
  • func (conn *WebSocketConnection) Close(_ context.Context) error
  • func (conn *WebSocketConnection) Name() string
  • func (conn *WebSocketConnection) Receive(_ context.Context) ([]byte, error)
  • func (conn *WebSocketConnection) Send(_ context.Context, data []byte) error

Variables

var (
	// ErrStreamAlreadyStarted 流已经开始了
	ErrStreamAlreadyStarted = errors.New("StreamAlreadyStarted")
	// ErrStreamNotFound 未找到流
	ErrStreamNotFound = errors.New("StreamNotFound")
	// ErrStreamIsFull 流满员了
	ErrStreamIsFull = errors.New("StreamIsFull")
	// ErrStreamAlreadyStopped 流已经停止了
	ErrStreamAlreadyStopped = errors.New("StreamAlreadyStopped")
	// ErrConnectionClosed 连接已关闭
	ErrConnectionClosed = errors.New("ConnectionClosed")
)

Types

type BufferedStream

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

BufferedStream 带缓冲的流

func NewBufferedStream

func NewBufferedStream() *BufferedStream

NewBufferedStream 创建 BufferedStream

func (*BufferedStream) ConnectionEvents

func (s *BufferedStream) ConnectionEvents() <-chan ConnectionEvent

ConnectionEvents 获取连接事件通道

func (*BufferedStream) Join

func (s *BufferedStream) Join(ctx context.Context, conn Connection) error

Join 将连接加入流

func (*BufferedStream) Start

func (s *BufferedStream) Start(_ context.Context) error

Start 开始传输

func (*BufferedStream) Stop

func (s *BufferedStream) Stop(ctx context.Context) error

Stop 停止传输

type Connection

type Connection interface {
	// Name 返回连接名
	Name() string
	// Send 发送
	Send(ctx context.Context, data []byte) error
	// Receive 接收
	Receive(ctx context.Context) ([]byte, error)
	// Close 关闭连接
	Close(ctx context.Context) error
}

Connection 连接

type ConnectionEvent

type ConnectionEvent struct {
	// 事件类型
	Type ConnectionEventType
	// 连接
	Connection Connection
}

ConnectionEvent 连接事件

type ConnectionEventType

type ConnectionEventType string

ConnectionEventType 连接事件类型

const (
	// JoinedEvent 连接已加入事件
	JoinedEvent ConnectionEventType = "Joined"
	// LeftEvent 连接已离开事件
	LeftEvent ConnectionEventType = "Left"
)

type ConnectionWithLog

type ConnectionWithLog struct {
	Connection
}

ConnectionWithLog 带日志的连接

func (ConnectionWithLog) Close

func (conn ConnectionWithLog) Close(ctx context.Context) error

Close 关闭连接

func (ConnectionWithLog) Receive

func (conn ConnectionWithLog) Receive(ctx context.Context) ([]byte, error)

Receive 接收

func (ConnectionWithLog) Send

func (conn ConnectionWithLog) Send(ctx context.Context, data []byte) error

Send 发送

type GRPCStreamClientConnection

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

GRPCStreamClientConnection 是 Connection 的基于 gRPC 流客户端的实现

func NewGRPCStreamClientConnection

func NewGRPCStreamClientConnection(
	name string,
	client streamv1grpc.Streams_ConnectStreamClient,
) *GRPCStreamClientConnection

NewGRPCStreamClientConnection 创建 GRPCStreamClientConnection

func (*GRPCStreamClientConnection) Close

func (conn *GRPCStreamClientConnection) Close(_ context.Context) error

Close 关闭连接

func (*GRPCStreamClientConnection) Name

func (conn *GRPCStreamClientConnection) Name() string

Name 返回连接名

func (*GRPCStreamClientConnection) Receive

func (conn *GRPCStreamClientConnection) Receive(_ context.Context) ([]byte, error)

Receive 接收

func (*GRPCStreamClientConnection) Send

func (conn *GRPCStreamClientConnection) Send(_ context.Context, data []byte) error

Send 发送

type GRPCStreamServerConnection

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

GRPCStreamServerConnection 是 Connection 的基于 gRPC 流服务端的实现

func NewGRPCStreamServerConnection

func NewGRPCStreamServerConnection(
	name string,
	server streamv1grpc.Streams_ConnectStreamServer,
) *GRPCStreamServerConnection

NewGRPCStreamServerConnection 创建 GRPCStreamServerConnection

func (*GRPCStreamServerConnection) Close

func (conn *GRPCStreamServerConnection) Close(_ context.Context) error

Close 关闭连接

func (*GRPCStreamServerConnection) Done

func (conn *GRPCStreamServerConnection) Done() <-chan struct{}

Done 返回完成 channel

func (*GRPCStreamServerConnection) Name

func (conn *GRPCStreamServerConnection) Name() string

Name 返回连接名

func (*GRPCStreamServerConnection) Receive

func (conn *GRPCStreamServerConnection) Receive(_ context.Context) ([]byte, error)

Receive 接收

func (*GRPCStreamServerConnection) Send

func (conn *GRPCStreamServerConnection) Send(_ context.Context, data []byte) error

Send 发送

type InMemoryManager

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

InMemoryManager 是 Manager 的基于内存的实现

func NewInMemoryManager

func NewInMemoryManager() *InMemoryManager

NewInMemoryManager 创建 InMemoryManager

func (*InMemoryManager) CreateStream

func (mgr *InMemoryManager) CreateStream(ctx context.Context, ins *StreamInstance) (*StreamInstance, error)

CreateStream 创建并启动流

func (*InMemoryManager) DeleteStream

func (mgr *InMemoryManager) DeleteStream(ctx context.Context, uid metav1.UID) error

DeleteStream 停止并删除流

func (*InMemoryManager) GetStream

func (mgr *InMemoryManager) GetStream(_ context.Context, uid metav1.UID) (*StreamInstance, error)

GetStream 获取流

func (*InMemoryManager) ListStreams

func (mgr *InMemoryManager) ListStreams(_ context.Context) ([]*StreamInstance, error)

ListStreams 列出流

type Manager

type Manager interface {
	// CreateStream 创建流
	CreateStream(ctx context.Context, stream *StreamInstance) (*StreamInstance, error)
	// ListStreams 列出流
	ListStreams(ctx context.Context) ([]*StreamInstance, error)
	// GetStream 获取流
	GetStream(ctx context.Context, uid metav1.UID) (*StreamInstance, error)
	// DeleteStream 删除流
	DeleteStream(ctx context.Context, uid metav1.UID) error
}

Manager 流管理器

type Stream

type Stream interface {
	// Start 开始传输
	Start(ctx context.Context) error
	// Join 将连接加入流
	Join(ctx context.Context, conn Connection) error
	// Stop 停止传输
	Stop(ctx context.Context) error
	// ConnectionEvents 获取连接事件通道
	ConnectionEvents() <-chan ConnectionEvent
}

Stream 流

type StreamInstance

type StreamInstance struct {
	Object streamv1.Stream
	Stream Stream
}

StreamInstance 流实例

func (*StreamInstance) Clone

func (ins *StreamInstance) Clone() *StreamInstance

Clone 返回流实例的一个拷贝

type WebSocketConnection

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

WebSocketConnection 是 Connection 的基于 WebSocket 的实现

func NewWebSocketConnection

func NewWebSocketConnection(name string, conn *websocket.Conn) *WebSocketConnection

NewWebSocketConnection 创建 WebSocketConnection

func (*WebSocketConnection) Close

func (conn *WebSocketConnection) Close(_ context.Context) error

Close 关闭连接

func (*WebSocketConnection) Name

func (conn *WebSocketConnection) Name() string

Name 返回连接名

func (*WebSocketConnection) Receive

func (conn *WebSocketConnection) Receive(_ context.Context) ([]byte, error)

Receive 接收

func (*WebSocketConnection) Send

func (conn *WebSocketConnection) Send(_ context.Context, data []byte) error

Send 发送