core

Documentation

Index

• Constants
• Variables
• func Discard(rdr io.Reader, b []byte) ([]byte, error)
• func NewBuryArg(data *CmdData) (*buryArg, error)
• func NewClient(useTube state.TubeName) *client
• func NewIDArg(data *CmdData) (*idArg, error)
• func NewKickNArg(data *CmdData) (*kickNArg, error)
• func NewPutArg(data *CmdData) (*putArg, error)
• func NewReleaseArg(data *CmdData) (*releaseArg, error)
• func NewReserveWithTimeoutArg(data *CmdData) (*reserveWithTimeoutArg, error)
• func NewTubeArg(data *CmdData) (*tubeArg, error)
• func Scan(rdr io.Reader, b []byte, limitBytes int) ([]byte, []byte, error)
• func ScanCmdLine(rdr io.Reader, b []byte) ([]byte, []byte, error)
• func ScanJobData(rdr io.Reader, b []byte, maxJobDataSizeBytes int) ([]byte, []byte, error)
• type ClientReg
  • func (cr ClientReg) String() string
• type ClientSet
  • func (cs ClientSet) Contains(c *client) bool
  • func (cs ClientSet) Find(clientID state.ClientID) (*client, error)
  • func (cs ClientSet) Len() int
  • func (cs ClientSet) Random() (*client, error)
  • func (cs ClientSet) Remove(c *client) error
  • func (cs ClientSet) Set(c *client) error
• type CmdData
  • func ParseCommandLine(cmdLine string, maxJobDataSizeBytes int) (*CmdData, error)
  • func (c CmdData) String() string
• type CmdRequest
  • func NewCmdRequest(cmdData *CmdData, clientID state.ClientID) (CmdRequest, error)
  • func (req CmdRequest) String() string
• type CmdResponse
  • func NewCmdResponseFromReq(req *CmdRequest) *CmdResponse
• type CmdType
  • func (i CmdType) String() string
• type CommandProcessor
  • func NewCommandProcessor(jsm state.JSM, cfg *Config) CommandProcessor
• type Config

Constants

const (
	TickDuration = 1000 * time.Millisecond

	// Max. Command size in bytes (inclusive of 2 byte delimiter)
	MaxCmdSizeBytes = 226

	// Max. Job Data size in bytes (inclusive of 2 byte delimiter)
	MaxJobDataSizeBytes = (128 * 1024) + 2
)

const (
	// Error message response indicating an internal server error. Typically, indicative
	// of a bug in the implementation.
	MsgInternalError = "INTERNAL_ERROR"

	// The client sent a command line that was not well-formed.
	// This can happen if the line's length exceeds 224 bytes including \r\n,
	// if the name of a tube exceeds 200 bytes, if non-numeric
	// characters occur where an integer is expected, if the wrong number of
	// arguments are present, or if the command line is mal-formed in any other way.
	MsgBadFormat = "BAD_FORMAT"

	// The client sent a command that the server does not know.
	MsgUnknownCommand = "UNKNOWN_COMMAND"

	// Error message if the client attempts to ignore the only tube in its watch list.
	MsgCannotIgnoreTube = "NOT_IGNORED"

	// Error message to indicate if a reservation request timed out
	MsgTimedOut = "TIMED_OUT"

	// Error message to indicate if a reservation request is in DeadlineSoon
	MsgDeadlineSoon = "DEADLINE_SOON"

	// Error message to indicate if the entity (job etc) cannot be found
	MsgNotFound = "NOT_FOUND"

	// Error message to indicate that the job is too big
	MsgJobTooBig = "JOB_TOO_BIG"

	// Error message to indicate that CRLF is needed
	MsgExpectCRLF = "EXPECTED_CRLF"
)

Variables

var (
	// ErrNoData - when the input stream has no data this can be the case
	// if the underlying reader return zero bytes  and is however not at EOF
	ErrNoData = errors.New("no data")

	// ErrDelimiterMissing - when the input stream has no newlines (\r\n)
	ErrDelimiterMissing = errors.New("delimiter (\\r\\n) missing in command")

	// ErrBadFormat The client sent a command line that was not well-formed.
	// This can happen if the line's length exceeds 224 bytes including \r\n,
	// if the name of a tube exceeds 200 bytes, if non-numeric
	// characters occur where an integer is expected, if the wrong number of
	// arguments are present, or if the command line is mal-formed in any other
	ErrBadFormat = errors.New("bad format command")

	// ErrJobSizeTooBig he client has requested to put a job with a body larger that the configured limit
	ErrJobSizeTooBig = errors.New("job size too big")

	// ErrCmdTokensMissing - when the provided command has no tokens
	ErrCmdTokensMissing = errors.New("bad command, cannot find atleast one token")

	// ErrCmdNotFound - the provided command is not found or supported
	ErrCmdNotFound = errors.New("command not found")
)

var (
	// Delimiter for commands and data
	DelimRe = regexp.MustCompile(`\r\n`)
)

Functions

func Discard

