cabbage

package
v0.0.0-...-5388c9c Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 27, 2024 License: MIT Imports: 13 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type CabbageBroker 

type CabbageBroker interface {
	SendCabbageMessage(queueName string, cbMessage *CabbageMessage) error
	GetCabbageMessage(queueName string) (*CabbageMessage, error)
	EnableQueueForWorker(queueName string) error
	Close()
}

CabbageBroker is interface for cabbage broker db

type CabbageClient 

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

CabbageClient provides API for sending cabbage tasks

func NewCabbageClient 

func NewCabbageClient(broker CabbageBroker) *CabbageClient

NewCabbageClient create new CabbageClient

func (*CabbageClient) Close 

func (cc *CabbageClient) Close()

Close connections

func (*CabbageClient) CreatePublisher 

func (cc *CabbageClient) CreatePublisher() *Publisher

CreatePublisher create publisher for publish data to broker

func (*CabbageClient) CreateScheduler 

func (cc *CabbageClient) CreateScheduler() *Scheduler

CreateScheduler create scheduler for schdule task

func (*CabbageClient) CreateWorker 

func (cc *CabbageClient) CreateWorker(queueName string, concurrency int) (*CabbageWorker, error)

CreateWorker create cabbage worker for cunsume data

func (*CabbageClient) RegisterTask 

func (cc *CabbageClient) RegisterTask(task *Task) error

RegisterTask register task for worker/publisher

type CabbageMessage 

type CabbageMessage struct {
	ID        string    `json:"id"`
	MessageId string    `json:"messageId"`
	Body      []byte    `json:"body"`
	TaskName  string    `json:"TaskName"`
	Timestamp time.Time `json:"timestamp"`
}

CabbageMessage base message for publish\consume

type CabbageWorker 

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

CabbageWorker represents distributed task worker

func (*CabbageWorker) RegisterTaskProccessersRoutes 

func (w *CabbageWorker) RegisterTaskProccessersRoutes(tr *TaskProccessersRoutes)

RegisterTaskProccessersRoutes register task proccessers routes

func (*CabbageWorker) RegisterTaskProcesser 

func (w *CabbageWorker) RegisterTaskProcesser(name string, tproccesser TaskProccesser)

RegisterTaskProcesser register task proccesser

func (*CabbageWorker) StartWorker 

func (w *CabbageWorker) StartWorker() error

StartWorker start cabbage worker

func (*CabbageWorker) StartWorkerWithContext 

func (w *CabbageWorker) StartWorkerWithContext(ctx context.Context) error

StartWorkerWithContext start cabbage worker with context

func (*CabbageWorker) StopWait 

func (w *CabbageWorker) StopWait()

StopWait waits for cabbage workers to terminate

func (*CabbageWorker) StopWorker 

func (w *CabbageWorker) StopWorker()

StopWorker stops cabbage workers

type ConsumingChannels 

type ConsumingChannels map[string]<-chan amqp.Delivery

type Entries 

type Entries []*Entry

type Entry 

type Entry struct {
	Schedule string
}

Entry schedule cron syntax

func EntryEvery 

func EntryEvery(minutes []int, hours []int, days []int, months []int, weekDays []int) (*Entry, error)

EntryEvery generate *Entry by minutes, hours, days, months, week days slices

func EntryEveryHour 

func EntryEveryHour(hour int) (*Entry, error)

EntryEveryHour create *Entry with every hour cron

func EntryEveryHourAtMinute 

func EntryEveryHourAtMinute(minute int, hour int) (*Entry, error)

EntryEveryHourAtMinute create *Entry with every hour and minute cron

func EntryEveryMinute 

func EntryEveryMinute(minute int) (*Entry, error)

EntryEveryMinute create *Entry with every minute cron

type Publisher 

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

Publisher cabbage task publisher

func (*Publisher) PublishTask 

func (p *Publisher) PublishTask(taskName string, tpublisher TaskPublisher) error

PublishTask publish task to broker

func (*Publisher) RegisterTask 

func (p *Publisher) RegisterTask(task *Task)

RegisterTask register task in publisher

type RabbitMQBroker 

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

RabbitMQBroker implement rabbtimq broker

func NewRabbitMQBroker 

