Refactor guid, message, queue

nsqd/backend_queue.go

@@ -1,5 +1,7 @@
 package nsqd
 
+import "bytes"
+
 // BackendQueue represents the behavior for the secondary message
 // storage system
 type BackendQueue interface {
@@ -9,4 +11,5 @@ type BackendQueue interface {
 	Delete() error
 	Depth() int64
 	Empty() error
+	WriteMsg(buf *bytes.Buffer, msg *Message) error
 }

nsqd/channel.go

@@ -224,7 +224,7 @@ func (c *Channel) flush() error {
 	// this will read until it's closed (exited)
 	for msg := range c.clientMsgChan {
 		c.ctx.nsqd.logf("CHANNEL(%s): recovered buffered message from clientMsgChan", c.name)
-		writeMessageToBackend(&msgBuf, msg, c.backend)
+		c.backend.WriteMsg(&msgBuf, msg)
 	}
 
 	if len(c.memoryMsgChan) > 0 || len(c.inFlightMessages) > 0 || len(c.deferredMessages) > 0 {
@@ -235,7 +235,7 @@ func (c *Channel) flush() error {
 	for {
 		select {
 		case msg := <-c.memoryMsgChan:
-			err := writeMessageToBackend(&msgBuf, msg, c.backend)
+			err := c.backend.WriteMsg(&msgBuf, msg)
 			if err != nil {
 				c.ctx.nsqd.logf("ERROR: failed to write message to backend - %s", err)
 			}
@@ -246,15 +246,15 @@ func (c *Channel) flush() error {
 
 finish:
 	for _, msg := range c.inFlightMessages {
-		err := writeMessageToBackend(&msgBuf, msg, c.backend)
+		err := c.backend.WriteMsg(&msgBuf, msg)
 		if err != nil {
 			c.ctx.nsqd.logf("ERROR: failed to write message to backend - %s", err)
 		}
 	}
 
 	for _, item := range c.deferredMessages {
 		msg := item.Value.(*Message)
-		err := writeMessageToBackend(&msgBuf, msg, c.backend)
+		err := c.backend.WriteMsg(&msgBuf, msg)
 		if err != nil {
 			c.ctx.nsqd.logf("ERROR: failed to write message to backend - %s", err)
 		}
@@ -318,7 +318,7 @@ func (c *Channel) put(m *Message) error {
 	case c.memoryMsgChan <- m:
 	default:
 		b := bufferPoolGet()
-		err := writeMessageToBackend(b, m, c.backend)
+		err := c.backend.WriteMsg(b, m)
 		bufferPoolPut(b)
 		if err != nil {
 			c.ctx.nsqd.logf("CHANNEL(%s) ERROR: failed to write message to backend - %s",

nsqd/diskqueue.go

@@ -630,3 +630,12 @@ exit:
 	syncTicker.Stop()
 	d.exitSyncChan <- 1
 }
+
+func (d *diskQueue) WriteMsg(buf *bytes.Buffer, msg *Message) error {
+	buf.Reset()
+	_, err := msg.WriteTo(buf)
+	if err != nil {
+		return err
+	}
+	return d.Put(buf.Bytes())
+}

nsqd/dummy_backend_queue.go

@@ -1,5 +1,9 @@
 package nsqd
 
+import (
+	"bytes"
+)
+
 type dummyBackendQueue struct {
 	readChan chan []byte
 }
@@ -31,3 +35,7 @@ func (d *dummyBackendQueue) Depth() int64 {
 func (d *dummyBackendQueue) Empty() error {
 	return nil
 }
+
+func (d *dummyBackendQueue) WriteMsg(buf *bytes.Buffer, msg *Message) error {
+	return nil
+}

nsqd/guid.go → nsqd/guid/guid.go

@@ -1,4 +1,4 @@
-package nsqd
+package guid
 
 // the core algorithm here was borrowed from:
 // Blake Mizerany's `noeqd` https://github.com/bmizerany/noeqd
@@ -10,7 +10,6 @@ package nsqd
 // behavior when sequences rollover for our specific implementation needs
 
 import (
-	"encoding/hex"
 	"errors"
 	"time"
 )
@@ -30,15 +29,15 @@ var ErrTimeBackwards = errors.New("time has gone backwards")
 var ErrSequenceExpired = errors.New("sequence expired")
 var ErrIDBackwards = errors.New("ID went backward")
 
-type guid int64
+type Guid int64
 
-type guidFactory struct {
+type GuidFactory struct {
 	sequence      int64
 	lastTimestamp int64
-	lastID        guid
+	lastID        Guid
 }
 
-func (f *guidFactory) NewGUID(workerID int64) (guid, error) {
+func (f *GuidFactory) NewGUID(workerID int64) (Guid, error) {
 	ts := time.Now().UnixNano() / 1e6
 
 	if ts < f.lastTimestamp {
@@ -56,7 +55,7 @@ func (f *guidFactory) NewGUID(workerID int64) (guid, error) {
 
 	f.lastTimestamp = ts
 
-	id := guid(((ts - twepoch) << timestampShift) |
+	id := Guid(((ts - twepoch) << timestampShift) |
 		(workerID << workerIDShift) |
 		f.sequence)
 
@@ -68,20 +67,3 @@ func (f *guidFactory) NewGUID(workerID int64) (guid, error) {
 
 	return id, nil
 }
-
-func (g guid) Hex() MessageID {
-	var h MessageID
-	var b [8]byte
-
-	b[0] = byte(g >> 56)
-	b[1] = byte(g >> 48)
-	b[2] = byte(g >> 40)
-	b[3] = byte(g >> 32)
-	b[4] = byte(g >> 24)
-	b[5] = byte(g >> 16)
-	b[6] = byte(g >> 8)
-	b[7] = byte(g)
-
-	hex.Encode(h[:], b[:])
-	return h
-}

nsqd/message.go

@@ -3,10 +3,13 @@ package nsqd
 import (
 	"bytes"
 	"encoding/binary"
+	"encoding/hex"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"time"
+
+	"github.com/bitly/nsq/nsqd/guid"
 )
 
 const (
@@ -16,6 +19,23 @@ const (
 
 type MessageID [MsgIDLength]byte
 
+func NewMessageID(g guid.Guid) MessageID {
+	var h MessageID
+	var b [8]byte
+
+	b[0] = byte(g >> 56)
+	b[1] = byte(g >> 48)
+	b[2] = byte(g >> 40)
+	b[3] = byte(g >> 32)
+	b[4] = byte(g >> 24)
+	b[5] = byte(g >> 16)
+	b[6] = byte(g >> 8)
+	b[7] = byte(g)
+
+	hex.Encode(h[:], b[:])
+	return h
+}
+
 type Message struct {
 	ID        MessageID
 	Body      []byte
@@ -90,16 +110,3 @@ func decodeMessage(b []byte) (*Message, error) {
 
 	return &msg, nil
 }
-
-func writeMessageToBackend(buf *bytes.Buffer, msg *Message, bq BackendQueue) error {
-	buf.Reset()
-	_, err := msg.WriteTo(buf)
-	if err != nil {
-		return err
-	}
-	err = bq.Put(buf.Bytes())
-	if err != nil {
-		return err
-	}
-	return nil
-}

nsqd/guid_test.go → nsqd/message_guid_test.go

@@ -1,6 +1,7 @@
 package nsqd
 
 import (
+	"github.com/bitly/nsq/nsqd/guid"
 	"testing"
 	"unsafe"
 )
@@ -23,12 +24,12 @@ func BenchmarkGUIDUnsafe(b *testing.B) {
 }
 
 func BenchmarkGUID(b *testing.B) {
-	factory := &guidFactory{}
+	factory := &guid.GuidFactory{}
 	for i := 0; i < b.N; i++ {
 		guid, err := factory.NewGUID(0)
 		if err != nil {
 			continue
 		}
-		guid.Hex()
+		NewMessageID(guid)
 	}
 }

nsqd/nsqd.go

@@ -24,6 +24,7 @@ import (
 	"github.com/bitly/nsq/internal/statsd"
 	"github.com/bitly/nsq/internal/util"
 	"github.com/bitly/nsq/internal/version"
+	"github.com/bitly/nsq/nsqd/guid"
 )
 
 const (
@@ -519,7 +520,7 @@ func (n *NSQD) DeleteExistingTopic(topicName string) error {
 }
 
 func (n *NSQD) idPump() {
-	factory := &guidFactory{}
+	factory := &guid.GuidFactory{}
 	lastError := time.Unix(0, 0)
 	workerID := n.getOpts().ID
 	for {
@@ -535,7 +536,7 @@ func (n *NSQD) idPump() {
 			continue
 		}
 		select {
-		case n.idChan <- id.Hex():
+		case n.idChan <- NewMessageID(id):
 		case <-n.exitChan:
 			goto exit
 		}

nsqd/topic.go

@@ -188,7 +188,7 @@ func (t *Topic) put(m *Message) error {
 	case t.memoryMsgChan <- m:
 	default:
 		b := bufferPoolGet()
-		err := writeMessageToBackend(b, m, t.backend)
+		err := t.backend.WriteMsg(b, m)
 		bufferPoolPut(b)
 		if err != nil {
 			t.ctx.nsqd.logf(
@@ -373,7 +373,7 @@ func (t *Topic) flush() error {
 	for {
 		select {
 		case msg := <-t.memoryMsgChan:
-			err := writeMessageToBackend(&msgBuf, msg, t.backend)
+			err := t.backend.WriteMsg(&msgBuf, msg)
 			if err != nil {
 				t.ctx.nsqd.logf(
 					"ERROR: failed to write message to backend - %s", err)

nsqd/topic_test.go

@@ -1,6 +1,7 @@
 package nsqd
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"io/ioutil"
@@ -52,12 +53,13 @@ func TestGetChannel(t *testing.T) {
 
 type errorBackendQueue struct{}
 
-func (d *errorBackendQueue) Put([]byte) error      { return errors.New("never gonna happen") }
-func (d *errorBackendQueue) ReadChan() chan []byte { return nil }
-func (d *errorBackendQueue) Close() error          { return nil }
-func (d *errorBackendQueue) Delete() error         { return nil }
-func (d *errorBackendQueue) Depth() int64          { return 0 }
-func (d *errorBackendQueue) Empty() error          { return nil }
+func (d *errorBackendQueue) Put([]byte) error                               { return errors.New("never gonna happen") }
+func (d *errorBackendQueue) ReadChan() chan []byte                          { return nil }
+func (d *errorBackendQueue) Close() error                                   { return nil }
+func (d *errorBackendQueue) Delete() error                                  { return nil }
+func (d *errorBackendQueue) Depth() int64                                   { return 0 }
+func (d *errorBackendQueue) Empty() error                                   { return nil }
+func (d *errorBackendQueue) WriteMsg(buf *bytes.Buffer, msg *Message) error { return nil }
 
 func TestHealth(t *testing.T) {
 	opts := NewOptions()