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

beanq

package module

Go to main page

Versions in this module

v3

v3.2.0

Jan 14, 2025

Changes in this version

+ const ExecuteFailed

+ const ExecuteSuccess

+ const RollbackFailed

+ const RollbackResultProcessFailed

+ const RollbackSuccess

+ var ErrExtendFailed = bstatus.BqError("failed to extend lock")

+ var ErrFailed = bstatus.BqError("failed to acquire lock")

+ var ErrLockAlreadyExpired = bstatus.BqError("failed to unlock, lock was already expired")

+ var NilCancel = errors.New("beanq:cancel is nil")

+ var NilHandle = errors.New("beanq:handle is nil")

+ func GetBrokerDriver[T any]() T

+ func WfOptionMux(mux WFMux) func(workflow *Workflow)

+ func WfOptionRecordErrorHandler(handler func(error)) func(workflow *Workflow)

+ type BQClient struct

+ func (b *BQClient) Dynamic(options ...DynamicOption) *BQClient

+ func (b *BQClient) GetId() string

+ func (b *BQClient) Priority(priority float64) *BQClient

+ func (b *BQClient) PublishInSequential(channel, topic string, payload []byte) *SequentialCmd

+ func (b *BQClient) SetId(id string) *BQClient

+ func (b *BQClient) WithContext(ctx context.Context) *BQClient

+ func (t BQClient) Publish(channel, topic string, payload []byte) error

+ func (t BQClient) PublishAtTime(channel, topic string, payload []byte, atTime time.Time) error

+ func (t BQClient) Subscribe(channel, topic string, handle IConsumeHandle) (IBaseSubscribeCmd, error)

+ func (t BQClient) SubscribeDelay(channel, topic string, handle IConsumeHandle) (IBaseSubscribeCmd, error)

+ func (t BQClient) SubscribeSequential(channel, topic string, handle IConsumeHandle) (IBaseSubscribeCmd, error)

+ type BaseTask struct

+ func (t *BaseTask) Execute() error

+ func (t *BaseTask) ID() string

+ func (t *BaseTask) OnExecute(fn func(task Task) error) *BaseTask

+ func (t *BaseTask) OnRollback(fn func(task Task) error) *BaseTask

+ func (t *BaseTask) Rollback() error

+ func (t *BaseTask) Statement() []byte

+ func (t *BaseTask) WithRecordStatement(statement []byte) *BaseTask

+ type BeanqConfig struct

+ Broker string

+ ConsumeTimeOut time.Duration

+ ConsumerPoolSize int

+ DeadLetterIdleTime time.Duration

+ DeadLetterTicker time.Duration

+ Health Health

+ JobMaxRetries int

+ KeepFailedJobsInHistory time.Duration

+ KeepSuccessJobsInHistory time.Duration

+ MinConsumers int64

+ PublishTimeOut time.Duration

+ Redis Redis

+ type Broker struct

+ func NewBroker(config *BeanqConfig) *Broker

+ func (t *Broker) AddConsumer(moodType btype.MoodType, channel, topic string, subscribe IConsumeHandle) error

+ func (t *Broker) Dequeue(ctx context.Context, channel, topic string, do public.CallBack)

+ func (t *Broker) Enqueue(ctx context.Context, data map[string]any) error

+ func (t *Broker) Migrate(ctx context.Context, data []map[string]any) error

+ func (t *Broker) Start(ctx context.Context)

+ func (t *Broker) Status(ctx context.Context, channel, topic, id string) (map[string]string, error)

+ func (t *Broker) WaitSignal(cancel context.CancelFunc) <-chan bool

+ type Client struct

+ Channel string

+ MaxLen int64

+ Priority float64

+ Retry int

+ TimeToRun time.Duration

+ Topic string

+ func New(config *BeanqConfig, options ...ClientOption) *Client

+ func (c *Client) BQ() *BQClient

+ func (c *Client) CheckAckStatus(ctx context.Context, channel, topic, id string) (*Message, error)

+ func (c *Client) Ping()

+ func (c *Client) ServeHttp(ctx context.Context)

+ func (c *Client) Wait(ctx context.Context)

+ type ClientOption func(client *Client)

+ func WithCaptureExceptionOption(handler func(ctx context.Context, err any)) ClientOption

+ type DebugLog struct

+ On bool

+ Path string

+ type DefaultHandle struct

+ DoCancel func(ctx context.Context, message *Message) error

+ DoError func(ctx context.Context, err error)

+ DoHandle func(ctx context.Context, message *Message) error

+ func (c DefaultHandle) Cancel(ctx context.Context, message *Message) error

+ func (c DefaultHandle) Error(ctx context.Context, err error)

+ func (c DefaultHandle) Handle(ctx context.Context, message *Message) error

+ type DelayFunc func(tries int) time.Duration

+ type DynamicOption func(option *dynamicOption)

+ func DynamicKeyOpt(key string) DynamicOption

+ type Handler struct

+ type Health struct

+ Host string

+ Port string

+ type History struct

+ Mongo struct{ ... }

+ On bool

+ type IBaseCmd interface

+ type IBaseSubscribeCmd interface

+ Run func(ctx context.Context)

+ type IConsumeCancel interface

+ Cancel func(ctx context.Context, message *Message) error

+ type IConsumeError interface

+ Error func(ctx context.Context, err error)

+ type IConsumeHandle interface

+ Handle func(ctx context.Context, message *Message) error

+ type Message struct

+ AddTime string

+ BeginTime time.Time

+ Channel string

+ Consumer string

+ EndTime time.Time

+ ExecuteTime time.Time

+ Id string

+ Info bstatus.FlagInfo

+ Level bstatus.LevelMsg

+ MaxLen int64

+ MoodType btype.MoodType

+ Payload string

+ PendingRetry int64

+ Priority float64

+ Response any

+ Retry int

+ RunTime string

+ Status bstatus.Status

+ TimeToRun time.Duration

+ Topic string

+ func (m Message) MarshalBinary() (data []byte, err error)

+ func (m Message) ToMap() map[string]any

+ type MessageM map[string]any

+ func (data MessageM) ToMessage() *Message

+ type MessageS map[string]string

+ func (data MessageS) ToMessage() *Message

+ type Mutex struct

+ func (m *Mutex) Eval(ctx context.Context, client redis.UniversalClient, script *Script, ...) (interface{}, error)

+ func (m *Mutex) ExtendContext(ctx context.Context) (bool, error)

+ func (m *Mutex) LockContext(ctx context.Context) error

+ func (m *Mutex) Name() string

+ func (m *Mutex) UnlockContext(ctx context.Context) (bool, error)

+ func (m *Mutex) Until() time.Time

+ func (m *Mutex) Value() string

+ type MuxClient struct

+ func NewMuxClient(client redis.UniversalClient) *MuxClient

+ func (p *MuxClient) NewMutex(name string, options ...MuxOption) *Mutex

+ func (p *MuxClient) SetExpireTime(expireTime time.Duration) *MuxClient

+ func (p *MuxClient) SetPrefix(prefix string) *MuxClient

+ type MuxOption interface

+ Apply func(*Mutex)

+ func WithDriftFactor(factor float64) MuxOption

+ func WithExpiry(expiry time.Duration) MuxOption

+ func WithFailFast(b bool) MuxOption

+ func WithGenValueFunc(genValueFunc func() (string, error)) MuxOption

+ func WithRetryDelay(delay time.Duration) MuxOption

+ func WithRetryDelayFunc(delayFunc DelayFunc) MuxOption

+ func WithSetNXOnExtend() MuxOption

+ func WithShufflePools(b bool) MuxOption

+ func WithTimeoutFactor(factor float64) MuxOption

+ func WithTries(tries int) MuxOption

+ func WithValue(v string) MuxOption

+ type OptionFunc func(*Mutex)

+ func (f OptionFunc) Apply(mutex *Mutex)

+ type Publish struct

+ type Queue struct

+ Channel string

+ DelayChannel string

+ DelayTopic string

+ MaxLen int64

+ Priority float64

+ TimeToRun time.Duration

+ Topic string

+ type Redis struct

+ Database int

+ DialTimeout time.Duration

+ Host string

+ MaxLen int64

+ MaxRetries int

+ MinIdleConnections int

+ Password string

+ PoolSize int

+ PoolTimeout time.Duration

+ Port string

+ Prefix string

+ ReadTimeout time.Duration

+ WriteTimeout time.Duration

+ type Script struct

+ Hash string

+ KeyCount int

+ Src string

+ func NewScript(keyCount int, src string) *Script

+ type SequentialCmd struct

+ func (s *SequentialCmd) Error() error

+ func (s *SequentialCmd) WaitingAck() (*Message, error)

+ type Subscribe struct

+ func (t *Subscribe) Run(ctx context.Context)

+ type Task interface

+ Execute func() error

+ ID func() string

+ Rollback func() error

+ Statement func() []byte

+ type TaskStatus struct

+ func (t *TaskStatus) Error() error

+ func (t *TaskStatus) Statement() string

+ func (t *TaskStatus) Status() string

+ func (t *TaskStatus) String() string

+ type UI struct

+ ExpiresAt time.Duration

+ GoogleAuth struct{ ... }

+ Issuer string

+ JwtKey string

+ Port string

+ Root struct{ ... }

+ SendGrid struct{ ... }

+ Stmt struct{ ... }

+ Subject string

+ type WFMux interface

+ ExtendContext func(ctx context.Context) (bool, error)

+ LockContext func(ctx context.Context) error

+ Name func() string

+ UnlockContext func(ctx context.Context) (bool, error)

+ Until func() time.Time

+ Value func() string

+ type Workflow struct

+ func NewWorkflow(ctx context.Context, message *Message) *Workflow

+ func (w *Workflow) CurrentTask() Task

+ func (w *Workflow) GetGid() string

+ func (w *Workflow) Init(opts ...func(workflow *Workflow))

+ func (w *Workflow) Message() *Message

+ func (w *Workflow) NewTask(ids ...string) *BaseTask

+ func (w *Workflow) OnRollbackResult(handler func(taskID string, err error) error) *Workflow

+ func (w *Workflow) Results() []error

+ func (w *Workflow) Run() (err error)

+ func (w *Workflow) SetGid(gid string)

+ func (w *Workflow) TrackRecord(taskID string, status TaskStatus)

+ type WorkflowHandler func(ctx context.Context, wf *Workflow) error

+ func (c WorkflowHandler) Handle(ctx context.Context, message *Message) error

+ type WorkflowRecord struct

+ func NewWorkflowRecord() *WorkflowRecord

+ func (w *WorkflowRecord) SyncWrite(ctx context.Context, data any) error

+ func (w *WorkflowRecord) Write(ctx context.Context, data any)

v3.1.0

Oct 23, 2024

v3.0.1

Oct 17, 2024

Other modules containing this package

github.com/retail-ai-inc/beanq