mq

Documentation

Index

• Constants
• Variables
• func ConsumerContext(traceID string, d time.Duration) (ctx context.Context, cancel context.CancelFunc)
• func ObserveMsg(topic, name, status, reason string, start time.Time)
• func WarpMessage(ctx context.Context, tag string, msg proto.Message) ([]byte, error)
• type Consumer
• type ConsumerMeta
• type HandleFunc
• type MqMessage
  • func Unmarshal(message []byte) (*MqMessage, error)
  • func (*MqMessage) Descriptor() ([]byte, []int)
  • func (m *MqMessage) GetMessage() []byte
  • func (m *MqMessage) GetTag() string
  • func (m *MqMessage) GetTraceId() string
  • func (m *MqMessage) GetTxId() string
  • func (m *MqMessage) Marshal() (dAtA []byte, err error)
  • func (m *MqMessage) MarshalTo(dAtA []byte) (int, error)
  • func (m *MqMessage) MarshalToSizedBuffer(dAtA []byte) (int, error)
  • func (*MqMessage) ProtoMessage()
  • func (m *MqMessage) Reset()
  • func (m *MqMessage) Size() (n int)
  • func (m *MqMessage) String() string
  • func (m *MqMessage) Unmarshal(dAtA []byte) error
  • func (m *MqMessage) XXX_DiscardUnknown()
  • func (m *MqMessage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
  • func (m *MqMessage) XXX_Merge(src proto.Message)
  • func (m *MqMessage) XXX_Size() int
  • func (m *MqMessage) XXX_Unmarshal(b []byte) error
• type Option
  • func WithNPoll(n int) Option
  • func WithNProc(n int) Option
  • func WithName(name string) Option
  • func WithProcTimeout(d time.Duration) Option

Constants

const (
	DefaultHandlerQueueSize = 100

	TagCreate        = "create"
	TagReply         = "reply"
	TagDelete        = "delete"
	TagSync          = "sync"
	TagListMissed    = "list_missed"
	TagSubListMissed = "sub_list_missed"
	TagLike          = "like"
	TagHate          = "hate"
)

Variables

var (
	ErrInvalidLengthMessage        = fmt.Errorf("proto: negative length found during unmarshaling")
	ErrIntOverflowMessage          = fmt.Errorf("proto: integer overflow")
	ErrUnexpectedEndOfGroupMessage = fmt.Errorf("proto: unexpected end of group")
)

var (
	ApiLogger *log.Logger
	ExcLogger *log.Logger

	MetricEventHandlerMsgDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
		Name:    "event_handler_msg_duration",
		Help:    "消息处理延时",
		Buckets: prometheus.DefBuckets,
	}, []string{"name", "topic", "status", "reason"})
)

Functions

func ConsumerContext

func ConsumerContext(traceID string, d time.Duration) (ctx context.Context, cancel context.CancelFunc)

func ObserveMsg

func ObserveMsg(topic, name, status, reason string, start time.Time)

func WarpMessage

func WarpMessage(ctx context.Context, tag string, msg proto.Message) ([]byte, error)

Types

type Consumer

type Consumer interface {
	Start() error
	Stop() error
}

type ConsumerMeta

type ConsumerMeta struct {
	Name        string
	Fn          HandleFunc
	NPoll       int
	NProc       int
	ProcTimeout time.Duration
	Done        chan struct{}
}

type HandleFunc

type HandleFunc func(ctx context.Context, msg *MqMessage) error

type MqMessage

type MqMessage struct {
	TxId                 string   `protobuf:"bytes,1,opt,name=tx_id,json=txId,proto3" json:"tx_id,omitempty"`
	TraceId              string   `protobuf:"bytes,2,opt,name=trace_id,json=traceId,proto3" json:"trace_id,omitempty"`
	Tag                  string   `protobuf:"bytes,3,opt,name=tag,proto3" json:"tag,omitempty"`
	Message              []byte   `protobuf:"bytes,4,opt,name=message,proto3" json:"message,omitempty"`
	XXX_NoUnkeyedLiteral struct{} `json:"-"`
	XXX_unrecognized     []byte   `json:"-"`
	XXX_sizecache        int32    `json:"-"`
}

func Unmarshal

func Unmarshal(message []byte) (*MqMessage, error)

func (*MqMessage) Descriptor

func (*MqMessage) Descriptor() ([]byte, []int)

func (*MqMessage) GetMessage

func (m *MqMessage) GetMessage() []byte

func (*MqMessage) GetTag

func (m *MqMessage) GetTag() string

func (*MqMessage) GetTraceId

func (m *MqMessage) GetTraceId() string

func (*MqMessage) GetTxId

func (m *MqMessage) GetTxId() string

func (*MqMessage) Marshal

func (m *MqMessage) Marshal() (dAtA []byte, err error)

func (*MqMessage) MarshalTo

func (m *MqMessage) MarshalTo(dAtA []byte) (int, error)

func (*MqMessage) MarshalToSizedBuffer

func (m *MqMessage) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*MqMessage) ProtoMessage

func (*MqMessage) ProtoMessage()

func (*MqMessage) Reset

func (m *MqMessage) Reset()

func (*MqMessage) Size

func (m *MqMessage) Size() (n int)

func (*MqMessage) String

func (m *MqMessage) String() string

func (*MqMessage) Unmarshal

func (m *MqMessage) Unmarshal(dAtA []byte) error

func (*MqMessage) XXX_DiscardUnknown

func (m *MqMessage) XXX_DiscardUnknown()

func (*MqMessage) XXX_Marshal

func (m *MqMessage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*MqMessage) XXX_Merge

func (m *MqMessage) XXX_Merge(src proto.Message)

func (*MqMessage) XXX_Size

func (m *MqMessage) XXX_Size() int

func (*MqMessage) XXX_Unmarshal

func (m *MqMessage) XXX_Unmarshal(b []byte) error

type Option

type Option func(*ConsumerMeta)

func WithNPoll

func WithNPoll(n int) Option

func WithNProc

func WithNProc(n int) Option

func WithName

func WithName(name string) Option

func WithProcTimeout

func WithProcTimeout(d time.Duration) Option