pubsubhelper

Documentation

Overview

Package 发布订阅器对象

Package 发布订阅器对象

Package pubsub 发布订阅器对象 非常适合作为简单的广播模式的中间件

pubsubhelper 满足pchelper规定的生产者和消费者接口

Index

• Variables
• func SerializeWithJSON() optparams.Option[Options]
• func SerializeWithMsgpack() optparams.Option[Options]
• func WithClientID(clientID string) optparams.Option[Options]
• func WithUUIDSnowflake() optparams.Option[Options]
• func WithUUIDSonyflake() optparams.Option[Options]
• func WithUUIDv4() optparams.Option[Options]
• type Consumer
  • func NewConsumer(cli redis.UniversalClient, opts ...optparams.Option[Options]) (*Consumer, error)
  • func (s *Consumer) Client() redis.UniversalClient
  • func (s *Consumer) Listen(topics string, opts ...optparams.Option[pchelper.ListenOptions]) error
  • func (s *Consumer) StopListening() error
• type Options
• type Producer
  • func NewProducer(cli redis.UniversalClient, opts ...optparams.Option[Options]) (*Producer, error)
  • func (p *Producer) Client() redis.UniversalClient
  • func (p *Producer) PubEvent(ctx context.Context, topic string, payload interface{}, ...) (*pchelper.Event, error)
  • func (p *Producer) Publish(ctx context.Context, topic string, payload interface{}, ...) error

Variables

var ErrNeedToPointOutTopics = errors.New("need to point out topics")

ErrNeedToPointOutTopics 需要指名发布订阅器

var ErrPubSubAlreadyListened = errors.New("pubsub already listened")

ErrPubSubAlreadyListened 发布订阅器已经被监听了

var ErrPubSubNotListeningYet = errors.New("pubsub not listening yet")

ErrPubSubNotListeningYet 发布订阅器未被监听

Functions

func SerializeWithJSON

func SerializeWithJSON() optparams.Option[Options]

SerializeWithJSON 使用JSON作为序列化反序列化的协议

func SerializeWithMsgpack

func SerializeWithMsgpack() optparams.Option[Options]

SerializeWithMsgpack 使用JSON作为序列化反序列化的协议

func WithClientID

func WithClientID(clientID string) optparams.Option[Options]

WithClientID 中间件通用设置,设置客户端id

func WithUUIDSnowflake

func WithUUIDSnowflake() optparams.Option[Options]

WithUUIDSnowflake 使用snowflake作为uuid的生成器

func WithUUIDSonyflake

func WithUUIDSonyflake() optparams.Option[Options]

WithUUIDSonyflake 使用sonyflake作为uuid的生成器

func WithUUIDv4

func WithUUIDv4() optparams.Option[Options]

WithUUIDv4 使用uuid4作为uuid的生成器

Types

type Consumer

type Consumer struct {
	*clientIdhelper.ClientIDAbc
	*pchelper.ConsumerABC
	// contains filtered or unexported fields
}

Consumer 发布订阅器消费者对象

func NewConsumer

func NewConsumer(cli redis.UniversalClient, opts ...optparams.Option[Options]) (*Consumer, error)

NewConsumer 创建一个新的发布订阅器消费者对象 @params cli redis.UniversalClient redis客户端对象 @params opts ...optparams.Option[Options] 消费者的配置

func (*Consumer) Client

func (s *Consumer) Client() redis.UniversalClient

Client 获取连接的redis客户端

func (*Consumer) Listen

func (s *Consumer) Listen(topics string, opts ...optparams.Option[pchelper.ListenOptions]) error

Listen 监听发布订阅器 @params topics string 监听的topic,复数topic用`,`隔开 @params opts ...optparams.Option[pchelper.ListenOptions] 监听时的一些配置,具体看listenoption.go说明

func (*Consumer) StopListening

func (s *Consumer) StopListening() error

StopListening 停止监听

type Options

type Options struct {
	ProducerConsumerOpts []optparams.Option[pchelper.Options]       //初始化pchelper的配置
	ClientIDOpts         []optparams.Option[clientIdhelper.Options] //初始化ClientID的配置
}

type Producer

type Producer struct {
	*pchelper.ProducerConsumerABC
	*clientIdhelper.ClientIDAbc
	// contains filtered or unexported fields
}

Producer 发布订阅器生产者对象

func NewProducer

func NewProducer(cli redis.UniversalClient, opts ...optparams.Option[Options]) (*Producer, error)

NewProducer 创建一个新的发布订阅器的生产者 @params cli redis.UniversalClient redis客户端对象 @params opts ...optparams.Option[Options] 生产者的配置

func (*Producer) Client

func (p *Producer) Client() redis.UniversalClient

Client 获取连接的redis客户端

func (*Producer) PubEvent

func (p *Producer) PubEvent(ctx context.Context, topic string, payload interface{}, opts ...optparams.Option[pchelper.PublishOptions]) (*pchelper.Event, error)

PubEvent 向发布订阅器中放入事件数据 @params ctx context.Context 请求的上下文 @params topic string 发送去的指定频道 @params payload []byte 发送的消息负载 @params opts ...optparams.Option[pchelper.PublishOptions] 无效 @returns *pchelper.Event 发送出去的消息对象

func (*Producer) Publish

func (p *Producer) Publish(ctx context.Context, topic string, payload interface{}, opts ...optparams.Option[pchelper.PublishOptions]) error

Publish 向队列中放入数据 @params ctx context.Context 请求的上下文 @params topic string 发送去的指定频道 @params payload interface{} 发送的消息负载,负载支持string,bytes,bool,number,以及可以被json或者msgpack序列化的对象 @params opts ...optparams.Option[pchelper.PublishOptions] 无效