events

Paddy 2015-07-15 Parent:987595765978 Child:df95f058a7b5

4:2d030022b4b5 Go to Latest

events/nsq.go

Create our Model helper. Create a Model interface that any type can implement by just providing getters that will be used to build Events produced around that type. Add a PublishModelEvent helper that takes a Publisher, a Model, and an action (as a string) and builds a consistent Event around the Model, then publishes it to the Publisher. This is a helpful wrapper in most of our systems, and will probably handle 99% of our NSQ publishes, because most of our infrastructure is going to be something like "a new profile was created" or "a new login was created" or "a profile was updated" and we just want to disseminate that information across all our services, to push based on it, or trigger actions. This abstracts it out nicely to a standard format and a nice, easy way to send those messages.

Download raw file
View source Diff to previous Annotate
History
1 package events
2
3 import (
4 "encoding/json"
5 "errors"
6 "fmt"
7
8 "github.com/bitly/go-nsq"
9 )
10
11 var (
12 ErrNilPublisher = errors.New("event publisher is nil")
13 )
14
15 type NSQPublisher nsq.Producer
16
17 func NewNSQPublisher(userAgent, address string) (*NSQPublisher, error) {
18 cfg := nsq.NewConfig()
19 cfg.UserAgent = fmt.Sprintf("%s go-nsq/%s", userAgent, nsq.VERSION)
20 err := cfg.Validate()
21 if err != nil {
22 return nil, err
23 }
24 producer, err := nsq.NewProducer(address, cfg)
25 if err != nil {
26 return nil, err
27 }
28 return (*NSQPublisher)(producer), nil
29 }
30
31 func (n *NSQPublisher) Publish(topic string, e Event) error {
32 if n == nil {
33 return ErrNilPublisher
34 }
35 data, err := json.Marshal(e)
36 if err != nil {
37 return err
38 }
39 return (*nsq.Producer)(n).Publish(topic, data)
40 }
41
42 type NSQSubscriber nsq.Consumer
43
44 func NewNSQSubscriber(lookupds []string, handler nsq.Handler, topic, channel, userAgent string) (*NSQSubscriber, error) {
45 cfg := nsq.NewConfig()
46 cfg.UserAgent = fmt.Sprintf("%s go-nsq/%s", userAgent, nsq.VERSION)
47 err := cfg.Validate()
48 if err != nil {
49 return nil, err
50 }
51 consumer, err := nsq.NewConsumer(topic, channel, cfg)
52 if err != nil {
53 return nil, err
54 }
55 consumer.AddHandler(handler)
56 err = consumer.ConnectToNSQLookupds(lookupds)
57 if err != nil {
58 return nil, err
59 }
60 return (*NSQSubscriber)(consumer), nil
61 }
62
63 func (sub *NSQSubscriber) Block() {
64 <-(*nsq.Consumer)(sub).StopChan
65 }