Skip to content
A tiny wrapper around NSQ topic and channel 🚀
Branch: master
Clone or download
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
.gitignore Release Jan 15, 2017
Gopkg.lock merging Feb 14, 2018
Gopkg.toml merging Feb 14, 2018
LICENSE Add License Mar 19, 2017
Makefile Add makefile Dec 3, 2017
README.md Table tests Dec 3, 2017
circle.yml Remove glide form ci Dec 3, 2017
config.go update ci Feb 14, 2018
config_test.go Fix config test Mar 23, 2017
docker-compose.yml Fix bugs and refactoring docker compose Feb 22, 2017
emitter.go fix test Feb 14, 2018
emitter_test.go
listener.go update ci Feb 14, 2018
listener_test.go tiny refactoring test style Feb 15, 2018
message.go
message_test.go
test.sh

README.md

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

You can’t perform that action at this time.