Skip to content

Commit 339262d

Browse files
mreifersonmreiferson
committed
nsqd: dont depend on go-nsq for non-client functions
1 parent 9803e56 commit 339262d

17 files changed

+352
-274
lines changed

‎nsqd/channel.go

+28-29
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@ import (
1111
"sync/atomic"
1212
"time"
1313

14-
"github.com/bitly/go-nsq"
1514
"github.com/bitly/nsq/util"
1615
"github.com/bitly/nsq/util/pqueue"
1716
)
@@ -50,9 +49,9 @@ type Channel struct {
5049

5150
backend BackendQueue
5251

53-
incomingMsgChan chan *nsq.Message
54-
memoryMsgChan chan *nsq.Message
55-
clientMsgChan chan *nsq.Message
52+
incomingMsgChan chan *Message
53+
memoryMsgChan chan *Message
54+
clientMsgChan chan *Message
5655
exitChan chan int
5756
waitGroup util.WaitGroupWrapper
5857
exitFlag int32
@@ -68,10 +67,10 @@ type Channel struct {
6867
e2eProcessingLatencyStream *util.Quantile
6968

7069
// TODO: these can be DRYd up
71-
deferredMessages map[nsq.MessageID]*pqueue.Item
70+
deferredMessages map[MessageID]*pqueue.Item
7271
deferredPQ pqueue.PriorityQueue
7372
deferredMutex sync.Mutex
74-
inFlightMessages map[nsq.MessageID]*pqueue.Item
73+
inFlightMessages map[MessageID]*pqueue.Item
7574
inFlightPQ pqueue.PriorityQueue
7675
inFlightMutex sync.Mutex
7776

@@ -80,7 +79,7 @@ type Channel struct {
8079
}
8180

8281
type inFlightMessage struct {
83-
msg *nsq.Message
82+
msg *Message
8483
clientID int64
8584
ts time.Time
8685
}
@@ -92,9 +91,9 @@ func NewChannel(topicName string, channelName string, context *context,
9291
c := &Channel{
9392
topicName: topicName,
9493
name: channelName,
95-
incomingMsgChan: make(chan *nsq.Message, 1),
96-
memoryMsgChan: make(chan *nsq.Message, context.nsqd.options.MemQueueSize),
97-
clientMsgChan: make(chan *nsq.Message),
94+
incomingMsgChan: make(chan *Message, 1),
95+
memoryMsgChan: make(chan *Message, context.nsqd.options.MemQueueSize),
96+
clientMsgChan: make(chan *Message),
9897
exitChan: make(chan int),
9998
clients: make(map[int64]Consumer),
10099
deleteCallback: deleteCallback,
@@ -136,8 +135,8 @@ func NewChannel(topicName string, channelName string, context *context,
136135
func (c *Channel) initPQ() {
137136
pqSize := int(math.Max(1, float64(c.context.nsqd.options.MemQueueSize)/10))
138137

139-
c.inFlightMessages = make(map[nsq.MessageID]*pqueue.Item)
140-
c.deferredMessages = make(map[nsq.MessageID]*pqueue.Item)
138+
c.inFlightMessages = make(map[MessageID]*pqueue.Item)
139+
c.deferredMessages = make(map[MessageID]*pqueue.Item)
141140

142141
c.inFlightMutex.Lock()
143142
c.inFlightPQ = pqueue.New(pqSize)
@@ -273,7 +272,7 @@ finish:
273272
}
274273

275274
for _, item := range c.deferredMessages {
276-
msg := item.Value.(*nsq.Message)
275+
msg := item.Value.(*Message)
277276
err := writeMessageToBackend(&msgBuf, msg, c.backend)
278277
if err != nil {
279278
log.Printf("ERROR: failed to write message to backend - %s", err.Error())
@@ -325,7 +324,7 @@ func (c *Channel) IsPaused() bool {
325324

326325
// PutMessage writes to the appropriate incoming message channel
327326
// (which will be routed asynchronously)
328-
func (c *Channel) PutMessage(msg *nsq.Message) error {
327+
func (c *Channel) PutMessage(msg *Message) error {
329328
c.RLock()
330329
defer c.RUnlock()
331330
if atomic.LoadInt32(&c.exitFlag) == 1 {
@@ -337,7 +336,7 @@ func (c *Channel) PutMessage(msg *nsq.Message) error {
337336
}
338337

339338
// TouchMessage resets the timeout for an in-flight message
340-
func (c *Channel) TouchMessage(clientID int64, id nsq.MessageID) error {
339+
func (c *Channel) TouchMessage(clientID int64, id MessageID) error {
341340
item, err := c.popInFlightMessage(clientID, id)
342341
if err != nil {
343342
return err
@@ -362,7 +361,7 @@ func (c *Channel) TouchMessage(clientID int64, id nsq.MessageID) error {
362361
}
363362

364363
// FinishMessage successfully discards an in-flight message
365-
func (c *Channel) FinishMessage(clientID int64, id nsq.MessageID) error {
364+
func (c *Channel) FinishMessage(clientID int64, id MessageID) error {
366365
item, err := c.popInFlightMessage(clientID, id)
367366
if err != nil {
368367
return err
@@ -381,7 +380,7 @@ func (c *Channel) FinishMessage(clientID int64, id nsq.MessageID) error {
381380
// `timeoutMs` > 0 - asynchronously wait for the specified timeout
382381
// and requeue a message (aka "deferred requeue")
383382
//
384-
func (c *Channel) RequeueMessage(clientID int64, id nsq.MessageID, timeout time.Duration) error {
383+
func (c *Channel) RequeueMessage(clientID int64, id MessageID, timeout time.Duration) error {
385384
// remove from inflight first
386385
item, err := c.popInFlightMessage(clientID, id)
387386
if err != nil {
@@ -427,7 +426,7 @@ func (c *Channel) RemoveClient(clientID int64) {
427426
}
428427
}
429428

430-
func (c *Channel) StartInFlightTimeout(msg *nsq.Message, clientID int64, timeout time.Duration) error {
429+
func (c *Channel) StartInFlightTimeout(msg *Message, clientID int64, timeout time.Duration) error {
431430
now := time.Now()
432431
value := &inFlightMessage{msg, clientID, now}
433432
absTs := now.Add(timeout).UnixNano()
@@ -440,7 +439,7 @@ func (c *Channel) StartInFlightTimeout(msg *nsq.Message, clientID int64, timeout
440439
return nil
441440
}
442441

443-
func (c *Channel) StartDeferredTimeout(msg *nsq.Message, timeout time.Duration) error {
442+
func (c *Channel) StartDeferredTimeout(msg *Message, timeout time.Duration) error {
444443
absTs := time.Now().Add(timeout).UnixNano()
445444
item := &pqueue.Item{Value: msg, Priority: absTs}
446445
err := c.pushDeferredMessage(item)
@@ -452,7 +451,7 @@ func (c *Channel) StartDeferredTimeout(msg *nsq.Message, timeout time.Duration)
452451
}
453452

454453
// doRequeue performs the low level operations to requeue a message
455-
func (c *Channel) doRequeue(msg *nsq.Message) error {
454+
func (c *Channel) doRequeue(msg *Message) error {
456455
c.RLock()
457456
defer c.RUnlock()
458457
if atomic.LoadInt32(&c.exitFlag) == 1 {
@@ -468,7 +467,7 @@ func (c *Channel) pushInFlightMessage(item *pqueue.Item) error {
468467
c.Lock()
469468
defer c.Unlock()
470469

471-
id := item.Value.(*inFlightMessage).msg.Id
470+
id := item.Value.(*inFlightMessage).msg.ID
472471
_, ok := c.inFlightMessages[id]
473472
if ok {
474473
return errors.New("ID already in flight")
@@ -479,7 +478,7 @@ func (c *Channel) pushInFlightMessage(item *pqueue.Item) error {
479478
}
480479

481480
// popInFlightMessage atomically removes a message from the in-flight dictionary
482-
func (c *Channel) popInFlightMessage(clientID int64, id nsq.MessageID) (*pqueue.Item, error) {
481+
func (c *Channel) popInFlightMessage(clientID int64, id MessageID) (*pqueue.Item, error) {
483482
c.Lock()
484483
defer c.Unlock()
485484

@@ -521,7 +520,7 @@ func (c *Channel) pushDeferredMessage(item *pqueue.Item) error {
521520
defer c.Unlock()
522521

523522
// TODO: these map lookups are costly
524-
id := item.Value.(*nsq.Message).Id
523+
id := item.Value.(*Message).ID
525524
_, ok := c.deferredMessages[id]
526525
if ok {
527526
return errors.New("ID already deferred")
@@ -531,7 +530,7 @@ func (c *Channel) pushDeferredMessage(item *pqueue.Item) error {
531530
return nil
532531
}
533532

534-
func (c *Channel) popDeferredMessage(id nsq.MessageID) (*pqueue.Item, error) {
533+
func (c *Channel) popDeferredMessage(id MessageID) (*pqueue.Item, error) {
535534
c.Lock()
536535
defer c.Unlock()
537536

@@ -578,7 +577,7 @@ func (c *Channel) router() {
578577
// it is also performs in-flight accounting and initiates the auto-requeue
579578
// goroutine
580579
func (c *Channel) messagePump() {
581-
var msg *nsq.Message
580+
var msg *Message
582581
var buf []byte
583582
var err error
584583

@@ -593,7 +592,7 @@ func (c *Channel) messagePump() {
593592
select {
594593
case msg = <-c.memoryMsgChan:
595594
case buf = <-c.backend.ReadChan():
596-
msg, err = nsq.DecodeMessage(buf)
595+
msg, err = decodeMessage(buf)
597596
if err != nil {
598597
log.Printf("ERROR: failed to decode message - %s", err.Error())
599598
continue
@@ -617,8 +616,8 @@ exit:
617616

618617
func (c *Channel) deferredWorker() {
619618
c.pqWorker(&c.deferredPQ, &c.deferredMutex, func(item *pqueue.Item) {
620-
msg := item.Value.(*nsq.Message)
621-
_, err := c.popDeferredMessage(msg.Id)
619+
msg := item.Value.(*Message)
620+
_, err := c.popDeferredMessage(msg.ID)
622621
if err != nil {
623622
return
624623
}
@@ -630,7 +629,7 @@ func (c *Channel) inFlightWorker() {
630629
c.pqWorker(&c.inFlightPQ, &c.inFlightMutex, func(item *pqueue.Item) {
631630
clientID := item.Value.(*inFlightMessage).clientID
632631
msg := item.Value.(*inFlightMessage).msg
633-
_, err := c.popInFlightMessage(clientID, msg.Id)
632+
_, err := c.popInFlightMessage(clientID, msg.ID)
634633
if err != nil {
635634
return
636635
}

‎nsqd/channel_test.go

+12-13
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@ import (
88
"testing"
99
"time"
1010

11-
"github.com/bitly/go-nsq"
1211
"github.com/bmizerany/assert"
1312
)
1413

@@ -24,12 +23,12 @@ func TestPutMessage(t *testing.T) {
2423
topic := nsqd.GetTopic(topicName)
2524
channel1 := topic.GetChannel("ch")
2625

27-
var id nsq.MessageID
28-
msg := nsq.NewMessage(id, []byte("test"))
26+
var id MessageID
27+
msg := NewMessage(id, []byte("test"))
2928
topic.PutMessage(msg)
3029

3130
outputMsg := <-channel1.clientMsgChan
32-
assert.Equal(t, msg.Id, outputMsg.Id)
31+
assert.Equal(t, msg.ID, outputMsg.ID)
3332
assert.Equal(t, msg.Body, outputMsg.Body)
3433
}
3534

@@ -46,16 +45,16 @@ func TestPutMessage2Chan(t *testing.T) {
4645
channel1 := topic.GetChannel("ch1")
4746
channel2 := topic.GetChannel("ch2")
4847

49-
var id nsq.MessageID
50-
msg := nsq.NewMessage(id, []byte("test"))
48+
var id MessageID
49+
msg := NewMessage(id, []byte("test"))
5150
topic.PutMessage(msg)
5251

5352
outputMsg1 := <-channel1.clientMsgChan
54-
assert.Equal(t, msg.Id, outputMsg1.Id)
53+
assert.Equal(t, msg.ID, outputMsg1.ID)
5554
assert.Equal(t, msg.Body, outputMsg1.Body)
5655

5756
outputMsg2 := <-channel2.clientMsgChan
58-
assert.Equal(t, msg.Id, outputMsg2.Id)
57+
assert.Equal(t, msg.ID, outputMsg2.ID)
5958
assert.Equal(t, msg.Body, outputMsg2.Body)
6059
}
6160

@@ -75,7 +74,7 @@ func TestInFlightWorker(t *testing.T) {
7574
channel := topic.GetChannel("channel")
7675

7776
for i := 0; i < count; i++ {
78-
msg := nsq.NewMessage(<-nsqd.idChan, []byte("test"))
77+
msg := NewMessage(<-nsqd.idChan, []byte("test"))
7978
channel.StartInFlightTimeout(msg, 0, options.MsgTimeout)
8079
}
8180

@@ -116,14 +115,14 @@ func TestChannelEmpty(t *testing.T) {
116115
topic := nsqd.GetTopic(topicName)
117116
channel := topic.GetChannel("channel")
118117

119-
msgs := make([]*nsq.Message, 0, 25)
118+
msgs := make([]*Message, 0, 25)
120119
for i := 0; i < 25; i++ {
121-
msg := nsq.NewMessage(<-nsqd.idChan, []byte("test"))
120+
msg := NewMessage(<-nsqd.idChan, []byte("test"))
122121
channel.StartInFlightTimeout(msg, 0, options.MsgTimeout)
123122
msgs = append(msgs, msg)
124123
}
125124

126-
channel.RequeueMessage(0, msgs[len(msgs)-1].Id, 100*time.Millisecond)
125+
channel.RequeueMessage(0, msgs[len(msgs)-1].ID, 100*time.Millisecond)
127126
assert.Equal(t, len(channel.inFlightMessages), 24)
128127
assert.Equal(t, len(channel.inFlightPQ), 24)
129128
assert.Equal(t, len(channel.deferredMessages), 1)
@@ -155,7 +154,7 @@ func TestChannelEmptyConsumer(t *testing.T) {
155154
channel.AddClient(client.ID, client)
156155

157156
for i := 0; i < 25; i++ {
158-
msg := nsq.NewMessage(<-nsqd.idChan, []byte("test"))
157+
msg := NewMessage(<-nsqd.idChan, []byte("test"))
159158
channel.StartInFlightTimeout(msg, 0, options.MsgTimeout)
160159
client.SendingMessage()
161160
}

‎nsqd/client_v2.go

+17-10
Original file line numberDiff line numberDiff line change
@@ -12,11 +12,18 @@ import (
1212
"sync/atomic"
1313
"time"
1414

15-
"github.com/bitly/go-nsq"
1615
"github.com/mreiferson/go-snappystream"
1716
)
1817

19-
const DefaultBufferSize = 16 * 1024
18+
const defaultBufferSize = 16 * 1024
19+
20+
const (
21+
stateInit = iota
22+
stateDisconnected
23+
stateConnected
24+
stateSubscribed
25+
stateClosing
26+
)
2027

2128
type identifyDataV2 struct {
2229
ShortId string `json:"short_id"` // TODO: deprecated, remove in 1.0
@@ -112,10 +119,10 @@ func newClientV2(id int64, conn net.Conn, context *context) *clientV2 {
112119

113120
Conn: conn,
114121

115-
Reader: bufio.NewReaderSize(conn, DefaultBufferSize),
116-
Writer: bufio.NewWriterSize(conn, DefaultBufferSize),
122+
Reader: bufio.NewReaderSize(conn, defaultBufferSize),
123+
Writer: bufio.NewWriterSize(conn, defaultBufferSize),
117124

118-
OutputBufferSize: DefaultBufferSize,
125+
OutputBufferSize: defaultBufferSize,
119126
OutputBufferTimeout: 250 * time.Millisecond,
120127

121128
MsgTimeout: context.nsqd.options.MsgTimeout,
@@ -125,7 +132,7 @@ func newClientV2(id int64, conn net.Conn, context *context) *clientV2 {
125132
ReadyStateChan: make(chan int, 1),
126133
ExitChan: make(chan int),
127134
ConnectTime: time.Now(),
128-
State: nsq.StateInit,
135+
State: stateInit,
129136

130137
ClientID: identifier,
131138
Hostname: identifier,
@@ -304,7 +311,7 @@ func (c *clientV2) StartClose() {
304311
// Force the client into ready 0
305312
c.SetReadyCount(0)
306313
// mark this client as closing
307-
atomic.StoreInt32(&c.State, nsq.StateClosing)
314+
atomic.StoreInt32(&c.State, stateClosing)
308315
}
309316

310317
func (c *clientV2) Pause() {
@@ -419,7 +426,7 @@ func (c *clientV2) UpgradeTLS() error {
419426
}
420427
c.tlsConn = tlsConn
421428

422-
c.Reader = bufio.NewReaderSize(c.tlsConn, DefaultBufferSize)
429+
c.Reader = bufio.NewReaderSize(c.tlsConn, defaultBufferSize)
423430
c.Writer = bufio.NewWriterSize(c.tlsConn, c.OutputBufferSize)
424431

425432
atomic.StoreInt32(&c.TLS, 1)
@@ -436,7 +443,7 @@ func (c *clientV2) UpgradeDeflate(level int) error {
436443
conn = c.tlsConn
437444
}
438445

439-
c.Reader = bufio.NewReaderSize(flate.NewReader(conn), DefaultBufferSize)
446+
c.Reader = bufio.NewReaderSize(flate.NewReader(conn), defaultBufferSize)
440447

441448
fw, _ := flate.NewWriter(conn, level)
442449
c.flateWriter = fw
@@ -456,7 +463,7 @@ func (c *clientV2) UpgradeSnappy() error {
456463
conn = c.tlsConn
457464
}
458465

459-
c.Reader = bufio.NewReaderSize(snappystream.NewReader(conn, snappystream.SkipVerifyChecksum), DefaultBufferSize)
466+
c.Reader = bufio.NewReaderSize(snappystream.NewReader(conn, snappystream.SkipVerifyChecksum), defaultBufferSize)
460467
c.Writer = bufio.NewWriterSize(snappystream.NewWriter(conn), c.OutputBufferSize)
461468

462469
atomic.StoreInt32(&c.Snappy, 1)

‎nsqd/guid.go

+2-4
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,6 @@ import (
1313
"encoding/hex"
1414
"errors"
1515
"time"
16-
17-
"github.com/bitly/go-nsq"
1816
)
1917

2018
const (
@@ -63,8 +61,8 @@ func (f *guidFactory) NewGUID(workerId int64) (guid, error) {
6361
return guid(id), nil
6462
}
6563

66-
func (g guid) Hex() nsq.MessageID {
67-
var h nsq.MessageID
64+
func (g guid) Hex() MessageID {
65+
var h MessageID
6866
var b [8]byte
6967

7068
b[0] = byte(g >> 56)

0 commit comments

Comments
 (0)