server

Documentation

Overview

Dispatch operations:

• Basic dequeue
• Connection sends POP q1, q2
• Job moved from Queue into Working
• Scheduled
• Job Pushed into Queue
• Job moved from Queue into Working
• Failure
• Job Pushed into Retries
• Push
• Job Pushed into Queue
• Ack
• Job removed from Working

How are jobs passed to waiting workers?

Socket sends "POP q1, q2, q3" Connection pops each queue:

store.GetQueue("q1").Pop()

and returns if it gets any non-nil data.

If all nil, the connection registers itself, blocking for a job.

Index

• Variables
• func CurrentState(s *Server) (map[string]interface{}, error)
• func OnStart(x func(*Server) error)
• type ClientWorker
  • func (worker *ClientWorker) BusyCount() int
  • func (worker *ClientWorker) IsQuiet() bool
  • func (worker *ClientWorker) Signal(sig string)
• type Connection
  • func (c *Connection) Close()
  • func (c *Connection) Error(cmd string, err error) error
  • func (c *Connection) Identity() string
  • func (c *Connection) Number(val int) error
  • func (c *Connection) Ok() error
  • func (c *Connection) Result(msg []byte) error
• type FailPayload
• type Reservation
• type RuntimeStats
• type Server
  • func NewServer(opts *ServerOptions) (*Server, error)
  • func (s *Server) Fail(jid, msg, errtype string, backtrace []string) error
  • func (s *Server) Fetch(fn func(*faktory.Job) error, ctx context.Context, queues ...string) (*faktory.Job, error)
  • func (s *Server) Heartbeats() map[string]*ClientWorker
  • func (s *Server) Start() error
  • func (s *Server) Stop(f func())
  • func (s *Server) Store() storage.Store
• type ServerOptions
• type TimedSet

Variables

var (
	DefaultTimeout = 1800
)

var (
	EventHandlers = make([]func(*Server) error, 0)
)

Functions

func CurrentState

func CurrentState(s *Server) (map[string]interface{}, error)

func OnStart

func OnStart(x func(*Server) error)

register a global handler to be called when the Server instance has finished booting but before it starts listening.

Types

type ClientWorker

type ClientWorker struct {
	Hostname     string   `json:"hostname"`
	Wid          string   `json:"wid"`
	Pid          int      `json:"pid"`
	Labels       []string `json:"labels"`
	Salt         string   `json:"salt"`
	PasswordHash string   `json:"pwdhash"`
	StartedAt    time.Time
	// contains filtered or unexported fields
}

func (*ClientWorker) BusyCount

func (worker *ClientWorker) BusyCount() int

func (*ClientWorker) IsQuiet

func (worker *ClientWorker) IsQuiet() bool

func (*ClientWorker) Signal

func (worker *ClientWorker) Signal(sig string)

* Send "quiet" or "terminate" to the given client * worker process. Other signals are undefined.

type Connection

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

* Represents a connection to a faktory client. * * faktory reuses the same wire protocol as Redis: RESP. * It's a nice trade-off between human-readable and efficient. * Shout out to antirez for his nice design document on it. * * https://redis.io/topics/protocol

func (*Connection) Close

func (c *Connection) Close()

func (*Connection) Error

func (c *Connection) Error(cmd string, err error) error

func (*Connection) Identity

func (c *Connection) Identity() string

func (*Connection) Number

func (c *Connection) Number(val int) error

func (*Connection) Ok

func (c *Connection) Ok() error

func (*Connection) Result

func (c *Connection) Result(msg []byte) error

type FailPayload

type FailPayload struct {
	Jid          string   `json:"jid"`
	ErrorMessage string   `json:"message"`
	ErrorType    string   `json:"errtype"`
	Backtrace    []string `json:"backtrace"`
}

type Reservation

type Reservation struct {
	Job    *faktory.Job `json:"job"`
	Since  string       `json:"reserved_at"`
	Expiry string       `json:"expires_at"`
	Wid    string       `json:"wid"`
	// contains filtered or unexported fields
}

type RuntimeStats

type RuntimeStats struct {
	Processed   int64
	Failures    int64
	Connections int64
	Commands    int64
	StartedAt   time.Time
}

type Server

type Server struct {
	Options   *ServerOptions
	Stats     *RuntimeStats
	TLSConfig *tls.Config
	Password  string
	// contains filtered or unexported fields
}

func NewServer

func NewServer(opts *ServerOptions) (*Server, error)

func (*Server) Fail

func (s *Server) Fail(jid, msg, errtype string, backtrace []string) error

func (*Server) Fetch

func (s *Server) Fetch(fn func(*faktory.Job) error, ctx context.Context, queues ...string) (*faktory.Job, error)

func (*Server) Heartbeats

func (s *Server) Heartbeats() map[string]*ClientWorker

func (*Server) Start

func (s *Server) Start() error

func (*Server) Stop

func (s *Server) Stop(f func())

func (*Server) Store

func (s *Server) Store() storage.Store

type ServerOptions

type ServerOptions struct {
	Binding          string
	StorageDirectory string
	ConfigDirectory  string
	Environment      string
	DisableTls       bool
}

type TimedSet

type TimedSet interface {
	AddElement(string, string, []byte) error
	RemoveElement(string, string) error
}