mq_http_sdk

package module
v0.0.0-...-b4f8e33 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 14, 2022 License: MIT Imports: 17 Imported by: 0

README

MQ GO HTTP SDK

Alyun MQ Documents: http://www.aliyun.com/product/ons

Aliyun MQ Console: https://ons.console.aliyun.com

Use

  1. import github.com/aliyunmq/mq-http-go-sdk
  2. setup GOPATH:
    • MAC/LINUX: export GOPATH={dir}, export GOBIN=$GOPATH/bin
    • WINDOWS: set GOPATH={dir}
  3. go get -x -v

Note

  1. Http consumer only support timer msg (less than 3 days), no matter the msg is produced from http or tcp protocol.
  2. Order is only supported at special server cluster.

Sample (github)

Publish Message

Consume Message

Transaction Message

Publish Order Message

Consume Order Message

Sample (code.aliyun.com)

Publish Message

Consume Message

Transaction Message

Publish Order Message

Consume Order Message

Documentation

Index

Constants

View Source 
const (
	ClientName           = "mq-go-sdk/1.0.3(fasthttp)"
	ClientVersion        = "2015-06-06"
	DefaultTimeout int64 = 35
)
View Source 
const (
	GET    Method = "GET"
	POST          = "POST"
	DELETE        = "DELETE"
)
View Source 
const (
	AUTHORIZATION  = "Authorization"
	CONTENT_TYPE   = "Content-Type"
	CONTENT_MD5    = "Content-MD5"
	MQ_VERSION     = "x-mq-version"
	HOST           = "Host"
	DATE           = "Date"
	KEEP_ALIVE     = "Keep-Alive"
	SECURITY_TOKEN = "security-token"
)
View Source 
const (
	ALIYUN_MQ_ERR_NS = "MQ"

	ALIYUN_MQ_ERR_TEMPSTR = "" /* 133-byte string literal not displayed */
)
View Source 
const (
	StartDeliverTime       = "__STARTDELIVERTIME"
	TransCheckImmunityTime = "__TransCheckT"
	Keys                   = "KEYS"
	SHARDING               = "__SHARDINGKEY"
)

Variables

View Source 
var (
	ErrSignMessageFailed       = errors.TN(ALIYUN_MQ_ERR_NS, 1, "sign message failed, {{.err}}")
	ErrMarshalMessageFailed    = errors.TN(ALIYUN_MQ_ERR_NS, 2, "marshal message filed, {{.err}}")
	ErrGeneralAuthHeaderFailed = errors.TN(ALIYUN_MQ_ERR_NS, 3, "general auth header failed, {{.err}}")
	ErrMessageProperty         = errors.TN(ALIYUN_MQ_ERR_NS, 4, "message property can not contains:\" ' < > & : |, {{.err}}")

	ErrSendRequestFailed = errors.TN(ALIYUN_MQ_ERR_NS, 5, "send request failed, {{.err}}")

	ErrUnmarshalErrorResponseFailed = errors.TN(ALIYUN_MQ_ERR_NS, 7, "unmarshal error response failed, {{.err}}, ResponseBody: {{.resp}}")
	ErrUnmarshalResponseFailed      = errors.TN(ALIYUN_MQ_ERR_NS, 8, "unmarshal response failed, {{.err}}")

	ErrMqServer = errors.TN(ALIYUN_MQ_ERR_NS, 101, ALIYUN_MQ_ERR_TEMPSTR)

	ErrAckMessage = errors.TN(ALIYUN_MQ_ERR_NS, 102, "aliyun_mq ack message error,code: {{.Code}}, message: {{.Message}}, receiptHandle: {{.ReceiptHandle}}, requestId: {{.RequestId}}")
)
View Source 
var (
	DefaultNumOfMessages int32 = 16
)

Functions

func CommitOrRollback 

func CommitOrRollback(p *AliyunMQTransProducer, receiptHandle string, trans string) (err error)

func ConstructPubMessage 

func ConstructPubMessage(pubMsgReq *PublishMessageRequest) (err error)

func ConstructRecMessage 

func ConstructRecMessage(entries *[]ConsumeMessageEntry)

func ContainsSpecialChar 

func ContainsSpecialChar(input string) (result bool)

func ParseError 

func ParseError(resp ErrorResponse, resource string) (err error)

Types

type AckMessageErrorEntry 

type AckMessageErrorEntry struct {
	XMLName xml.Name `xml:"Error" json:"-"`
	// Ack消息出错的错误码
	ErrorCode string `xml:"ErrorCode" json:"error_code"`
	// Ack消息出错的错误描述
	ErrorMessage string `xml:"ErrorMessage" json:"error_messages"`
	// Ack消息出错的消息句柄
	ReceiptHandle string `xml:"ReceiptHandle,omitempty" json:"receipt_handle"`
}

type AckMessageErrorResponse 

type AckMessageErrorResponse struct {
	XMLName        xml.Name               `xml:"Errors" json:"-"`
	FailedMessages []AckMessageErrorEntry `xml:"Error" json:"errors"`
}

type AliyunMQAckMsgDecoder 

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

func (*AliyunMQAckMsgDecoder) Decode 

func (p *AliyunMQAckMsgDecoder) Decode(reader io.Reader, v interface{}) (err error)

func (*AliyunMQAckMsgDecoder) DecodeError 

func (p *AliyunMQAckMsgDecoder) DecodeError(bodyBytes []byte, resource string, requestId string) (decodedError error, err error)

type AliyunMQClient 

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

func (*AliyunMQClient) GetConsumer 

func (p *AliyunMQClient) GetConsumer(instanceId string, topicName string, consumer string, messageTag string) MQConsumer

func (*AliyunMQClient) GetCredential 

func (p *AliyunMQClient) GetCredential() Credential

func (*AliyunMQClient) GetEndpoint 

func (p *AliyunMQClient) GetEndpoint() string

func (*AliyunMQClient) GetProducer 

func (p *AliyunMQClient) GetProducer(instanceId string, topicName string) MQProducer

func (*AliyunMQClient) GetTransProducer 

func (p *AliyunMQClient) GetTransProducer(instanceId string, topicName string, groupId string) MQTransProducer

func (*AliyunMQClient) Send 

func (p *AliyunMQClient) Send(decoder MQDecoder, method Method, headers map[string]string, message interface{}, resource string, v interface{}) (statusCode int, err error)

func (*AliyunMQClient) Send0 

func (p *AliyunMQClient) Send0(method Method, headers map[string]string, message interface{}, resource string) (*fasthttp.Response, error)

type AliyunMQConsumer 

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

func (*AliyunMQConsumer) AckMessage 

func (p *AliyunMQConsumer) AckMessage(receiptHandles []string) (err error)

func (*AliyunMQConsumer) ConsumeMessage 

func (p *AliyunMQConsumer) ConsumeMessage(respChan chan ConsumeMessageResponse, errChan chan error, numOfMessages int32, waitseconds int64)

func (*AliyunMQConsumer) ConsumeMessageOrderly 

func (p *AliyunMQConsumer) ConsumeMessageOrderly(respChan chan ConsumeMessageResponse, errChan chan error, numOfMessages int32, waitseconds int64)

func (*AliyunMQConsumer) Consumer 

func (p *AliyunMQConsumer) Consumer() string

func (*AliyunMQConsumer) InstanceId 

func (p *AliyunMQConsumer) InstanceId() string

func (*AliyunMQConsumer) MessageTag 

func (p *AliyunMQConsumer) MessageTag() string

func (*AliyunMQConsumer) TopicName 

func (p *AliyunMQConsumer) TopicName() string

type AliyunMQDecoder 

type AliyunMQDecoder struct {
}

func (*AliyunMQDecoder) Decode 

func (p *AliyunMQDecoder) Decode(reader io.Reader, v interface{}) (err error)

func (*AliyunMQDecoder) DecodeError 

func (p *AliyunMQDecoder) DecodeError(bodyBytes []byte, resource string, requestId string) (decodedError error, err error)

type AliyunMQProducer 

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

func (*AliyunMQProducer) InstanceId 

func (p *AliyunMQProducer) InstanceId() string

func (*AliyunMQProducer) PublishMessage 

func (p *AliyunMQProducer) PublishMessage(message PublishMessageRequest) (resp PublishMessageResponse, err error)

func (*AliyunMQProducer) TopicName 

func (p *AliyunMQProducer) TopicName() string

type AliyunMQTransProducer 

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

func (*AliyunMQTransProducer) Commit 

func (p *AliyunMQTransProducer) Commit(receiptHandle string) (err error)

func (*AliyunMQTransProducer) ConsumeHalfMessage 

func (p *AliyunMQTransProducer) ConsumeHalfMessage(respChan chan ConsumeMessageResponse, errChan chan error, numOfMessages int32, waitseconds int64)

func (*AliyunMQTransProducer) GroupId 

func (p *AliyunMQTransProducer) GroupId() string

func (*AliyunMQTransProducer) InstanceId 

func (p *AliyunMQTransProducer) InstanceId() string

func (*AliyunMQTransProducer) PublishMessage 

func (p *AliyunMQTransProducer) PublishMessage(message PublishMessageRequest) (resp PublishMessageResponse, err error)

func (*AliyunMQTransProducer) Rollback 

