bus

package module
Version: v0.0.0-...-9523542 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Feb 15, 2018 License: MIT Imports: 14 Imported by: 7

README

Event Bus NSQ

  • A tiny wrapper around go-nsq topic and channel.
  • Protect nsq calls with gobreaker.

Installation

go get -u github.com/rafaeljesus/nsq-event-bus

Usage

The nsq-event-bus package exposes a interface for emitting and listening events.

Emitter

import "github.com/rafaeljesus/nsq-event-bus"

topic := "events"
emitter, err := bus.NewEmitter(bus.EmitterConfig{
  Address: "localhost:4150",
  MaxInFlight: 25,
})

e := event{}
if err = emitter.Emit(topic, &e); err != nil {
  // handle failure to emit message
}

// emitting messages on a async fashion
if err = emitter.EmitAsync(topic, &e); err != nil {
  // handle failure to emit message
}

Listener

import "github.com/rafaeljesus/nsq-event-bus"

if err = bus.On(bus.ListenerConfig{
  Topic:              "topic",
  Channel:            "test_on",
  HandlerFunc:        handler,
  HandlerConcurrency: 4,
}); err != nil {
  // handle failure to listen a message
}

func handler(message *Message) (reply interface{}, err error) {
  e := event{}
  if err = message.DecodePayload(&e); err != nil {
    message.Finish()
    return
  }

  if message.Attempts > MAX_DELIVERY_ATTEMPTS {
    message.Finish()
    return
  }

  err, _ = doWork(&e)
  if err != nil {
    message.Requeue(BACKOFF_TIME)
    return
  }

  message.Finish()
  return
}

Request (Request/Reply)

import "github.com/rafaeljesus/nsq-event-bus"

topic := "user_signup"
emitter, err = bus.NewEmitter(bus.EmitterConfig{})

e := event{Login: "rafa", Password: "ilhabela_is_the_place"}
if err = bus.Request(topic, &e, handler); err != nil {
  // handle failure to listen a message
}

func handler(message *Message) (reply interface{}, err error) {
  e := event{}
  if err = message.DecodePayload(&e); err != nil {
    message.Finish()
    return
  }

  reply = &Reply{}
  message.Finish()
  return
}

Contributing

  • Fork it
  • Create your feature branch (git checkout -b my-new-feature)
  • Commit your changes (git commit -am 'Add some feature')
  • Push to the branch (git push origin my-new-feature)
  • Create new Pull Request

Badges

Build Status Go Report Card Go Doc


GitHub @rafaeljesus  ·  Medium @_jesus_rafael  ·  Twitter @_jesus_rafael

Documentation

Index

Constants

This section is empty.

Variables

View Source
var (
	// ErrTopicRequired is returned when topic is not passed as parameter.
	ErrTopicRequired = errors.New("topic is mandatory")
	// ErrHandlerRequired is returned when handler is not passed as parameter.
	ErrHandlerRequired = errors.New("handler is mandatory")
	// ErrChannelRequired is returned when channel is not passed as parameter in bus.ListenerConfig.
	ErrChannelRequired = errors.New("channel is mandatory")
)

Functions

func On

func On(lc ListenerConfig) error

On listen to a message from a specific topic using nsq consumer, returns an error if topic and channel not passed or if an error occurred while creating nsq consumer.

Types

type Breaker

type Breaker struct {
	// Interval is the cyclic period of the closed state for CircuitBreaker to clear the internal counts,
	// If Interval is 0, CircuitBreaker doesn't clear the internal counts during the closed state.
	Interval time.Duration
	// Timeout is the period of the open state, after which the state of CircuitBreaker becomes half-open.
	// If Timeout is 0, the timeout value of CircuitBreaker is set to 60 seconds.
	Timeout time.Duration
	// Threshold when a threshold of failures has been reached, future calls to the broker will not run.
	// During this state, the circuit breaker will periodically allow the calls to run and, if it is successful,
	// will start running the function again. Default value is 5.
	Threshold uint32
	// OnStateChange is called whenever the state of CircuitBreaker changes.
	OnStateChange func(name, from, to string)
}

Breaker carries the configuration for circuit breaker

type Emitter

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

Emitter is the emitter wrapper over nsq.

func NewEmitter

