queue

Documentation

Index

• Constants
• func NewErrorHandler(wsHub *websocket.Hub) asynq.ErrorHandlerFunc
• func ShouldNotRetryError(err error) bool
• type Client
  • func NewClient(redisAddr string) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) EnqueueDatabaseOperation(ctx context.Context, action string, payload interface{}, priority string) (*asynq.TaskInfo, error)
  • func (c *Client) EnqueueMikrotikCommand(ctx context.Context, action string, payload interface{}, priority string) (*asynq.TaskInfo, error)
  • func (c *Client) EnqueueTask(ctx context.Context, taskType string, payload interface{}, priority string, ...) (*asynq.TaskInfo, error)
• type DatabaseOperationPayload
• type DatabaseQueueHandler
  • func NewDatabaseQueueHandler(wsHub *websocket.Hub) *DatabaseQueueHandler
  • func (h *DatabaseQueueHandler) HandleTask(ctx context.Context, task *asynq.Task) error
• type ErrorHandlerFunc
• type GenericTaskPayload
• type Handler
• type Handlers
• type MikrotikCommandPayload
• type MikrotikQueueHandler
  • func NewMikrotikQueueHandler(mikroTikService *service.MikroTikMangerService, wsHub *websocket.Hub) *MikrotikQueueHandler
  • func (h *MikrotikQueueHandler) HandleTask(ctx context.Context, task *asynq.Task) error
• type Server
  • func NewServer(redisAddr string, manager *mikrotik.Manager, wsHub *websocket.Hub, ...) (*Server, error)
  • func (s *Server) GracefullyShutdown()
  • func (s *Server) Start() error
  • func (s *Server) Stop()

Constants

const (
	TypeMikrotikCommand     = "mikrotik:command"
	TypeDatabaseOperation   = "database:operation"
	ActionSaveMpesaCallback = "action:save_payment_callback"
	ActionMikrotikLoginUser = "action:mikrotik_login_user"
	ActionMikrotikCommand   = "action:mikrotik_command"

	QueueCritical  = "critical"  // For login/logout, authentication, critical DB updates
	QueueDefault   = "default"   // For regular commands, standard DB operations
	QueueReporting = "reporting" // For logs, stats collection, non-critical DB reads/writes
)

Functions

func NewErrorHandler

func NewErrorHandler(wsHub *websocket.Hub) asynq.ErrorHandlerFunc

func ShouldNotRetryError

func ShouldNotRetryError(err error) bool

Types

type Client

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

Client is the client for enqueuing tasks

func NewClient

func NewClient(redisAddr string) (*Client, error)

NewClient creates a new queue client

func (*Client) Close

func (c *Client) Close() error

Close closes the queue client

func (*Client) EnqueueDatabaseOperation

func (c *Client) EnqueueDatabaseOperation(ctx context.Context, action string, payload interface{}, priority string) (*asynq.TaskInfo, error)

EnqueueDatabaseOperation enqueues a database operation task

func (*Client) EnqueueMikrotikCommand

func (c *Client) EnqueueMikrotikCommand(ctx context.Context, action string, payload interface{}, priority string) (*asynq.TaskInfo, error)

EnqueueMikrotikCommand enqueues a MikroTik command task

func (*Client) EnqueueTask

func (c *Client) EnqueueTask(ctx context.Context, taskType string, payload interface{}, priority string, opts ...asynq.Option) (*asynq.TaskInfo, error)

EnqueueTask enqueues a generic task

type DatabaseOperationPayload

type DatabaseOperationPayload struct {
	Ip      string          `json:"ip"`
	System  string          `json:"system"`
	Action  string          `json:"action"`
	Payload json.RawMessage `json:"payload"`
}

type DatabaseQueueHandler

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

DatabaseQueueHandler handles database-related tasks.

func NewDatabaseQueueHandler

func NewDatabaseQueueHandler(wsHub *websocket.Hub) *DatabaseQueueHandler

NewDatabaseQueueHandler creates a new DatabaseQueueHandler.

func (*DatabaseQueueHandler) HandleTask

func (h *DatabaseQueueHandler) HandleTask(ctx context.Context, task *asynq.Task) error

HandleTask processes database-related tasks.

type ErrorHandlerFunc

type ErrorHandlerFunc func(ctx context.Context, task *asynq.Task, err error) (noRetry bool)

type GenericTaskPayload

type GenericTaskPayload struct {
	System  string          `json:"system"`
	Action  string          `json:"action"`
	Payload json.RawMessage `json:"payload"` // Raw JSON, to be decoded later
	Ip      string          `json:"ip"`      // IP address related to the command (e.g., user IP)
}

type Handler

type Handler interface {
	HandleTask(ctx context.Context, task *asynq.Task) error
}

Handler interface

type Handlers

type Handlers struct {
	MikrotikQueueHandler MikrotikQueueHandler // Use the struct directly, not the pointer
	DatabaseQueueHandler DatabaseQueueHandler // Use the struct directly, not the pointer

}

Handlers struct to hold all handler instances.

type MikrotikCommandPayload

type MikrotikCommandPayload struct {
	DeviceID    string   `json:"device_id"`    // ID of the MikroTik device
	Command     string   `json:"command"`      // Command to execute on the device
	Args        []string `json:"args"`         // Arguments for the command
	Ip          string   `json:"ip"`           // IP address related to the command (e.g., user IP)
	CallbackURL string   `json:"callback_url"` // Optional URL for callback after command execution
}

Define payload structs

type MikrotikQueueHandler

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

MikrotikQueueHandler handles MikroTik related tasks.

func NewMikrotikQueueHandler

func NewMikrotikQueueHandler(mikroTikService *service.MikroTikMangerService, wsHub *websocket.Hub) *MikrotikQueueHandler

func (*MikrotikQueueHandler) HandleTask

func (h *MikrotikQueueHandler) HandleTask(ctx context.Context, task *asynq.Task) error

type Server

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

Server is the server for processing tasks

func NewServer

func NewServer(redisAddr string, manager *mikrotik.Manager, wsHub *websocket.Hub, handlers *Handlers) (*Server, error)

NewServer creates a new queue server

func (*Server) GracefullyShutdown

func (s *Server) GracefullyShutdown()

GracefullyShutdown gracefully shuts down the queue server

func (*Server) Start

func (s *Server) Start() error

Start starts the queue server

func (*Server) Stop

func (s *Server) Stop()

Stop stops the queue server