enqueuestomp

README

enqueuestomp

Use

package main

import (
    "log"

    "github.com/globocom/enqueuestomp"
    "github.com/go-stomp/stomp"
)

func main() {
    enqueueConfig := enqueuestomp.Config{
        Addr:  "localhost:61613",
    }
    enqueueConfig.SetOptions(
        stomp.ConnOpt.HeartBeat(0*time.Second, 0*time.Second),
    )

    enqueue, err := enqueuestomp.NewEnqueueStomp(
        enqueueConfig,
    )
    if err != nil {
        log.Fatalf("error %s", err)
    }

    sc := enqueuestomp.SendConfig{}
    sc.SetOptions(
        stomp.SendOpt.Header("persistent", "true"),
    )

    // If you want to add new fields in logs
    sc.AddLogField("testKey", "testValue")

    name := "queueName"
    body := []byte("queueBody")

    err := enqueue.SendQueue(name, body, sc)
    if err != nil {
        fmt.Printf("error %s", err)
    }
}

Enqueue config

type Config struct {
    // Default is tcp
    Network string

    // host:port address
    // Default is localhost:61613
    Addr string

    // https://pkg.go.dev/github.com/go-stomp/stomp
    Options []func(*stomp.Conn) error

    // The maxWorkers parameter specifies the maximum number of workers that can
    // execute tasks concurrently.  When there are no incoming tasks, workers are
    // gradually stopped until there are no remaining workers.
    // Default is runtime.NumCPU()
    MaxWorkers int

    // Default is 3, Max is 5
    RetriesConnect int

    // Used to determine how long a retry request should wait until attempted.
    // Default is ExponentialBackoff
    BackoffConnect BackoffStrategy

    // File path to write logging output to
    WriteOutputPath string

    // Logger that will be used
    // Default is nothing
    Logger Logger

    // create unique identifier
    // Default google/uuid
    IdentifierFunc func() string
}

Send config

type SendConfig struct {
    // The content type should be specified, according to the STOMP specification, but if contentType is an empty
    // string, the message will be delivered without a content-type header entry.
    // Default is text/plain.
    ContentType string

    // Any number of options can be specified in opts. See the examples for usage. Options include whether
    // to receive a RECEIPT, should the content-length be suppressed, and sending custom header entries.
    // https://pkg.go.dev/github.com/go-stomp/stomp/frame
    Options []func(*frame.Frame) error

    BeforeSend func(identifier string, destinationType string, destinationName string, body []byte, startTime time.Time)

    AfterSend func(identifier string, destinationType string, destinationName string, body []byte, startTime time.Time, err error)

    // the name of the CircuitBreaker.
    // Default is empty
    CircuitName string
}

CircuitBreaker config

type CircuitBreakerConfig struct {
    // how long to wait for command to complete, in milliseconds
    // Default is 10000
    Timeout int

    // how many commands of the same type can run at the same time
    // Default is 10000
    MaxConcurrentRequests int

    // the minimum number of requests needed before a circuit can be tripped due to health
    // Default is 100
    RequestVolumeThreshold int

    //  how long, in milliseconds, to wait after a circuit opens before testing for recovery
    // Default is 500
    SleepWindow int

    // causes circuits to open once the rolling measure of errors exceeds this percent of requests
    // Default is 5
    ErrorPercentThreshold int
}

Documentation

Full documentation for the package can be viewed online using the GoDoc site here: https://pkg.go.dev/github.com/globocom/enqueuestomp

Documentation

Index

• Constants
• Variables
• func ConstantBackOff(_ int) time.Duration
• func ExponentialBackoff(i int) time.Duration
• func LinearBackoff(i int) time.Duration
• type BackoffStrategy
• type CircuitBreakerConfig
• type Config
  • func (c *Config) AddOption(opt func(*stomp.Conn) error)
  • func (c *Config) SetOptions(opts ...func(*stomp.Conn) error)
• type EnqueueStomp
  • func NewEnqueueStomp(config Config) (EnqueueStomp, error)
• type EnqueueStompImpl
  • func (emq *EnqueueStompImpl) CheckQueue(queueName string) error
  • func (emq *EnqueueStompImpl) CheckTopic(topicName string) error
  • func (emq *EnqueueStompImpl) Config() Config
  • func (emq *EnqueueStompImpl) ConfigureCircuitBreaker(name string, cb CircuitBreakerConfig)
  • func (emq *EnqueueStompImpl) Disconnect() error
  • func (emq *EnqueueStompImpl) QueueSize() int
  • func (emq *EnqueueStompImpl) SendQueue(queueName string, body []byte, sc SendConfig) error
  • func (emq *EnqueueStompImpl) SendTopic(topicName string, body []byte, sc SendConfig) error
• type LogField
• type LogFieldImpl
• type Logger
• type NoopLogger
  • func (l NoopLogger) Debugf(template string, args ...interface{})
  • func (l NoopLogger) Errorf(template string, args ...interface{})
• type SendConfig
  • func (sc *SendConfig) AddLogField(key, value string)
  • func (sc *SendConfig) AddOption(opt func(*frame.Frame) error)
  • func (sc *SendConfig) SetOptions(opts ...func(*frame.Frame) error)

Constants

const (
	DefaultRetriesConnect    = 3
	DefaultMaxRetriesConnect = 5
)

const (
	DestinationTypeQueue = "queue"
	DestinationTypeTopic = "topic"
)

const DefaultInitialBackOff = 100 * time.Millisecond

Variables

var (
	ErrEmptyBody      = errors.New("empty body")
	ErrEmptyQueueName = errors.New("empty queue name")
	ErrEmptyTopicName = errors.New("empty topic name")

	DefaultExpiresCheck = 1 * time.Minute
	DefaultBodyCheck    = []byte("PING")
)

