Fix some data races

Makefile

@@ -31,7 +31,7 @@ dev : build
 	@./msgbusd -v
 
 cli:
-	@$(GOCMD) build -tags "netgo static_build" -installsuffix netgo \
+	@$(GOCMD) build $(FLAGS) -tags "netgo static_build" -installsuffix netgo \
 		-ldflags "-w \
 		-X $(shell go list).Version=$(VERSION) \
 		-X $(shell go list).Commit=$(COMMIT)" \

client/client.go

@@ -38,6 +38,8 @@ const (
 
 // Client ...
 type Client struct {
+	sync.RWMutex
+
 	url string
 
 	reconnectInterval    time.Duration
@@ -92,6 +94,9 @@ func (c *Client) Handle(msg *msgbus.Message) error {
 
 // Pull ...
 func (c *Client) Pull(topic string) (msg *msgbus.Message, err error) {
+	c.RLock()
+	defer c.RUnlock()
+
 	url := fmt.Sprintf("%s/%s", c.url, topic)
 	client := &http.Client{}
 
@@ -135,6 +140,9 @@ func (c *Client) Pull(topic string) (msg *msgbus.Message, err error) {
 
 // Publish ...
 func (c *Client) Publish(topic, message string) error {
+	c.RLock()
+	defer c.RUnlock()
+
 	var payload bytes.Buffer
 
 	payload.Write([]byte(message))
@@ -219,17 +227,21 @@ func NewSubscriber(client *Client, topic string, handler msgbus.HandlerFunc) *Su
 
 func (s *Subscriber) closeAndReconnect() {
 	s.closeWriteChan <- true
+	s.RLock()
 	s.conn.Close()
+	s.RUnlock()
 	go s.connect()
 }
 
 func (s *Subscriber) connect() {
+	s.RLock()
 	b := &backoff.Backoff{
 		Min:    s.reconnectInterval,
 		Max:    s.maxReconnectInterval,
 		Factor: 2,
 		Jitter: false,
 	}
+	s.RUnlock()
 
 	for {
 		d := b.Duration()
@@ -331,5 +343,7 @@ func (s *Subscriber) Stop() {
 		log.Warnf("error closing connection: %s", err)
 	}
 
+	s.Lock()
 	s.conn = nil
+	s.Unlock()
 }

examples/hello.sh

@@ -0,0 +1,7 @@
+#!/bin/sh
+
+payload="$(cat - | jq -r '.payload' | base64 -d)"
+
+echo "$payload"
+
+sleep 3