README
¶
PGQ consumers in GO
Package for writing PGQ consumers in Golang.
Installation
go get github.com/sasha-alias/pgq
Description
There are two types of consumers you can create:
- pgq.Consumer - a single instance consumer
- pgq.CoopConsumer - a cooperative consumer, running several subconsumers in goroutines
In order to implement own event or batch processing you have to define the following interfaces:
func MyEventHandler(event pgq.Event) error {}
func MyBatchHandler(event []pgq.Event) error {}
Then create an instance of appropriate consumer and pass your function to it via EventHandler or BatchHandler methods.
Consumer
package main
import (
"github.com/sasha-alias/pgq"
"log"
)
// Define own event handler
func ProcessEvent(event pgq.Event) error {
log.Printf("%+v", event)
return nil
}
func main() {
consumer, _ := pgq.NewConsumer("consumer_name", "queue_name", "postgresql connect string") // Create consumer
consumer.EventHandler(ProcessEvent) // Set the event handler you defined before
consumer.Work() // Start events processing
}
Cooperative consumer
package main
import (
"github.com/sasha-alias/pgq"
"log"
)
func ProcessEvent(event pgq.Event) error {
log.Printf("%+v", event)
return nil
}
func main() {
consumer, _ := pgq.NewCoopConsumer("consumer_name", number_of_subconsumers, "queue_name", "postgresql connect string")
consumer.EventHandler(ProcessEvent)
consumer.Work()
}
Documentation
¶
Overview ¶
Package pgq provides an easy framework for writing consumers for PGQ. PGQ is an implementation of transactional queues on top of Postgresql.
Index ¶
- type Consumer
- func (self *Consumer) BatchHandler(handler func([]Event) error)
- func (self *Consumer) BatchId() int64
- func (self *Consumer) CloseBatch() error
- func (self *Consumer) ConsumerName() string
- func (self *Consumer) EventHandler(handler func(Event) error)
- func (self *Consumer) QueueName() string
- func (self *Consumer) SleepInterval(interval time.Duration)
- func (self *Consumer) SubconsumerName() string
- func (self *Consumer) Subscribe() error
- func (self *Consumer) Unsubscribe() error
- func (self *Consumer) Work() error
- type CoopConsumer
- func (self *CoopConsumer) BatchHandler(handler func([]Event) error)
- func (self *CoopConsumer) BatchId() int64
- func (self *CoopConsumer) CloseBatch() error
- func (self *CoopConsumer) ConsumerId() string
- func (self *CoopConsumer) ConsumerName() string
- func (self *CoopConsumer) EventHandler(handler func(Event) error)
- func (self *CoopConsumer) QueueName() string
- func (self *CoopConsumer) SleepInterval(interval time.Duration)
- func (self *CoopConsumer) SubconsumerName() string
- func (self *CoopConsumer) Subscribe() error
- func (self *CoopConsumer) Unsubscribe() error
- func (self *CoopConsumer) Work() error
- type Event
- type IConsumer
- type Subconsumer
- func (self *Subconsumer) BatchHandler(handler func([]Event) error)
- func (self *Subconsumer) BatchId() int64
- func (self *Subconsumer) CloseBatch() error
- func (self *Subconsumer) ConsumerName() string
- func (self *Subconsumer) EventHandler(handler func(Event) error)
- func (self *Subconsumer) QueueName() string
- func (self *Subconsumer) SleepInterval(interval time.Duration)
- func (self *Subconsumer) SubconsumerName() string
- func (self *Subconsumer) Subscribe() error
- func (self *Subconsumer) Unsubscribe() error
- func (self *Subconsumer) Work() error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
func (*Consumer) BatchHandler ¶
Set batch handler
func (*Consumer) EventHandler ¶
Set event handler
func (*Consumer) SleepInterval ¶
Set sleep interval
type CoopConsumer ¶
type CoopConsumer struct {
Count int
// contains filtered or unexported fields
}
Cooperative consumer
func (*CoopConsumer) BatchHandler ¶
func (self *CoopConsumer) BatchHandler(handler func([]Event) error)
Set batch handler
func (*CoopConsumer) CloseBatch ¶
func (self *CoopConsumer) CloseBatch() error
func (*CoopConsumer) EventHandler ¶
func (self *CoopConsumer) EventHandler(handler func(Event) error)
Set event handler
func (*CoopConsumer) SleepInterval ¶
func (self *CoopConsumer) SleepInterval(interval time.Duration)
Set sleep interval
func (*CoopConsumer) SubconsumerName ¶
func (self *CoopConsumer) SubconsumerName() string
Subconsumer name
func (*CoopConsumer) Subscribe ¶
func (self *CoopConsumer) Subscribe() error
Register consumer on the queue
func (*CoopConsumer) Unsubscribe ¶
func (self *CoopConsumer) Unsubscribe() error
Unregister consumer from the queue
type Event ¶
type Event struct {
Ev_id int64
Ev_time time.Time
Ev_txid int64
Ev_retry sql.NullInt64
Ev_type sql.NullString
Ev_data sql.NullString
Ev_extra1 sql.NullString
Ev_extra2 sql.NullString
Ev_extra3 sql.NullString
Ev_extra4 sql.NullString
Consumer IConsumer
}
type IConsumer ¶
type IConsumer interface {
Subscribe() error
Unsubscribe() error
EventHandler(handler func(event Event) error)
BatchHandler(handler func(batch []Event) error)
Work() error
SleepInterval(time.Duration)
BatchId() int64
ConsumerName() string
SubconsumerName() string
QueueName() string
CloseBatch() error
}
func NewConsumer ¶
Register consumer in pgq
type Subconsumer ¶
type Subconsumer struct {
// contains filtered or unexported fields
}
Subconsumer
func (*Subconsumer) BatchHandler ¶
func (self *Subconsumer) BatchHandler(handler func([]Event) error)
Set batch handler
func (*Subconsumer) EventHandler ¶
func (self *Subconsumer) EventHandler(handler func(Event) error)
Set event handler
func (*Subconsumer) SleepInterval ¶
func (self *Subconsumer) SleepInterval(interval time.Duration)
Set sleep interval
func (*Subconsumer) SubconsumerName ¶
func (self *Subconsumer) SubconsumerName() string
Subconsumer name
func (*Subconsumer) Unsubscribe ¶
func (self *Subconsumer) Unsubscribe() error
Unregister subconsumer
Source Files
Click to show internal directories.
Click to hide internal directories.