publisher

Documentation

Index

• func NewPublisher(publishTimeout time.Duration, buffer int) *publisher
• type ClientOption
  • func WithClientName(name string) ClientOption
• type Message
• type Publisher
• type PublisherClient
  • func NewClient(addr string, opts ...ClientOption) (PublisherClient, error)
  • func NewClientWithDialer(bufDialer func(context.Context, string) (net.Conn, error), ...) (PublisherClient, error)
• type PubsubService
  • func NewPubsubService(logger *zap.Logger) *PubsubService
  • func (ps *PubsubService) Publish(ctx context.Context, req *api.PublishRequest) (*api.PublishResponse, error)
  • func (ps *PubsubService) Subscribe(req *api.SubscribeRequest, stream api.PubSubService_SubscribeServer) error
  • func (ps *PubsubService) Unsubscribe(ctx context.Context, request *api.UnsubscribeRequest) (*api.UnsubscribeResponse, error)

Functions

func NewPublisher

func NewPublisher(publishTimeout time.Duration, buffer int) *publisher

NewPublisher creates a new pub/sub publisher to broadcast messages. The duration is used as the send timeout as to not block the publisher publishing messages to other clients if one SignalingClient is slow or unresponsive. The buffer is used when creating new channels for topicSubscribers.

Types

type ClientOption

type ClientOption interface {
	// contains filtered or unexported methods
}

func WithClientName

func WithClientName(name string) ClientOption

WithClientName 返回一个设置客户端名称的选项

type Message

type Message struct {
	Topic string
	Data  []byte
}

type Publisher

type Publisher interface {
	// Len 返回当前订阅者数量。
	Len() int

	// Subscribe 添加一个新的订阅者，并返回一个频道供订阅者接收消息。
	Subscribe() chan interface{}

	// SubscribeTopic 添加一个新的订阅者，并根据指定的主题过滤消息。
	// 返回一个频道供订阅者接收消息。
	SubscribeTopic(topic topicFunc) chan interface{}

	// SubscribeTopicWithBuffer 添加一个新的订阅者，并根据指定的主题过滤消息。
	// 创建一个带有指定缓冲区大小的频道供订阅者接收消息。
	SubscribeTopicWithBuffer(topic topicFunc, buffer int) chan interface{}

	// Evict 移除指定的订阅者，使其不再接收任何消息。
	Evict(sub chan interface{})

	// Publish 向所有当前注册的订阅者发送数据。
	Publish(v interface{})

	// Close 关闭所有订阅者的频道。
	Close()
}

type PublisherClient

type PublisherClient interface {
	// Publish 将消息发布到指定的主题
	Publish(ctx context.Context, message *Message) error

	// Subscribe 订阅指定主题的消息，并提供一个处理函数来处理收到的消息
	Subscribe(ctx context.Context, topic string, handler func(*Message) error) error

	// Unsubscribe 取消订阅指定主题的消息
	Unsubscribe(ctx context.Context, topic string) error

	// Close 关闭客户端连接
	Close() error
}

PublisherClient 是用于发布和订阅消息的客户端接口

func NewClient

func NewClient(addr string, opts ...ClientOption) (PublisherClient, error)

func NewClientWithDialer

func NewClientWithDialer(bufDialer func(context.Context, string) (net.Conn, error), opts ...ClientOption) (PublisherClient, error)

NewClientWithDialer 使用 bufDialer 来创建一个 gRPC 客户端

type PubsubService

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

func NewPubsubService

func NewPubsubService(logger *zap.Logger) *PubsubService

func (*PubsubService) Publish

func (ps *PubsubService) Publish(ctx context.Context, req *api.PublishRequest) (*api.PublishResponse, error)

func (*PubsubService) Subscribe

func (ps *PubsubService) Subscribe(req *api.SubscribeRequest, stream api.PubSubService_SubscribeServer) error

func (*PubsubService) Unsubscribe

func (ps *PubsubService) Unsubscribe(ctx context.Context, request *api.UnsubscribeRequest) (*api.UnsubscribeResponse, error)