client

Documentation

Index

• Constants
• Variables
• type Client
  • func Dial(srv *Server, password string) (*Client, error)
  • func Open() (*Client, error)
  • func (c *Client) Ack(jid string) error
  • func (c *Client) Beat() (string, error)
  • func (c *Client) Close() error
  • func (c *Client) Fail(jid string, err error, backtrace []byte) error
  • func (c *Client) Fetch(q ...string) (*Job, error)
  • func (c *Client) Flush() error
  • func (c *Client) Generic(cmdline string) (string, error)
  • func (c *Client) Info() (map[string]interface{}, error)
  • func (c *Client) Push(job *Job) error
• type ClientData
• type Failure
• type Job
  • func NewJob(jobtype string, args ...interface{}) *Job
  • func (j *Job) GetCustom(name string) (interface{}, bool)
  • func (j *Job) SetCustom(name string, value interface{})
• type ProtocolError
  • func (pe *ProtocolError) Error() string
• type Server
  • func DefaultServer() *Server
  • func (s *Server) Open() (*Client, error)
  • func (s *Server) ReadFromEnv() error

Constants

const (
	// This is the protocol version supported by this client.
	// The server might be running an older or newer version.
	ExpectedProtocolVersion = 2
)

Variables

var (
	Name      = "Faktory"
	License   = "GPLv3"
	Licensing = "Licensed under the GNU Public License 3.0"
	Version   = "0.9.6"
)

var (
	// Set this to a non-empty value in a consumer process
	// e.g. see how faktory_worker_go sets this.
	RandomProcessWid = ""
)

Types

type Client

type Client struct {
	Location string
	Options  *ClientData
	// contains filtered or unexported fields
}

The Client structure represents a thread-unsafe connection to a Faktory server. It is recommended to use a connection pool of Clients in a multi-threaded process. See faktory_worker_go's internal connection pool for example.

TODO Provide a connection pool as part of this package?

func Dial

func Dial(srv *Server, password string) (*Client, error)

Dial connects to the remote faktory server.

client.Dial(client.Localhost, "topsecret")

func Open

func Open() (*Client, error)

Open connects to a Faktory server based on environment variable conventions:

• Use FAKTORY_PROVIDER to point to a custom URL variable. • Use FAKTORY_URL as a catch-all default.

Use the URL to configure any necessary password:

tcp://:mypassword@localhost:7419

By default Open assumes localhost with no password which is appropriate for local development.

func (*Client) Ack

func (c *Client) Ack(jid string) error

func (*Client) Beat

func (c *Client) Beat() (string, error)

func (*Client) Close

func (c *Client) Close() error

func (*Client) Fail

func (c *Client) Fail(jid string, err error, backtrace []byte) error

Fail notifies Faktory that a job failed with the given error. If backtrace is non-nil, it is assumed to be the output from runtime/debug.Stack().

func (*Client) Fetch

func (c *Client) Fetch(q ...string) (*Job, error)

func (*Client) Flush

func (c *Client) Flush() error

func (*Client) Generic

func (c *Client) Generic(cmdline string) (string, error)

func (*Client) Info

func (c *Client) Info() (map[string]interface{}, error)

func (*Client) Push

func (c *Client) Push(job *Job) error

type ClientData

type ClientData struct {
	Hostname string   `json:"hostname"`
	Wid      string   `json:"wid"`
	Pid      int      `json:"pid"`
	Labels   []string `json:"labels"`
	// Hash is hex(sha256(password + nonce))
	PasswordHash string `json:"pwdhash"`
	// The protocol version used by this client.
	// The server can reject this connection if the version will not work
	// The server advertises its protocol version in the HI.
	Version int `json:"v"`
}

ClientData is serialized to JSON and sent with the HELLO command. PasswordHash is required if the server is not listening on localhost. The WID (worker id) must be random and unique for each worker process. It can be a UUID, etc. Non-worker processes should leave WID empty.

The other elements can be useful for debugging and are displayed on the Busy tab.

type Failure

type Failure struct {
	RetryCount   int      `json:"retry_count"`
	FailedAt     string   `json:"failed_at"`
	NextAt       string   `json:"next_at,omitempty"`
	ErrorMessage string   `json:"message,omitempty"`
	ErrorType    string   `json:"errtype,omitempty"`
	Backtrace    []string `json:"backtrace,omitempty"`
}

type Job

type Job struct {
	// required
	Jid   string        `json:"jid"`
	Queue string        `json:"queue"`
	Type  string        `json:"jobtype"`
	Args  []interface{} `json:"args"`

	// optional
	CreatedAt  string                 `json:"created_at,omitempty"`
	EnqueuedAt string                 `json:"enqueued_at,omitempty"`
	At         string                 `json:"at,omitempty"`
	ReserveFor int                    `json:"reserve_for,omitempty"`
	Retry      int                    `json:"retry,omitempty"`
	Backtrace  int                    `json:"backtrace,omitempty"`
	Failure    *Failure               `json:"failure,omitempty"`
	Custom     map[string]interface{} `json:"custom,omitempty"`
}

func NewJob

func NewJob(jobtype string, args ...interface{}) *Job

func (*Job) GetCustom

func (j *Job) GetCustom(name string) (interface{}, bool)

func (*Job) SetCustom

func (j *Job) SetCustom(name string, value interface{})

type ProtocolError

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

func (*ProtocolError) Error

func (pe *ProtocolError) Error() string

type Server

type Server struct {
	Network  string
	Address  string
	Password string
	Timeout  time.Duration
	TLS      *tls.Config
}

func DefaultServer

func DefaultServer() *Server

func (*Server) Open

func (s *Server) Open() (*Client, error)

func (*Server) ReadFromEnv

func (s *Server) ReadFromEnv() error