func (p *AliyunMQTransProducer) Rollback(receiptHandle string) (err error)

func (*AliyunMQTransProducer) TopicName 

func (p *AliyunMQTransProducer) TopicName() string

type ConsumeMessageEntry 

type ConsumeMessageEntry struct {
	MessageResponse
	// 消息ID
	MessageId string `xml:"MessageId" json:"message_id"`
	// 消息句柄
	ReceiptHandle string `xml:"ReceiptHandle" json:"receipt_handle"`
	// 消息体MD5
	MessageBodyMD5 string `xml:"MessageBodyMD5" json:"message_body_md5"`
	// 消息体
	MessageBody string `xml:"MessageBody" json:"message_body"`
	// 消息发送时间
	PublishTime int64 `xml:"PublishTime" json:"publish_time"`
	// 下次消费消息的时间（如果这次消费的消息没有Ack）
	NextConsumeTime int64 `xml:"NextConsumeTime" json:"next_consume_time"`
	// 第一次消费的时间,此值对于顺序消费没有意义
	FirstConsumeTime int64 `xml:"FirstConsumeTime" json:"first_consume_time"`
	// 消费次数
	ConsumedTimes int64 `xml:"ConsumedTimes" json:"consumed_times"`
	// 消息标签
	MessageTag string `xml:"MessageTag" json:"message_tag"`
	// 消息属性
	Properties map[string]string `xml:"-" json:"-"`
	// 序列化属性请勿使用
	PropInner string `xml:"Properties,omitempty" json:"properties,omitempty"`
	// 消息KEY
	MessageKey string `xml:"-" json:"-"`
	// 定时消息，单位毫秒（ms）,在指定时间戳（当前时间之后）进行投递
	StartDeliverTime int64 `xml:"-" json:"-"`
	// 顺序消息分区Key
	ShardingKey string `xml:"-" json:"-"`
	// 在消息属性中添加第一次消息回查的最快时间,单位秒,并且表征这是一条事务消息
	TransCheckImmunityTime int `xml:"-" json:"-"`
}

type ConsumeMessageResponse 

type ConsumeMessageResponse struct {
	XMLName  xml.Name              `xml:"Messages" json:"-"`
	Messages []ConsumeMessageEntry `xml:"Message" json:"messages"`
}

func (*ConsumeMessageResponse) Responses 

func (r *ConsumeMessageResponse) Responses() []*MessageResponse

type Credential 

type Credential interface {
	Signature(method Method, headers map[string]string, resource string) (signature string, err error)
	AccessKeyId() string
	AccessKeySecret() string
	SecurityToken() string
}

type ErrAckItem 

type ErrAckItem struct {
	ErrorHandle string
	ErrorMsg    string
	ErrorCode   string
}

type ErrorResponse 

type ErrorResponse struct {
	XMLName xml.Name `xml:"Error" json:"-"`
	// 错误码
	Code string `xml:"Code,omitempty" json:"code,omitempty"`
	// 错误描述
	Message string `xml:"Message,omitempty" json:"message,omitempty"`
	// 请求ID
	RequestId string `xml:"RequestId,omitempty" json:"request_id,omitempty"`
	// 请求HOST
	HostId string `xml:"HostId,omitempty" json:"host_id,omitempty"`
}

type MQClient 

type MQClient interface {
	GetProducer(instanceId string, topicName string) MQProducer
	GetTransProducer(instanceId string, topicName string, groupId string) MQTransProducer
	GetConsumer(instanceId string, topicName string, consumer string, messageTag string) MQConsumer
	GetCredential() Credential
	GetEndpoint() string
}

func NewAliyunMQClient 

func NewAliyunMQClient(endpoint, accessKeyId, accessKeySecret, securityToken string) MQClient

func NewAliyunMQClientWithTimeout 

func NewAliyunMQClientWithTimeout(endpoint, accessKeyId, accessKeySecret, securityToken string, timeout time.Duration) MQClient

type MQConsumer 

type MQConsumer interface {
	// 主题名字
	TopicName() string
	// 实例ID，可空
	InstanceId() string
	// 消费者的名字
	Consumer() string
	// 消费消息过滤的标签
	MessageTag() string
	// 消费消息，如果该条消息没有被 {AckMessage} 确认消费成功，即在NextConsumeTime时会再次消费到该条消息
	ConsumeMessage(respChan chan ConsumeMessageResponse, errChan chan error, numOfMessages int32, waitseconds int64)
	// 顺序消费消息，获取的消息可能是多个分区但是一个分区内消息一定是顺序的，对于一个分区的消息必须要全部ACK成功才能消费下一批消息
	// 否则在NextConsumeTime后会再次消费到相同的消息
	ConsumeMessageOrderly(respChan chan ConsumeMessageResponse, errChan chan error, numOfMessages int32, waitseconds int64)
	// 确认消息消费成功
	AckMessage(receiptHandles []string) (err error)
}

