server

Documentation

Index

• Constants
• Variables
• func RegisterTaskQ(h beehive.Hive, opts ...Option) error
• type Ack
• type AckHTTPHandler
  • func (h *AckHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)
• type AckHandler
  • func (h AckHandler) Map(msg beehive.Msg, ctx beehive.MapContext) beehive.MappedCells
  • func (h AckHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error
• type Acked
• type ConnHandler
  • func NewConnHandler(conn net.Conn) *ConnHandler
  • func (h *ConnHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error
  • func (h *ConnHandler) Start(ctx beehive.RcvContext)
  • func (h *ConnHandler) Stop(ctx beehive.RcvContext)
• type DeQHTTPHandler
  • func (h *DeQHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)
• type DeQHandler
  • func (h DeQHandler) Map(msg beehive.Msg, ctx beehive.MapContext) beehive.MappedCells
  • func (h DeQHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error
• type Deque
• type Dequed
• type EnQHTTPHandler
  • func (h *EnQHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)
• type EnQHandler
  • func (h EnQHandler) Map(msg beehive.Msg, ctx beehive.MapContext) beehive.MappedCells
  • func (h EnQHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error
• type Enque
• type Enqued
• type Error
  • func (e *Error) Error() string
• type Option
  • func Address(a string) Option
  • func MaxRate(r uint64) Option
  • func ReplicationFactor(f int) Option
• type ProtoHandler
  • func NewProtoHandler(addr string) (h *ProtoHandler, err error)
  • func (h *ProtoHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error
  • func (h *ProtoHandler) Start(ctx beehive.RcvContext)
  • func (h *ProtoHandler) Stop(ctx beehive.RcvContext)
• type Queue
• type ReqID
  • func (r ReqID) String() string
• type Task
• type TaskID
  • func (id TaskID) String() string
• type Timeout
• type TimeoutHandler
  • func (h TimeoutHandler) Map(msg beehive.Msg, ctx beehive.MapContext) beehive.MappedCells
  • func (h TimeoutHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error

Constants

const (
	ReqLen = 3 // ReqLen is length of a TaskQ request verb.
	ResLen = 5 // ResLen is length of a TaskQ response verb.
)

Variables

var (
	// ErrEmptyQueue is returned when dequeing an empty queue.
	ErrEmptyQueue = errors.New("taskq: no task in queue")
	// ErrNoSuchTask is returned when acking or accessing a task that does not
	// exist.
	ErrNoSuchTask = errors.New("taskq: no such task")
	// ErrInvalidCmd is returned by the protocol handler whenever an invalid
	// command is sent to the server.
	ErrInvalidCmd = errors.New("taskq: invalid command")
	// ErrNotEnoughData is returned by the protocol handler whenever the data sent
	// by the client is less than the length specificed.
	ErrNotEnoughData = errors.New("taskq: not enough data")
	// ErrNoNewLine is returned by the protocol handler whenever the request sent
	// by the client does not end in a new line.
	ErrNoNewLine = errors.New("taskq: new line expected")
)

var (
	ProtoReqEnQ = []byte("enq") // ProtoReqEnQ represents an enqueue.
	ProtoReqDeQ = []byte("deq") // ProtoReqDeQ represents a dequeue.
	ProtoReqAck = []byte("ack") // ProtoReqAck represents an acknowledgement.

	ProtoResEnQed = []byte("enqed") // ProtoResEnQed is a response to an enqueue.
	ProtoResDeQed = []byte("deqed") // ProtoResDeQed is a response to a dequeue.
	ProtoResAcked = []byte("acked") // ProtoResAcked is a response to an ack.
	ProtoResError = []byte("error") // ProtoResError is an error response.
)

Functions

func RegisterTaskQ

func RegisterTaskQ(h beehive.Hive, opts ...Option) error

RegisterTaskQ registers the TaskQ application and all its handler in the hive.

Types

type Ack

type Ack struct {
	ID     ReqID  // The client-assigned request ID.
	Queue  Queue  // The queue.
	TaskID TaskID // The task ID.
}

Ack represents a message emitted to acknowledge a previously dequeued task.

type AckHTTPHandler

type AckHTTPHandler httpHandler

AckHTTPHandler provides the HTTP endpoint for acknowledging tasks.

func (*AckHTTPHandler) ServeHTTP

func (h *AckHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)

type AckHandler

type AckHandler struct{}

AckHandler handles Ack messages.

func (AckHandler) Map

func (h AckHandler) Map(msg beehive.Msg,
	ctx beehive.MapContext) beehive.MappedCells

func (AckHandler) Rcv

func (h AckHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error

type Acked

type Acked struct {
	ID     ReqID  // The request ID.
	Queue  Queue  // The queue.
	TaskID TaskID // The ID of the acknowledged task.
}

Acked represents a message emitted as a reply to a successful Ack request.

type ConnHandler

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

ConnHandler is a detached handler that handles a single TaskQ connection.

func NewConnHandler

func NewConnHandler(conn net.Conn) *ConnHandler

NewConnHandler creates a new connection handler for the given connection.

func (*ConnHandler) Rcv

func (h *ConnHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error

func (*ConnHandler) Start

func (h *ConnHandler) Start(ctx beehive.RcvContext)

func (*ConnHandler) Stop

func (h *ConnHandler) Stop(ctx beehive.RcvContext)

type DeQHTTPHandler

type DeQHTTPHandler httpHandler

DeQHTTPHandler provides the HTTP endpoint for dequeuing tasks.

func (*DeQHTTPHandler) ServeHTTP

func (h *DeQHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)

type DeQHandler

type DeQHandler struct{}

DeQHandler handles Deque messages.

func (DeQHandler) Map

func (h DeQHandler) Map(msg beehive.Msg,
	ctx beehive.MapContext) beehive.MappedCells

func (DeQHandler) Rcv

func (h DeQHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error

type Deque

type Deque struct {
	ID    ReqID // The client-assigned request ID.
	Queue Queue // The queue to dequeue a task from.
}

Deque represents a message emitted to dequeue a task from a queue.

type Dequed

type Dequed struct {
	ID   ReqID // The request ID.
	Task Task  // The task that is dequeued.
}

Dequed represents a message emitted as a reply to a successful Deque request.

type EnQHTTPHandler

type EnQHTTPHandler httpHandler

EnQHTTPHandler provides the HTTP endpoint for enqueuing tasks.

func (*EnQHTTPHandler) ServeHTTP

func (h *EnQHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)

type EnQHandler

type EnQHandler struct{}

EnQHandler handles Enque messages.

func (EnQHandler) Map

func (h EnQHandler) Map(msg beehive.Msg,
	ctx beehive.MapContext) beehive.MappedCells

func (EnQHandler) Rcv

func (h EnQHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error

type Enque

type Enque struct {
	ID    ReqID  // The client-assigned request ID.
	Queue Queue  // The queue to enque the task.
	Body  []byte // The body of the task to be enqued.
}

Enque enqueus a task.

type Enqued

type Enqued struct {
	ID     ReqID  // The client-assigned request ID.
	Queue  Queue  // The queue that the task is enqueued in.
	TaskID TaskID // The assigned ID of the enqueued task.
}

Enqued represents a message emitted as a reply to a successful Enque request.

type Error

type Error struct {
	ID      ReqID  // The request ID.
	Message string // The error message.
}

Error is a message replied when there is an error handling a request.

func (*Error) Error

func (e *Error) Error() string

type Option

type Option args.V

Option represents TaskQ server options.

func Address

func Address(a string) Option

Address is an option representing a TaskQ address.

func MaxRate

func MaxRate(r uint64) Option

MaxRate is an option represeting the maximum rate of messages a client can send per second.

func ReplicationFactor

func ReplicationFactor(f int) Option

ReplicationFactor is an option represeting the replication factor of TaskQ.

type ProtoHandler

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

ProtoHandler is the detached handler that implements the TaskQ protocol.

func NewProtoHandler

func NewProtoHandler(addr string) (h *ProtoHandler, err error)

NewProtoHandler creates a detached handler for the TaskQ protocol listening on the given address. addr should be in the form of IP:PORT.

func (*ProtoHandler) Rcv

func (h *ProtoHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error

func (*ProtoHandler) Start

func (h *ProtoHandler) Start(ctx beehive.RcvContext)

func (*ProtoHandler) Stop

func (h *ProtoHandler) Stop(ctx beehive.RcvContext)

type Queue

type Queue string

Queue represents a named queue.

type ReqID

type ReqID uint64

ReqID contains a client assigned request ID.

func (ReqID) String

func (r ReqID) String() string

type Task

type Task struct {
	ID    TaskID `json:"id"`    // Task's globally unique ID assigned by TaskQ.
	Queue Queue  `json:"queue"` // Task's queue.
	Body  []byte `json:"body"`  // Task's client data.
}

Task represents a task in a queue.

type TaskID

type TaskID uint64

TaskID represents the ID of a task.

func (TaskID) String

func (id TaskID) String() string

type Timeout

type Timeout time.Time

Timeout represents a timeout message.

type TimeoutHandler

type TimeoutHandler struct {
	// ExpDur is the duration after which a dequed and unacknowledged task
	// is returned to the out-of-order dictionary.
	ExpDur time.Duration
}

TimeoutHandler handles Ack messages.

func (TimeoutHandler) Map

func (h TimeoutHandler) Map(msg beehive.Msg,
	ctx beehive.MapContext) beehive.MappedCells

func (TimeoutHandler) Rcv

func (h TimeoutHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error