Functions

func ConstantBackOff

func ConstantBackOff(_ int) time.Duration

ConstantBackOff always returns DefaultInitialBackOff.

func ExponentialBackoff

func ExponentialBackoff(i int) time.Duration

ExponentialBackoff returns ever increasing backoffs by a power of 2.

func LinearBackoff

func LinearBackoff(i int) time.Duration

LinearBackoff returns increasing durations.

Types

type BackoffStrategy

type BackoffStrategy func(retry int) time.Duration

BackoffStrategy is used to determine how long a retry request should wait until attempted.

type CircuitBreakerConfig

type CircuitBreakerConfig struct {
	// how long to wait for command to complete, in milliseconds
	// Default is 10000
	Timeout int

	// how many commands of the same type can run at the same time
	// Default is 10000
	MaxConcurrentRequests int

	// the minimum number of requests needed before a circuit can be tripped due to health
	// Default is 100
	RequestVolumeThreshold int

	//  how long, in milliseconds, to wait after a circuit opens before testing for recovery
	// Default is 500
	SleepWindow int

	// causes circuits to open once the rolling measure of errors exceeds this percent of requests
	// Default is 5
	ErrorPercentThreshold int
}

type Config

type Config struct {
	// Default is tcp
	Network string

	// host:port address
	// Default is localhost:61613
	Addr string

	// https://pkg.go.dev/github.com/go-stomp/stomp
	Options []func(*stomp.Conn) error

	// The maxWorkers parameter specifies the maximum number of workers that can
	// execute tasks concurrently.  When there are no incoming tasks, workers are
	// gradually stopped until there are no remaining workers.
	// Default is runtime.NumCPU()
	MaxWorkers int

	// Default is 3, Max is 5
	RetriesConnect int

	// Used to determine how long a retry request should wait until attempted.
	// Default is ExponentialBackoff
	BackoffConnect BackoffStrategy

	// File path to write logging output to
	WriteOutputPath string

	// Logger that will be used
	// Default is nothing
	Logger Logger

	// create unique identifier
	// Default google/uuid
	IdentifierFunc func() string
}

func (*Config) AddOption

func (c *Config) AddOption(opt func(*stomp.Conn) error)

func (*Config) SetOptions

func (c *Config) SetOptions(opts ...func(*stomp.Conn) error)

type EnqueueStomp

type EnqueueStomp interface {
	SendQueue(queueName string, body []byte, sc SendConfig) error
	SendTopic(topicName string, body []byte, sc SendConfig) error
	QueueSize() int
	Config() Config
	CheckQueue(queueName string) error
	CheckTopic(topicName string) error
	Disconnect() error
	ConfigureCircuitBreaker(name string, cb CircuitBreakerConfig)
}

func NewEnqueueStomp

func NewEnqueueStomp(config Config) (EnqueueStomp, error)

type EnqueueStompImpl

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

func (*EnqueueStompImpl) CheckQueue

func (emq *EnqueueStompImpl) CheckQueue(queueName string) error

func (*EnqueueStompImpl) CheckTopic

func (emq *EnqueueStompImpl) CheckTopic(topicName string) error

func (*EnqueueStompImpl) Config

func (emq *EnqueueStompImpl) Config() Config

func (*EnqueueStompImpl) ConfigureCircuitBreaker

func (emq *EnqueueStompImpl) ConfigureCircuitBreaker(name string, cb CircuitBreakerConfig)

func (*EnqueueStompImpl) Disconnect

func (emq *EnqueueStompImpl) Disconnect() error

func (*EnqueueStompImpl) QueueSize

func (emq *EnqueueStompImpl) QueueSize() int

func (*EnqueueStompImpl) SendQueue

func (emq *EnqueueStompImpl) SendQueue(queueName string, body []byte, sc SendConfig) error

SendQueue The body array contains the message body, and its content should be consistent with the specified content type.

func (*EnqueueStompImpl) SendTopic

func (emq *EnqueueStompImpl) SendTopic(topicName string, body []byte, sc SendConfig) error

SendTopic The body array contains the message body, and its content should be consistent with the specified content type.

type LogField

type LogField interface {
	// contains filtered or unexported methods
}

type LogFieldImpl

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

type Logger

type Logger interface {
	Debugf(template string, args ...interface{})
	Errorf(template string, args ...interface{})
}

type NoopLogger

type NoopLogger struct{}

NoopLogger does not log anything.

func (NoopLogger) Debugf

func (l NoopLogger) Debugf(template string, args ...interface{})

Debugf does nothing.

func (NoopLogger) Errorf

func (l NoopLogger) Errorf(template string, args ...interface{})

Errorf does nothing.

type SendConfig

type SendConfig struct {
	// The content type should be specified, according to the STOMP specification, but if contentType is an empty
	// string, the message will be delivered without a content-type header entry.
	// Default is text/plain.
	ContentType string

	// Any number of options can be specified in opts. See the examples for usage. Options include whether
	// to receive a RECEIPT, should the content-length be suppressed, and sending custom header entries.
	// https://pkg.go.dev/github.com/go-stomp/stomp/frame
	Options []func(*frame.Frame) error

	BeforeSend func(identifier string, destinationType string, destinationName string, body []byte, startTime time.Time)

	AfterSend func(identifier string, destinationType string, destinationName string, body []byte, startTime time.Time, err error)

	// the name of the CircuitBreaker.
	// Default is empty
	CircuitName string
	// contains filtered or unexported fields
}

func (*SendConfig) AddLogField

func (sc *SendConfig) AddLogField(key, value string)

func (*SendConfig) AddOption

func (sc *SendConfig) AddOption(opt func(*frame.Frame) error)

func (*SendConfig) SetOptions

func (sc *SendConfig) SetOptions(opts ...func(*frame.Frame) error)