iqueues

README

Introduction

Interface to interact between cluster components (including router).

Documentation

Index

• Constants
• Variables
• func TestImpl(actx context.Context, t *testing.T)
• type Event
• type HTTPMethod
• type NonPartyHandler
• type NonPartyHandlerChunked
• type PartitionHandler
• type PartitionHandlerFactory
• type PartitionedHandler
  • func (p *PartitionedHandler) Close(ctx context.Context)
  • func (p *PartitionedHandler) Handle(ctx context.Context, req *Request) *Response
• type QueueDef
• type Request
• type Response
  • func CreateResponse(code int, message string) *Response

Constants

const (
	HTTPMethodGET = HTTPMethod(iota)
	HTTPMethodPOST
	HTTPMethodPUT
	HTTPMethodPATCH
	HTTPMethodDELETE
)

HTTP methods

Variables

var DefaultTimeout = 10 * time.Second

DefaultTimeout to use in Invoke* functions

var ErrTimeoutExpired = errors.New("Timeout expired")

ErrTimeoutExpired s.e.

var HTTPMethodToName = map[HTTPMethod]string{HTTPMethodGET: "GET", HTTPMethodPOST: "POST", HTTPMethodPUT: "PUT", HTTPMethodPATCH: "PATCH", HTTPMethodDELETE: "DELETE"}

HTTPMethodToName s.e.

var Invoke func(ctx context.Context, request *Request, timeout time.Duration) (response *Response, err error)

Invoke sends a message to a given queue and returns a response

var InvokeFromHTTPRequest func(ctx context.Context, request *Request, resp http.ResponseWriter, timeout time.Duration)

InvokeFromHTTPRequest uses requests structure from https://github.com/untillpro/igoonce/blob/master/docs/requests.md

var NameToHTTPMethod = map[string]HTTPMethod{"GET": HTTPMethodGET, "POST": HTTPMethodPOST, "PUT": HTTPMethodPUT, "PATCH": HTTPMethodPATCH, "DELETE": HTTPMethodDELETE}

NameToHTTPMethod maps RFC 7231 names to HTTPMethod

var NonPartyHandlers map[string]NonPartyHandlerChunked

NonPartyHandlers should be registered here using godif.ProvideKeyValue() Key means QueueID to be handled

var NonPartyInvoke func(ctx context.Context, request *Request, inChunks <-chan []byte, timeout time.Duration) (
	response *Response, err error,
	outChunks <-chan []byte, outChunksError *error)

NonPartyInvoke sends a message to a NonParty Queue given queue and returns a response nil response means timeout expired *outChunksError valid after channel closing

var PartitionHandlerFactories map[string]PartitionHandlerFactory

PartitionHandlerFactories should be registered here using godif.ProvideKeyValue() Key: <QueueID>':'<Number of Partitions> Handlers will be automatically balanced between processes

var QueueDefs map[string]*QueueDef

QueueDefs keeps queue definitions

Functions

func TestImpl

func TestImpl(actx context.Context, t *testing.T)

TestImpl s.e. Storage must be empty before testing

Types

type Event

type Event struct {

	// Log EventBody should be written to using iloq.Put()
	LogID string

	EventBody []byte

	// Will be used by PartitionHandler.Deduplicate
	DeduplicationInfo map[string]interface{}
}

Event s.e.

type HTTPMethod

type HTTPMethod int

HTTPMethod s.e. see const.go/HTTPMethodGET...

type NonPartyHandler

type NonPartyHandler func(ctx context.Context, request *Request) *Response

NonPartyHandler s.e.

type NonPartyHandlerChunked

type NonPartyHandlerChunked func(ctx context.Context, request *Request, inChunks <-chan []byte) (response *Response, outChunks <-chan []byte, outChunksError *error)

NonPartyHandlerChunked s.e. *outChunksError valid after channel closing

var AirBoView NonPartyHandlerChunked = func(ctx context.Context, req *Request, inChunks <-chan []byte) (
	response *Response, outChunks <-chan []byte, outChunksError *error) {
	data, _ := json.Marshal(req)
	return &Response{
		Status:     http.StatusText(http.StatusOK),
		StatusCode: http.StatusOK,
		Data:       data,
	}, nil, nil
}

AirBoView use unusual http codes for testing

type PartitionHandler

type PartitionHandler interface {
	Handle(ctx context.Context, r *Request) *Response

	// Will be called when partition processing stops
	Close(ctx context.Context)
}

PartitionHandler s.e.

type PartitionHandlerFactory

type PartitionHandlerFactory func(ctx context.Context, queueID string, partitionNumber int) (handler PartitionHandler, err *Response)

PartitionHandlerFactory if err is not nil handler must be nil

var Factory PartitionHandlerFactory = func(ctx context.Context, queueID string,
	partitionNumber int) (handler PartitionHandler, err *Response) {
	return &PartitionedHandler{}, nil
}

Factory s.e.

type PartitionedHandler

type PartitionedHandler struct{}

PartitionedHandler s.e.

func (*PartitionedHandler) Close

func (p *PartitionedHandler) Close(ctx context.Context)

Close s.e.

func (*PartitionedHandler) Handle

func (p *PartitionedHandler) Handle(ctx context.Context, req *Request) *Response

Handle use unusual http codes for testing

type QueueDef

type QueueDef struct {
	ID               string
	PartitionDevisor int
	Handler          func(request *Request) *Response
}

QueueDef s.e.

type Request

type Request struct {
	Method HTTPMethod

	QueueID string

	// Always 0 for non-party queues
	WSID int64
	// Calculated from PartitionDividend, 0 for non-party queues
	PartitionNumber int

	Header map[string][]string

	// Part of URL which follows: queue alias in non-party queues, part dividend in partitioned queues
	Resource string

	// Part of URL which follows ? (URL.Query())
	Query map[string][]string

	// Content of http.Request JSON-parsed Body
	Body []byte

	// attachment-name => attachment-id
	// Must be non-null
	Attachments map[string]string
}

Request s.e.

type Response

type Response struct {
	ContenType string
	Status     string
	StatusCode int
	Data       []byte
}

Response s.e.

func CreateResponse

func CreateResponse(code int, message string) *Response

CreateResponse s.e.