Merge pull request #471 from ripienaar/470

(#470) replace satori/go.uuid with gofrs/uuid

broker/adapter/natsstream/output.go

@@ -14,9 +14,9 @@ import (
 	"github.com/choria-io/go-choria/broker/adapter/stats"
 	"github.com/choria-io/go-choria/choria"
 	"github.com/choria-io/go-choria/srvcache"
+	uuid "github.com/gofrs/uuid"
 	stan "github.com/nats-io/go-nats-streaming"
 	"github.com/prometheus/client_golang/prometheus"
-	uuid "github.com/satori/go.uuid"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -62,15 +62,20 @@ func newStream(name string, work chan adaptable, logger *log.Entry) ([]*stream,
 
 	clusterID := cfg.Option(prefix+"clusterid", "")
 	if clusterID == "" {
-		return nil, fmt.Errorf("No ClusterID configured, please set %s", prefix+"clusterid'")
+		return nil, fmt.Errorf("no ClusterID configured, please set %s", prefix+"clusterid'")
 	}
 
 	workers := []*stream{}
 
 	for i := 0; i < instances; i++ {
 		logger.Infof("Creating NATS Streaming Adapter %s NATS Streaming instance %d / %d publishing to %s on cluster %s", name, i, instances, topic, clusterID)
 
-		iname := fmt.Sprintf("%s_%d-%s", name, i, strings.Replace(uuid.NewV4().String(), "-", "", -1))
+		wid, err := uuid.NewV4()
+		if err != nil {
+			return nil, fmt.Errorf("could not start output worker %d: %s", i, err)
+		}
+
+		iname := fmt.Sprintf("%s_%d-%s", name, i, strings.Replace(wid.String(), "-", "", -1))
 
 		st := &stream{
 			clusterID: clusterID,

broker/federation/choria_nats_egest_test.go

@@ -19,7 +19,6 @@ var _ = Describe("Choria NATS Egest", func() {
 		connector *pooledWorker
 		manager   *stubConnectionManager
 		in        chainmessage
-		err       error
 		logtxt    *bufio.Writer
 		logbuf    *bytes.Buffer
 		logger    *log.Entry
@@ -31,7 +30,10 @@ var _ = Describe("Choria NATS Egest", func() {
 		ctx, cancel = context.WithCancel(context.Background())
 		logger, logtxt, logbuf = newDiscardLogger()
 
-		request, err = c.NewRequest(protocol.RequestV1, "test", "tester", "choria=tester", 60, c.NewRequestID(), "mcollective")
+		rid, err := c.NewRequestID()
+		Expect(err).ToNot(HaveOccurred())
+
+		request, err = c.NewRequest(protocol.RequestV1, "test", "tester", "choria=tester", 60, rid, "mcollective")
 		Expect(err).ToNot(HaveOccurred())
 		request.SetMessage(`{"hello":"world"}`)
 

broker/federation/choria_nats_ingest_test.go

@@ -20,7 +20,6 @@ var _ = Describe("Choria NATS Ingest", func() {
 		connector *pooledWorker
 		manager   *stubConnectionManager
 		in        *choria.ConnectorMessage
-		err       error
 		logtxt    *bufio.Writer
 		logbuf    *bytes.Buffer
 		logger    *log.Entry
@@ -33,7 +32,10 @@ var _ = Describe("Choria NATS Ingest", func() {
 		ctx, cancel = context.WithCancel(context.Background())
 		logger, logtxt, logbuf = newDiscardLogger()
 
-		request, err = c.NewRequest(protocol.RequestV1, "test", "tester", "choria=tester", 60, c.NewRequestID(), "mcollective")
+		rid, err := c.NewRequestID()
+		Expect(err).ToNot(HaveOccurred())
+
+		request, err = c.NewRequest(protocol.RequestV1, "test", "tester", "choria=tester", 60, rid, "mcollective")
 		Expect(err).ToNot(HaveOccurred())
 		request.SetMessage(`{"hello":"world"}`)
 

broker/federation/federation_test.go

@@ -141,8 +141,8 @@ func (s *stubConnection) Close() {
 	return
 }
 
-func (s *stubConnection) ReplyTarget(msg *choria.Message) string {
-	return ""
+func (s *stubConnection) ReplyTarget(msg *choria.Message) (string, error) {
+	return "stubreplytarget", nil
 }
 
 func (s *stubConnection) Nats() *nats.Conn {

broker/federation/reply_transformer_test.go

@@ -37,7 +37,10 @@ var _ = Describe("Reply Transformer", func() {
 		c, err = choria.New("testdata/federation.cfg")
 		Expect(err).ToNot(HaveOccurred())
 
-		request, err = c.NewRequest(protocol.RequestV1, "test", "tester", "choria=tester", 60, c.NewRequestID(), "mcollective")
+		rid, err := c.NewRequestID()
+		Expect(err).ToNot(HaveOccurred())
+
+		request, err = c.NewRequest(protocol.RequestV1, "test", "tester", "choria=tester", 60, rid, "mcollective")
 		Expect(err).ToNot(HaveOccurred())
 		request.SetMessage(`{"hello":"world"}`)
 

broker/federation/request_transformer_test.go

@@ -35,7 +35,10 @@ var _ = Describe("RequestTransformer", func() {
 		c, err = choria.New("testdata/federation.cfg")
 		Expect(err).ToNot(HaveOccurred())
 
-		request, err = c.NewRequest(protocol.RequestV1, "test", "tester", "choria=tester", 60, c.NewRequestID(), "mcollective")
+		rid, err := c.NewRequestID()
+		Expect(err).ToNot(HaveOccurred())
+
+		request, err = c.NewRequest(protocol.RequestV1, "test", "tester", "choria=tester", 60, rid, "mcollective")
 		Expect(err).ToNot(HaveOccurred())
 
 		request.SetMessage(`{"hello":"world"}`)

choria/connection.go

@@ -69,7 +69,7 @@ type InstanceConnector interface {
 type Connector interface {
 	InstanceConnector
 
-	ReplyTarget(msg *Message) string
+	ReplyTarget(msg *Message) (string, error)
 	ChanQueueSubscribe(name string, subject string, group string, capacity int) (chan *ConnectorMessage, error)
 	Connect(ctx context.Context) (err error)
 	Nats() *nats.Conn
@@ -486,8 +486,13 @@ func ReplyTarget(msg *Message, requestid string) string {
 	return fmt.Sprintf("%s.reply.%s.%s", msg.Collective(), msg.SenderID, requestid)
 }
 
-func (conn *Connection) ReplyTarget(msg *Message) string {
-	return ReplyTarget(msg, conn.choria.NewRequestID())
+func (conn *Connection) ReplyTarget(msg *Message) (string, error) {
+	id, err := conn.choria.NewRequestID()
+	if err != nil {
+		return "", err
+	}
+
+	return ReplyTarget(msg, id), nil
 }
 
 func (conn *Connection) federationTarget(federation string, side string) string {

choria/framework.go

@@ -471,7 +471,7 @@ func (fw *Framework) PuppetAIOCmd(command string, def string) string {
 }
 
 // NewRequestID Creates a new RequestID
-func (fw *Framework) NewRequestID() string {
+func (fw *Framework) NewRequestID() (string, error) {
 	return NewRequestID()
 }
 

choria/message.go

@@ -65,9 +65,14 @@ func NewMessageFromRequest(req protocol.Request, replyto string, choria *Framewo
 
 // NewMessage constructs a basic Message instance
 func NewMessage(payload string, agent string, collective string, msgType string, request *Message, choria *Framework) (msg *Message, err error) {
+	id, err := choria.NewRequestID()
+	if err != nil {
+		return
+	}
+
 	msg = &Message{
 		Payload:         payload,
-		RequestID:       choria.NewRequestID(),
+		RequestID:       id,
 		TTL:             choria.Config.TTL,
 		DiscoveredHosts: []string{},
 		SenderID:        choria.Config.Identity,

choria/message_test.go

@@ -129,7 +129,10 @@ var _ = Describe("Choria/Message", func() {
 		})
 
 		It("Should support reply", func() {
-			req, err := fw.NewRequest(protocol.RequestV1, "test_agent", "sender.example.net", "test=sender", 60, fw.NewRequestID(), "test_collective")
+			rid, err := fw.NewRequestID()
+			Expect(err).ToNot(HaveOccurred())
+
+			req, err := fw.NewRequest(protocol.RequestV1, "test_agent", "sender.example.net", "test=sender", 60, rid, "test_collective")
 			Expect(err).ToNot(HaveOccurred())
 			req.SetMessage("hello world")
 
@@ -194,7 +197,10 @@ var _ = Describe("Choria/Message", func() {
 		})
 
 		It("Should set up the transport", func() {
-			req, err := fw.NewRequest(protocol.RequestV1, "test_agent", "sender.example.net", "test=sender", 60, fw.NewRequestID(), "test_collective")
+			rid, err := fw.NewRequestID()
+			Expect(err).ToNot(HaveOccurred())
+
+			req, err := fw.NewRequest(protocol.RequestV1, "test_agent", "sender.example.net", "test=sender", 60, rid, "test_collective")
 			Expect(err).ToNot(HaveOccurred())
 			req.SetMessage("hello world")
 

choria/util.go

@@ -10,7 +10,7 @@ import (
 	"strings"
 
 	"github.com/choria-io/go-choria/puppet"
-	uuid "github.com/satori/go.uuid"
+	uuid "github.com/gofrs/uuid"
 )
 
 // UserConfig determines what is the active config file for a user
@@ -162,6 +162,11 @@ func MatchAnyRegex(str []byte, regex []string) bool {
 }
 
 // NewRequestID Creates a new RequestID
-func NewRequestID() string {
-	return strings.Replace(uuid.NewV4().String(), "-", "", -1)
+func NewRequestID() (string, error) {
+	id, err := uuid.NewV4()
+	if err != nil {
+		return "", err
+	}
+
+	return strings.Replace(id.String(), "-", "", -1), nil
 }

glide.yaml

@@ -11,8 +11,6 @@ import:
   version: ^1
 - package: github.com/nats-io/go-nats-streaming
   version: ^0.4.0
-- package: github.com/satori/go.uuid
-  version: ^1.2.0
 - package: github.com/tidwall/gjson
 - package: github.com/tidwall/match
 - package: github.com/xeipuuv/gojsonschema
@@ -58,3 +56,5 @@ import:
 - package: github.com/choria-io/go-mcoshim
 - package: github.com/choria-io/go-lifecycle
   version: ^0.2.0
+- package: github.com/gofrs/uuid
+  version: ^3.1.1

server/discovery/discovery_test.go

@@ -44,7 +44,10 @@ var _ = Describe("Server/Discovery", func() {
 
 	BeforeEach(func() {
 		mgr = New(fw, log)
-		req, err = fw.NewRequest(protocol.RequestV1, "test", "testid", "callerid", 60, fw.NewRequestID(), "mcollective")
+		rid, err := fw.NewRequestID()
+		Expect(err).ToNot(HaveOccurred())
+
+		req, err = fw.NewRequest(protocol.RequestV1, "test", "testid", "callerid", 60, rid, "mcollective")
 		Expect(err).ToNot(HaveOccurred())
 
 		filter = req.NewFilter()