wip

nsqd/stats.go

@@ -22,8 +22,8 @@ func NewTopicStats(t *Topic, channels []ChannelStats) TopicStats {
 	return TopicStats{
 		TopicName:    t.name,
 		Channels:     channels,
-		Depth:        t.Depth(),
-		BackendDepth: 0, // TODO: (WAL) what do we do with this?
+		Depth:        0,
+		BackendDepth: t.Depth(),
 		MessageCount: atomic.LoadUint64(&t.messageCount),
 		Paused:       t.IsPaused(),
 
@@ -49,8 +49,8 @@ type ChannelStats struct {
 func NewChannelStats(c *Channel, clients []ClientStats) ChannelStats {
 	return ChannelStats{
 		ChannelName:   c.name,
-		Depth:         c.Depth(),
-		BackendDepth:  0, // TODO: (WAL) what do we do with this?
+		Depth:         0,
+		BackendDepth:  c.Depth(),
 		InFlightCount: len(c.inFlightMessages),
 		DeferredCount: len(c.deferredMessages),
 		MessageCount:  atomic.LoadUint64(&c.messageCount),

nsqd/topic.go

@@ -142,14 +142,14 @@ func (t *Topic) DeleteExistingChannel(channelName string) error {
 	return nil
 }
 
-func (t *Topic) Pub(data [][]byte) error {
+func (t *Topic) Pub(data [][]byte, crc []uint32) error {
 	t.RLock()
 	defer t.RUnlock()
 	if atomic.LoadInt32(&t.exitFlag) == 1 {
 		return errors.New("exiting")
 	}
 	// TODO: health
-	startIdx, endIdx, err := t.wal.Append(data)
+	startIdx, endIdx, err := t.wal.Append(data, crc)
 	if err != nil {
 		return err
 	}