func NewRabbitMQBroker(url string, rate int) (*RabbitMQBroker, error)

NewRabbitMQBroker constructor for RabbitmqBroker

func (*RabbitMQBroker) Close 

func (b *RabbitMQBroker) Close()

Close close broker connections

func (*RabbitMQBroker) EnableQueueForWorker 

func (b *RabbitMQBroker) EnableQueueForWorker(queueName string) error

EnableQueueForWorker create queue and start consume from rabbitmq broker

func (*RabbitMQBroker) GetCabbageMessage 

func (b *RabbitMQBroker) GetCabbageMessage(queueName string) (*CabbageMessage, error)

GetCabbageMessage get cabbage message from broker

func (*RabbitMQBroker) SendCabbageMessage 

func (b *RabbitMQBroker) SendCabbageMessage(queueName string, cbMessage *CabbageMessage) error

SendCabbageMessage send cabbage message to broker

type RabbitMQQueue 

type RabbitMQQueue struct {
	Name       string
	Durable    bool
	AutoDelete bool
}

RabbitMQQueue queue for rabbitmq

type RedisBroker 

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

RedisBroker is cabbage broker for redis

func NewRedisBroker 

func NewRedisBroker(url string) (*RedisBroker, error)

NewRedisBroker creates with given redis connection

func NewRedisBrokerWithContext 

func NewRedisBrokerWithContext(ctx context.Context, url string) (*RedisBroker, error)

NewRedisBroker creates with given redis connection with context

func (*RedisBroker) Close 

func (b *RedisBroker) Close()

Close redis broker

func (*RedisBroker) EnableQueueForWorker 

func (b *RedisBroker) EnableQueueForWorker(queueName string) error

EnableQueueForWorker ...

func (*RedisBroker) GetCabbageMessage 

func (b *RedisBroker) GetCabbageMessage(queueName string) (*CabbageMessage, error)

GetCabbageMessage get cabbage message from redis broker

func (*RedisBroker) SendCabbageMessage 

func (b *RedisBroker) SendCabbageMessage(queueName string, cbMessage *CabbageMessage) error

SendCabbageMessage send cabbage message to redis broker

type ScheduleTask 

type ScheduleTask struct {
	Name      string
	QueueName string
	Func      func() (tpublisher TaskPublisher)
	Entries   Entries // cron schedule slice of Entry
}

ScheduleTask task scheduler

type Scheduler 

type Scheduler struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Scheduler cabbage jobs scheduler

func (*Scheduler) AddScheduleTask 

func (shd *Scheduler) AddScheduleTask(shTask *ScheduleTask) error

AddScheduleTask add ScheduleTask to scheduler

func (*Scheduler) AddScheduleTasks 

func (shd *Scheduler) AddScheduleTasks(shTasks []*ScheduleTask) error

AddScheduleTasks add slice of ScheduleTasks to scheduler

func (*Scheduler) Shutdown 

func (shd *Scheduler) Shutdown()

Shutdown scheduler

func (*Scheduler) Start 

func (shd *Scheduler) Start() chan bool

Start scheduler

func (*Scheduler) StartWithContext 

func (shd *Scheduler) StartWithContext(ctx context.Context) chan bool

type Task 

type Task struct {
	Name        string
	QueueName   string
	TProccesser TaskProccesser
	WithPublish bool
}

Task cabbage task struct

func NewTask 

func NewTask(name string, queueName string, tproccesser TaskProccesser, withPublish bool) (*Task, error)

NewTask construct cabbage Task

type TaskProccesser 

type TaskProccesser interface {
	ProccessTask(ctx context.Context, body []byte, ID string) error
}

TaskProccesser interface for proccess consume message

type TaskProccessersRoutes 

type TaskProccessersRoutes struct {
	Routes map[string]TaskProccesser
}

TaskProccessersRoutes router for workers consume

func NewTaskProccessersRoutes 

func NewTaskProccessersRoutes(routes map[string]TaskProccesser) *TaskProccessersRoutes

NewTaskProccessersRoutes construct TaskProccessersRoutes

type TaskPublisher 

type TaskPublisher interface {
	ToPublish() ([]byte, error)
}

TaskPublisher interface for publish message

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.