queue

README

Queue component of Enorith

Usage

Register connection

queue.DefaultManager.RegisterConnection("nsq", func() (contracts.Connection, error) {
    return connections.NewNsqFromConfig(connections.NsqConfig{
        Nsqd: "127.0.0.1:4150",
    }), nil
})

Register worker

c, _ := queue.DefaultManager.GetConnection("nsq")
queue.DefaultManager.RegisterWorker("nsq", std.NewWorker(4, c))

Run worker

done := make(chan struct{}, 1)
queue.DefaultManager.Work(done, "nsq")

exit := make(chan os.Signal, 1)
signal.Notify(exit, os.Interrupt, syscall.SIGTERM, syscall.SIGINT)
<-exit
done <- struct{}{}

Documentation

Index

• Variables
• type ConnectionRegister
• type Dispatcher
  • func (d Dispatcher) After(delay time.Duration) *PayloadHolder
  • func (d Dispatcher) Dispatch(payload interface{}) error
  • func (d Dispatcher) On(on string) *PayloadHolder
• type Manager
  • func NewManager() *Manager
  • func (m *Manager) Close(workers ...string)
  • func (m *Manager) GetConnection(connection string) (contracts.Connection, error)
  • func (m *Manager) GetWorker(worker string) (contracts.Worker, bool)
  • func (m *Manager) RegisterConnection(connection string, cr ConnectionRegister)
  • func (m *Manager) RegisterWorker(name string, worker contracts.Worker)
  • func (m *Manager) Work(done chan struct{}, workers ...string)
• type PayloadHolder
  • func (ph *PayloadHolder) After(delay time.Duration) *PayloadHolder
  • func (ph *PayloadHolder) Dispatch(payload ...interface{}) error
  • func (ph *PayloadHolder) On(on string) *PayloadHolder

Variables

var DefaultManager = NewManager()

Types

type ConnectionRegister

type ConnectionRegister func() (contracts.Connection, error)

type Dispatcher

type Dispatcher struct {
	DefaultConnection string
}

var DefaultDispatcher Dispatcher

func (Dispatcher) After

func (d Dispatcher) After(delay time.Duration) *PayloadHolder

func (Dispatcher) Dispatch

func (d Dispatcher) Dispatch(payload interface{}) error

func (Dispatcher) On

func (d Dispatcher) On(on string) *PayloadHolder

type Manager

type Manager struct {
	ErrorHandler contracts.ErrorHandler
	// contains filtered or unexported fields
}

func NewManager

func NewManager() *Manager

func (*Manager) Close

func (m *Manager) Close(workers ...string)

func (*Manager) GetConnection

func (m *Manager) GetConnection(connection string) (contracts.Connection, error)

GetConnection, get queue connection

func (*Manager) GetWorker

func (m *Manager) GetWorker(worker string) (contracts.Worker, bool)

GetWorker, get queue worker

func (*Manager) RegisterConnection

func (m *Manager) RegisterConnection(connection string, cr ConnectionRegister)

RegisterConnection, register queue connection

func (*Manager) RegisterWorker

func (m *Manager) RegisterWorker(name string, worker contracts.Worker)

RegisterWorker, register queue worker

func (*Manager) Work

func (m *Manager) Work(done chan struct{}, workers ...string)

Work, run worker process

type PayloadHolder

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

func (*PayloadHolder) After

func (ph *PayloadHolder) After(delay time.Duration) *PayloadHolder

func (*PayloadHolder) Dispatch

func (ph *PayloadHolder) Dispatch(payload ...interface{}) error

func (*PayloadHolder) On

func (ph *PayloadHolder) On(on string) *PayloadHolder