func NewEmitter(ec EmitterConfig) (*Emitter, error)

NewEmitter returns a new Emitter configured with the variables from the config parameter, or returning an non-nil err if an error occurred while creating nsq producer.

func (*Emitter) Emit

func (e *Emitter) Emit(topic string, payload interface{}) error

Emit emits a message to a specific topic using nsq producer, returning an error if encoding payload fails or if an error occurred while publishing the message.

func (*Emitter) EmitAsync

func (e *Emitter) EmitAsync(topic string, payload interface{}) error

Emit emits a message to a specific topic using nsq producer, but does not wait for the response from `nsqd`. Returns an error if encoding payload fails and logs to console if an error occurred while publishing the message.

func (*Emitter) Request

func (e *Emitter) Request(topic string, payload interface{}, handler HandlerFunc) error

Request a RPC like method which implements request/reply pattern using nsq producer and consumer. Returns an non-nil err if an error occurred while creating or listening to the internal reply topic or encoding the message payload fails or while publishing the message.

type EmitterConfig

type EmitterConfig struct {
	Address                 string
	DialTimeout             time.Duration
	ReadTimeout             time.Duration
	WriteTimeout            time.Duration
	LocalAddr               net.Addr
	LookupdPollInterval     time.Duration
	LookupdPollJitter       float64
	MaxRequeueDelay         time.Duration
	DefaultRequeueDelay     time.Duration
	BackoffStrategy         nsq.BackoffStrategy
	MaxBackoffDuration      time.Duration
	BackoffMultiplier       time.Duration
	MaxAttempts             uint16
	LowRdyIdleTimeout       time.Duration
	RDYRedistributeInterval time.Duration
	ClientID                string
	Hostname                string
	UserAgent               string
	HeartbeatInterval       time.Duration
	SampleRate              int32
	TLSV1                   bool
	TLSConfig               *tls.Config
	Deflate                 bool
	DeflateLevel            int
	Snappy                  bool
	OutputBufferSize        int64
	OutputBufferTimeout     time.Duration
	MaxInFlight             int
	MsgTimeout              time.Duration
	AuthSecret              string
	// Breaker circuit breaker configuration
	Breaker
}

EmitterConfig carries the different variables to tune a newly started emitter, it exposes the same configuration available from official nsq go client.

type HandlerFunc

type HandlerFunc func(m *Message) (interface{}, error)

HandlerFunc is the handler function to handle the massage.

type ListenerConfig

type ListenerConfig struct {
	Topic                   string
	Channel                 string
	Lookup                  []string
	HandlerFunc             HandlerFunc
	HandlerConcurrency      int
	DialTimeout             time.Duration
	ReadTimeout             time.Duration
	WriteTimeout            time.Duration
	LocalAddr               net.Addr
	LookupdPollInterval     time.Duration
	LookupdPollJitter       float64
	MaxRequeueDelay         time.Duration
	DefaultRequeueDelay     time.Duration
	BackoffStrategy         nsq.BackoffStrategy
	MaxBackoffDuration      time.Duration
	BackoffMultiplier       time.Duration
	MaxAttempts             uint16
	LowRdyIdleTimeout       time.Duration
	RDYRedistributeInterval time.Duration
	ClientID                string
	Hostname                string
	UserAgent               string
	HeartbeatInterval       time.Duration
	SampleRate              int32
	TLSV1                   bool
	TLSConfig               *tls.Config
	Deflate                 bool
	DeflateLevel            int
	Snappy                  bool
	OutputBufferSize        int64
	OutputBufferTimeout     time.Duration
	MaxInFlight             int
	MsgTimeout              time.Duration
	AuthSecret              string
}

ListenerConfig carries the different variables to tune a newly started consumer, it exposes the same configuration available from official nsq go client.

type Message

type Message struct {
	*nsq.Message
	ReplyTo string
	Payload []byte
}

Message carries nsq.Message fields and methods and adds extra fields for handling messages internally.

func NewMessage

func NewMessage(p []byte, r string) *Message

NewMessage returns a new bus.Message.

func (*Message) DecodePayload

func (m *Message) DecodePayload(v interface{}) (err error)

DecodePayload deserializes data (as []byte) and creates a new struct passed by parameter.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL