Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

rmq

package module

Go to main page

Versions in this module

v1

v1.1.4

Feb 22, 2023

v1.1.3

Dec 23, 2022

Changes in this version

type Message

+ func (m *Message) SetRawTask(name string, data any) (msg *Message, err error)

v1.1.2

Nov 1, 2022

Changes in this version

+ var DefaultLog = newDefaultLog()

+ var DefaultMeta = Meta

+ var DefaultRedisBrokerConfig = RedisBrokerConfig

+ var DefaultRetryRule = []int

+ var RetryMeta = Meta

+ func BrokerHookProtect(ctx context.Context, f func(ctx context.Context) error) (err error)

+ func Json(v any) string

+ func Protect(f func() error) (err error)

+ type BatchError struct

+ func (b *BatchError) HasError() bool

+ type Broker interface

+ Decode func(bytes []byte) (msg *Message, err error)

+ Encode func(*Message) ([]byte, error)

+ Pop func(ctx context.Context) (msg *Message, err error)

+ Push func(context.Context, ...*Message) error

+ type BrokerAfterExit interface

+ AfterExit func(ctx context.Context) error

+ type BrokerAfterStart interface

+ AfterStart func(ctx context.Context) error

+ type BrokerBeforeExit interface

+ BeforeExit func(ctx context.Context) error

+ type BrokerBeforeStart interface

+ BeforeStart func(ctx context.Context) error

+ type Callback func(ctx context.Context, msg *Message) (string, error)

+ type CommandTask struct

+ Command []string

+ Shell string

+ func NewCommandTask(shell string, command ...string) *CommandTask

+ func (c *CommandTask) Run(ctx context.Context) (result string, err error)

+ func (c *CommandTask) TaskName() string

+ type Hook struct

+ func (h *Hook) OnComplete(v func(ctx context.Context, runtime *TaskRuntime) error)

+ func (h *Hook) OnContext(v func(ctx context.Context, runtime *TaskRuntime) context.Context)

+ func (h *Hook) OnPush(v func(ctx context.Context, msg ...*Message) ([]*Message, error))

+ func (h *Hook) OnRun(v func(ctx context.Context, runtime *TaskRuntime) error)

+ type HttpTask struct

+ Body json.RawMessage

+ Header map[string]string

+ Method string

+ Url string

+ func NewHttpTaskGet(format string, arg ...any) *HttpTask

+ func NewHttpTaskJsonPost(url string, data any) *HttpTask

+ func NewHttpTaskPostPostForm(url string, data url.Values) *HttpTask

+ func (h *HttpTask) Load(ctx context.Context, msg *Message) (err error)

+ func (h *HttpTask) Message() (msg *Message, err error)

+ func (h *HttpTask) Run(ctx context.Context) (result string, err error)

+ func (h *HttpTask) Scan(src []byte) (err error)

+ func (h *HttpTask) SetBody(data []byte) *HttpTask

+ func (h *HttpTask) SetHeader(k, v string) *HttpTask

+ func (h *HttpTask) SetHeaders(headers map[string]string) *HttpTask

+ func (h *HttpTask) SetMethod(method string) *HttpTask

+ func (h *HttpTask) TaskName() string

+ type Logger interface

+ Errorf func(format string, arg ...interface{})

+ Infof func(format string, arg ...interface{})

+ Warningf func(format string, args ...interface{})

+ type Message struct

+ CreatedAt Timestamp

+ Data json.RawMessage

+ ExpiredAt Timestamp

+ Id string

+ Meta Meta

+ RunAt Timestamp

+ Task string

+ func NewBlankMsg() *Message

+ func NewMsg() *Message

+ func (m *Message) SetCallback(callbackName string, data any) (msg *Message, err error)

+ func (m *Message) SetData(data any) *Message

+ func (m *Message) SetDelay(delay time.Duration) *Message

+ func (m *Message) SetExpire(d time.Duration) *Message

+ func (m *Message) SetExpiredAt(t time.Time) *Message

+ func (m *Message) SetMaxRetry(retry int) *Message

+ func (m *Message) SetMeta(meta Meta) *Message

+ func (m *Message) SetRawData(data json.RawMessage) *Message

+ func (m *Message) SetRetryRule(rule []int) *Message

+ func (m *Message) SetTask(task Task) (message *Message, err error)

+ func (m *Message) SetTimeout(t time.Duration) *Message

+ func (m *Message) SetTraceId(traceId string) *Message

+ func (m *Message) String() string

+ func (m *Message) TryRetry(delay time.Duration) *Message

+ type Messages []*Message

+ type Meta struct

+ Delay int

+ Retry [2]int

+ RetryRule []int

+ Timeout int

+ TraceId string

+ type OnComplete interface

+ OnComplete func(ctx context.Context)

+ type OnFail interface

+ OnFail func(ctx context.Context)

+ type OnLoad interface

+ Load func(ctx context.Context, msg *Message) error

+ type OnSuccess interface

+ OnSuccess func(ctx context.Context)

+ type Process interface

+ Exec func(ctx context.Context, msg *TaskRuntime) (err error)

+ type Producer interface

+ Push func(ctx context.Context)

+ type RedisBroker struct

+ func NewRedisBroker(rd *redis.Client, c RedisBrokerConfig, log Logger) *RedisBroker

+ func (r *RedisBroker) AfterExit(ctx context.Context) error

+ func (r *RedisBroker) AfterStart(ctx context.Context) error

+ func (r *RedisBroker) BeforeExit(ctx context.Context) error

+ func (r *RedisBroker) BeforeStart(ctx context.Context) error

+ func (r *RedisBroker) Decode(bytes []byte) (msg *Message, err error)

+ func (r *RedisBroker) Encode(msg *Message) ([]byte, error)

+ func (r *RedisBroker) Pop(ctx context.Context) (msg *Message, err error)

+ func (r *RedisBroker) Push(ctx context.Context, msg ...*Message) (err error)

+ type RedisBrokerConfig struct

+ DelayKey string

+ DelayWaitDuration time.Duration

+ Key string

+ WaitDuration time.Duration

+ type Register struct

+ func (r *Register) AllTask() map[string]*TaskInfo

+ func (r *Register) CreateTask(name string) (task Task, v *TaskInfo, err error)

+ func (r *Register) Register(v Task)

+ func (r *Register) RegisterFunc(name string, callback Callback)

+ type Rmq struct

+ Hook *Hook

+ func NewRmq(broker Broker, logger Logger) *Rmq

+ func (q *Rmq) Exit()

+ func (q *Rmq) Push(ctx context.Context, msg ...*Message) (err error)

+ func (q *Rmq) Register(task ...Task)

+ func (q *Rmq) RegisterFunc(name string, callback Callback)

+ func (q *Rmq) SetBroker(b Broker)

+ func (q *Rmq) SetProcess(p Process)

+ func (q *Rmq) StartWorker(c *WorkerConfig)

+ func (q *Rmq) Tasks() map[string]*TaskInfo

+ func (q *Rmq) TryRetry(ctx context.Context, msg *Message) (err error)

+ func (q *Rmq) TryRun(ctx context.Context, msg *Message)

+ type Task interface

+ Run func(ctx context.Context) (result string, err error)

+ TaskName func() string

+ type TaskInfo struct

+ Callback Callback

+ IsCallback bool

+ Name string

+ ReflectType reflect.Type

+ type TaskName interface

+ TaskName func() string

+ type TaskRuntime struct

+ Duration time.Duration

+ EndTime time.Time

+ Error error

+ Msg *Message

+ Result string

+ StartTime time.Time

+ TaskError error

+ func (a *TaskRuntime) IsSuccess() bool

+ type TaskScanner interface

+ Scan func(src []byte) error

+ type TaskValuer interface

+ Value func() ([]byte, error)

+ type Timestamp int64

+ func NewTimestamp(t time.Time) Timestamp

+ func Now() Timestamp

+ func (t Timestamp) Add(d time.Duration) Timestamp

+ func (t Timestamp) Date() string

+ func (t Timestamp) DateTime() string

+ func (t Timestamp) Format(layout string) string

+ func (t Timestamp) Time() time.Time

+ type WorkerConfig struct

+ Concurrent int

+ WorkerNum int