MQ的消息消费者

type MQCredential 

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

func NewMQCredential 

func NewMQCredential(accessKeyId string, accessKeySecret string, securityToken string) *MQCredential

func (*MQCredential) AccessKeyId 

func (p *MQCredential) AccessKeyId() string

func (*MQCredential) AccessKeySecret 

func (p *MQCredential) AccessKeySecret() string

func (*MQCredential) SecurityToken 

func (p *MQCredential) SecurityToken() string

func (*MQCredential) Signature 

func (p *MQCredential) Signature(method Method, headers map[string]string, resource string) (signature string, err error)

type MQDecoder 

type MQDecoder interface {
	Decode(reader io.Reader, v interface{}) (err error)
	DecodeError(bodyBytes []byte, resource string, requestId string) (decodedError error, err error)
}

func NewAliyunMQAckMsgDecoder 

func NewAliyunMQAckMsgDecoder() MQDecoder

func NewAliyunMQDecoder 

func NewAliyunMQDecoder() MQDecoder

type MQProducer 

type MQProducer interface {
	// 主题名字
	TopicName() string
	// 实例ID，可空
	InstanceId() string
	// 发送消息
	PublishMessage(message PublishMessageRequest) (resp PublishMessageResponse, err error)
}

MQ的消息生产者

type MQTransProducer 

type MQTransProducer interface {
	// 主题名字
	TopicName() string
	// 实例ID，可空
	InstanceId() string
	// GroupId,非空
	GroupId() string
	// 发送消息
	PublishMessage(message PublishMessageRequest) (resp PublishMessageResponse, err error)
	// 消费事务半消息
	ConsumeHalfMessage(respChan chan ConsumeMessageResponse, errChan chan error, numOfMessages int32, waitseconds int64)
	// 提交事务消息
	Commit(receiptHandle string) (err error)
	// 回滚事务消息
	Rollback(receiptHandle string) (err error)
}

MQ的事务消息生产者

type MessageResponder 

type MessageResponder interface {
	Responses() []*MessageResponse
}

type MessageResponse 

type MessageResponse struct {
	XMLName xml.Name `xml:"Message" json:"-"`
	// 错误码
	Code string `xml:"Code,omitempty" json:"code,omitempty"`
	// 错误描述
	Message string `xml:"Message,omitempty" json:"message,omitempty"`
	// 请求ID
	RequestId string `xml:"RequestId,omitempty" json:"request_id,omitempty"`
	// 请求HOST
	HostId string `xml:"HostId,omitempty" json:"host_id,omitempty"`
}

type Method 

type Method string

type PublishMessageRequest 

type PublishMessageRequest struct {
	XMLName xml.Name `xml:"Message" json:"-"`
	// 消息体
	MessageBody string `xml:"MessageBody" json:"message_body"`
	// 消息标签
	MessageTag string `xml:"MessageTag,omitempty" json:"message_tag,omitempty"`
	// 消息属性
	Properties map[string]string `xml:"-" json:"-"`
	// 消息KEY
	MessageKey string `xml:"-" json:"-"`
	// 定时消息，单位毫秒（ms）,在指定时间戳（当前时间之后）进行投递
	StartDeliverTime int64 `xml:"-" json:"-"`
	// 在消息属性中添加第一次消息回查的最快时间,单位秒,并且表征这是一条事务消息, 10~300s
	TransCheckImmunityTime int `xml:"-" json:"-"`
	// 分区顺序消息中区分不同分区的关键字段，sharding key 于普通消息的 key 是完全不同的概念。全局顺序消息，该字段可以设置为任意非空字符串。
	ShardingKey string `xml:"-" json:"-"`
	// contains filtered or unexported fields
}

type PublishMessageResponse 

type PublishMessageResponse struct {
	MessageResponse
	// 消息ID
	MessageId string `xml:"MessageId" json:"message_id"`
	// 消息体MD5
	MessageBodyMD5 string `xml:"MessageBodyMD5" json:"message_body_md5"`
	// 消息句柄只有事务消息才有
	ReceiptHandle string `xml:"ReceiptHandle" json:"receipt_handle"`
}

func (*PublishMessageResponse) Responses 

func (r *PublishMessageResponse) Responses() []*MessageResponse

type ReceiptHandles 

type ReceiptHandles struct {
	XMLName xml.Name `xml:"ReceiptHandles"`
	// 消息句柄
	ReceiptHandles []string `xml:"ReceiptHandle"`
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.