set memoryMsgChan as nil when -mem-queue-size=0

nsqd/channel.go

@@ -77,11 +77,18 @@ func NewChannel(topicName string, channelName string, ctx *context,
 	c := &Channel{
 		topicName:      topicName,
 		name:           channelName,
-		memoryMsgChan:  make(chan *Message, ctx.nsqd.getOpts().MemQueueSize),
 		clients:        make(map[int64]Consumer),
 		deleteCallback: deleteCallback,
 		ctx:            ctx,
 	}
+	// Note: generally, we intend to save all the message in the disk-queue
+	// when set `-mem-queue-size=0` option, but `make(chan *Message, 0)` just
+	// create a unbuffered chan which can also send/receive message as well.
+	if ctx.nsqd.getOpts().MemQueueSize > 0 {
+		c.memoryMsgChan = make(chan *Message, ctx.nsqd.getOpts().MemQueueSize)
+	} else {
+		c.memoryMsgChan = nil
+	}
 	if len(ctx.nsqd.getOpts().E2EProcessingLatencyPercentiles) > 0 {
 		c.e2eProcessingLatencyStream = quantile.New(
 			ctx.nsqd.getOpts().E2EProcessingLatencyWindowTime,

nsqd/topic.go

@@ -47,7 +47,6 @@ func NewTopic(topicName string, ctx *context, deleteCallback func(*Topic)) *Topi
 	t := &Topic{
 		name:              topicName,
 		channelMap:        make(map[string]*Channel),
-		memoryMsgChan:     make(chan *Message, ctx.nsqd.getOpts().MemQueueSize),
 		startChan:         make(chan int, 1),
 		exitChan:          make(chan int),
 		channelUpdateChan: make(chan int),
@@ -57,6 +56,14 @@ func NewTopic(topicName string, ctx *context, deleteCallback func(*Topic)) *Topi
 		deleteCallback:    deleteCallback,
 		idFactory:         NewGUIDFactory(ctx.nsqd.getOpts().ID),
 	}
+	// Note: generally, we intend to save all the message in the disk-queue
+	// when set `-mem-queue-size=0` option, but `make(chan *Message, 0)` just
+	// create a unbuffered chan which can also send/receive message as well.
+	if ctx.nsqd.getOpts().MemQueueSize > 0 {
+		t.memoryMsgChan = make(chan *Message, ctx.nsqd.getOpts().MemQueueSize)
+	} else {
+		t.memoryMsgChan = nil
+	}
 
 	if strings.HasSuffix(topicName, "#ephemeral") {
 		t.ephemeral = true