aliyunmns

Documentation

Overview

Package aliyunmns Message queue adapter on AliyunMNS.

Index

• Constants
• Variables
• type AgentManager
• type Consumer
  • func NewConsumer(id, parallel int) *Consumer
  • func (o *Consumer) Dispatcher(v func(*base.Task, *base.Message) bool)
  • func (o *Consumer) Processor() process.Processor
• type Producer
  • func NewProducer() *Producer
  • func (o *Producer) Processor() process.Processor
  • func (o *Producer) Publish(payload *base.Payload) (string, error)
• type Remoter
  • func NewRemoter() *Remoter
  • func (o *Remoter) Build(ctx context.Context, task *base.Task) (err error)
  • func (o *Remoter) BuildById(ctx context.Context, id int) (err error)
  • func (o *Remoter) Destroy(ctx context.Context, task *base.Task) (err error)
  • func (o *Remoter) DestroyById(ctx context.Context, id int) (err error)
  • func (o *Remoter) Processor() process.Processor
• type TopicMessage
  • func (o *TopicMessage) Release()

Constants

const (
	DefaultLoggingEnable          = true
	DefaultMaxMessageSize         = 65536
	DefaultMessageRetentionPeriod = 604800
	DefaultNotifyContentFormat    = "JSON"
	DefaultNotifyStrategy         = mns.BACKOFF_RETRY
	DefaultPollingWaitSeconds     = 30
	DefaultSlices                 = 0
	DefaultVisibilityTimeout      = 30
)

Variables

var (
	Agent AgentManager

	RegexMessageNotExist      = regexp.MustCompile(`code\s*:\s*MessageNotExist`)
	RegexQueueNotExist        = regexp.MustCompile(`code\s*:\s*QueueNotExist`)
	RegexSubscriptionNotExist = regexp.MustCompile(`code\s*:\s*SubscriptionNotExist`)
	RegexTopicNotExist        = regexp.MustCompile(`code\s*:\s*TopicNotExist`)
)

var (
	TopicMessagePool *sync.Pool
)

Types

type AgentManager

type AgentManager interface {
	// Build
	// create remote relation if not exists.
	Build(ctx context.Context, task *base.Task) error

	// Destroy
	// delete remote relation if exists.
	Destroy(ctx context.Context, task *base.Task) error

	// GenQueueName
	// generate and return queue name.
	//
	//   Agent.GenQueueName(1) // return "X-Q1"
	GenQueueName(id int) string

	// GenSubscriptionName
	// generate and return subscription name.
	//
	//   Agent.GenSubscriptionName(1) // return "X-S1"
	GenSubscriptionName(id int) string

	// GenTopicName
	// generate and return topic name.
	//
	//   Agent.GenTopicName("Topic") // return "X-Topic"
	GenTopicName(name string) string

	// GetQueueClient
	// return queue client instance.
	GetQueueClient(id int) mns.AliMNSQueue

	// GetQueueManager
	// return queue manager interface.
	GetQueueManager() mns.AliQueueManager

	// GetTopicClient
	// return topic client instance.
	GetTopicClient(name string) mns.AliMNSTopic

	// GetTopicManager
	// return topic manager interface.
	GetTopicManager() mns.AliTopicManager
}

AgentManager interface of aliyunmns agent.

type Consumer

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

Consumer struct for aliyun mns consumer.

func NewConsumer

func NewConsumer(id, parallel int) *Consumer

func (*Consumer) Dispatcher

func (o *Consumer) Dispatcher(v func(*base.Task, *base.Message) bool)

func (*Consumer) Processor

func (o *Consumer) Processor() process.Processor

type Producer

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

Producer struct for aliyun mns producer.

func NewProducer

func NewProducer() *Producer

func (*Producer) Processor

func (o *Producer) Processor() process.Processor

func (*Producer) Publish

func (o *Producer) Publish(payload *base.Payload) (string, error)

Publish send topic message to aliyunmns.

type Remoter

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

Remoter struct for aliyun mns remoter.

func NewRemoter

func NewRemoter() *Remoter

func (*Remoter) Build

func (o *Remoter) Build(ctx context.Context, task *base.Task) (err error)

func (*Remoter) BuildById

func (o *Remoter) BuildById(ctx context.Context, id int) (err error)

func (*Remoter) Destroy

func (o *Remoter) Destroy(ctx context.Context, task *base.Task) (err error)

func (*Remoter) DestroyById

func (o *Remoter) DestroyById(ctx context.Context, id int) (err error)

func (*Remoter) Processor

func (o *Remoter) Processor() process.Processor

type TopicMessage

type TopicMessage struct {
	Message   string `json:"Message"`
	MessageId string `json:"MessageId"`
	TopicName string `json:"TopicName"`
}

func (*TopicMessage) Release

func (o *TopicMessage) Release()