func Discard(rdr io.Reader, b []byte) ([]byte, error)

func NewBuryArg

func NewBuryArg(data *CmdData) (*buryArg, error)

func NewClient

func NewClient(useTube state.TubeName) *client

func NewIDArg

func NewIDArg(data *CmdData) (*idArg, error)

func NewKickNArg

func NewKickNArg(data *CmdData) (*kickNArg, error)

func NewPutArg

func NewPutArg(data *CmdData) (*putArg, error)

func NewReleaseArg

func NewReleaseArg(data *CmdData) (*releaseArg, error)

func NewReserveWithTimeoutArg

func NewReserveWithTimeoutArg(data *CmdData) (*reserveWithTimeoutArg, error)

func NewTubeArg

func NewTubeArg(data *CmdData) (*tubeArg, error)

func Scan

func Scan(rdr io.Reader, b []byte, limitBytes int) ([]byte, []byte, error)

func ScanCmdLine

func ScanCmdLine(rdr io.Reader, b []byte) ([]byte, []byte, error)

ScanCmdLine - Scans the provided "b" byte slice in search of a newline (\\r\\n) delimiter. If the provided slice does not have a newline, then scan the reader upto "MaxCmdSizeBytes-len(b)" bytes in search of the delimiter. Returns a triple of command, extra byte slice and an error The extra byte slice are any additional bytes read after encountering the delimiter.

func ScanJobData

func ScanJobData(rdr io.Reader, b []byte, maxJobDataSizeBytes int) ([]byte, []byte, error)

ScanJobData - Scans the provided "b" byte slice in search of a newline (\\r\\n) delimiter. If the provided slice does'nt have a newline, then scan the reader upto "maxJobDataSizeBytes-len(b)" bytes in search of a delimiter Returns a triple of command, extra byte slice and an error The extra byte slice are any additional bytes read after encountering the delimiter.

Types

type ClientReg

type ClientReg struct {
	// A unique identifier for this client
	ID state.ClientID

	// Responce channel on which all responses are sent to this client
	ResponseCh <-chan CmdResponse

	// Represents an error, encountered during registration
	Error error
}

Encapsulates a Client Registration information

func (ClientReg) String

func (cr ClientReg) String() string

type ClientSet

type ClientSet map[state.ClientID]*client

func (ClientSet) Contains

func (cs ClientSet) Contains(c *client) bool

func (ClientSet) Find

func (cs ClientSet) Find(clientID state.ClientID) (*client, error)

func (ClientSet) Len

func (cs ClientSet) Len() int

func (ClientSet) Random

func (cs ClientSet) Random() (*client, error)

func (ClientSet) Remove

func (cs ClientSet) Remove(c *client) error

func (ClientSet) Set

func (cs ClientSet) Set(c *client) error

type CmdData

type CmdData struct {
	CmdType  CmdType
	Args     string
	Data     []byte
	NeedData bool
}

func ParseCommandLine

func ParseCommandLine(cmdLine string, maxJobDataSizeBytes int) (*CmdData, error)

ParseCommandLine parses the command line string provided a connected client into a valid CmdData struct

func (CmdData) String

func (c CmdData) String() string

type CmdRequest

type CmdRequest struct {
	ID       string
	ClientID state.ClientID
	CmdType  CmdType
	// contains filtered or unexported fields
}

func NewCmdRequest

func NewCmdRequest(cmdData *CmdData, clientID state.ClientID) (CmdRequest, error)

func (CmdRequest) String

func (req CmdRequest) String() string

type CmdResponse

type CmdResponse struct {
	RequestID string
	ClientID  state.ClientID
	Response  []byte
	HasMore   bool
}

func NewCmdResponseFromReq

func NewCmdResponseFromReq(req *CmdRequest) *CmdResponse

type CmdType

type CmdType int

const (
	Unknown CmdType = iota
	Bury
	Delete
	Ignore
	Kick
	KickJob
	ListTubes
	ListTubeUser
	PauseTube
	Peek
	PeekBuried
	PeekDelayed
	PeekReady
	Put
	Quit
	Release
	Reserve
	ReserveJob
	ReserveWithTimeout
	Stats
	StatsJob
	StatsTube
	Touch
	Use
	Watch
	Max
)

func (CmdType) String

func (i CmdType) String() string

type CommandProcessor

type CommandProcessor interface {
	// Register a new client with this system
	// Returns an ID and error
	RegisterClient() *ClientReg

	// Dispatch this request to the command process
	DispatchRequest(request CmdRequest)

	// Run this processor
	Run()

	// Shutdown this processor
	Shutdown()

	// Return the maximum job data size in Bytes
	MaxJobDataSize() int
}

func NewCommandProcessor

func NewCommandProcessor(jsm state.JSM, cfg *Config) CommandProcessor

type Config

type Config struct {
	ListenAddr            string
	ListenPort            int
	UpstreamAddrs         string
	ConnectTimeout        int
	MaxJobSize            int
	MaxReservationTimeout int
}