aliyun_rocketmq_go

README

aliyun-rocketmq-go

Simple packaging for Aliyun RocketMQ; For Go language

Documentation

Index

• type ConsumeFrom
• type RocketHelper
  • func NewRocketHelper(opts *RocketHelperOptions) *RocketHelper
  • func (rh *RocketHelper) CreateMessage(topic string, body []byte, tag string, keys []string, ...) *primitive.Message
  • func (rh *RocketHelper) CreatePushConsumer() (rocketmq.PushConsumer, error)
  • func (rh *RocketHelper) NewProducer() (rocketmq.Producer, error)
  • func (rh *RocketHelper) PushConsume(topic string, tagFilter string, onMessage func(*primitive.MessageExt) error) error
  • func (rh *RocketHelper) PushConsumeByConsumer(c rocketmq.PushConsumer, topic string, selector consumer.MessageSelector, ...) error
  • func (rh *RocketHelper) SendMessage(ctx context.Context, msg *primitive.Message) (*primitive.SendResult, error)
  • func (rh *RocketHelper) SendMessageByProducer(ctx context.Context, p rocketmq.Producer, msg *primitive.Message) (*primitive.SendResult, error)
• type RocketHelperOptions
  • func (opts *RocketHelperOptions) GetCredentials() primitive.Credentials

Types

type ConsumeFrom

type ConsumeFrom consumer.ConsumeFromWhere

const (
	ConsumeFromLast  ConsumeFrom = ConsumeFrom(consumer.ConsumeFromLastOffset)
	ConsumeFromFirst ConsumeFrom = ConsumeFrom(consumer.ConsumeFromFirstOffset)
)

type RocketHelper

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

func NewRocketHelper

func NewRocketHelper(opts *RocketHelperOptions) *RocketHelper

func (*RocketHelper) CreateMessage

func (rh *RocketHelper) CreateMessage(topic string, body []byte, tag string, keys []string, properties map[string]string) *primitive.Message

func (*RocketHelper) CreatePushConsumer

func (rh *RocketHelper) CreatePushConsumer() (rocketmq.PushConsumer, error)

func (*RocketHelper) NewProducer

func (rh *RocketHelper) NewProducer() (rocketmq.Producer, error)

func (*RocketHelper) PushConsume

func (rh *RocketHelper) PushConsume(topic string, tagFilter string, onMessage func(*primitive.MessageExt) error) error

func (*RocketHelper) PushConsumeByConsumer

func (rh *RocketHelper) PushConsumeByConsumer(c rocketmq.PushConsumer, topic string, selector consumer.MessageSelector, onMessage func(*primitive.MessageExt) error) error

func (*RocketHelper) SendMessage

func (rh *RocketHelper) SendMessage(ctx context.Context, msg *primitive.Message) (*primitive.SendResult, error)

func (*RocketHelper) SendMessageByProducer

func (rh *RocketHelper) SendMessageByProducer(ctx context.Context, p rocketmq.Producer, msg *primitive.Message) (*primitive.SendResult, error)

type RocketHelperOptions

type RocketHelperOptions struct {
	Endpoint        string      // Aliyun RocketMQ 服务接入点
	InstanceId      string      // Aliyun RocketMQ 服务实例标识
	GroupId         string      // 客户端 Group 标识
	ConsumeFrom     ConsumeFrom // 初次消息消费开始位置
	AccessKeyId     string
	AccessKeySecret string
}

func (*RocketHelperOptions) GetCredentials

func (opts *RocketHelperOptions) GetCredentials() primitive.Credentials