"channel.ctx.nsqd.Notify(c)" should after "t.channelMap[channelName] = channel",

[penghuazhou]

"channel.ctx.nsqd.Notify(c)" should after "t.channelMap[channelName] = channel", otherwise perhaps it will not update channel to disk.

c.nsqd.Notify(c, !c.ephemeral)

nsq/nsqd/topic.go

Line 125 in 3103474

t.channelMap[channelName] = channel

[ljluestc]

package nsqd

import (
	"context"
	"sync"
	"time"

	"gl.zego.im/goserver/zg-nsq/nsq/internal/protocol"
)

// NSQD struct as per NSQ v1.0.0-compat
type NSQD struct {
	sync.RWMutex
	// Other fields...
	topicMap map[string]*Topic
	// Other fields...
}

// Topic and Channel structs (simplified)
type Topic struct {
	sync.RWMutex
	name      string
	channelMap map[string]*Channel
	// Other fields...
}

type Channel struct {
	sync.RWMutex
	name string
	// Other fields...
}

// StatsData holds data for stats output
type StatsData struct {
	Topics []TopicStats
}

type TopicStats struct {
	Name     string
	Channels []ChannelStats
}

type ChannelStats struct {
	Name string
	// Other fields...
}

// GetStats retrieves NSQD statistics with a timeout
func (n *NSQD) GetStats(ctx context.Context, topic string, detailed bool) (*StatsData, error) {
	// Set a timeout for the operation
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	default:
	}

	start := time.Now()
	n.RLock()
	// Copy topic names to minimize lock duration
	topics := make([]string, 0, len(n.topicMap))
	for name := range n.topicMap {
		if topic == "" || name == topic {
			topics = append(topics, name)
		}
	}
	n.RUnlock()

	// Sort topics for deterministic output
	sort.Strings(topics)

	// Collect stats outside the lock
	data := &StatsData{Topics: make([]TopicStats, 0, len(topics))}
	for _, topicName := range topics {
		n.RLock()
		topic, exists := n.topicMap[topicName]
		if !exists {
			n.RUnlock()
			continue
		}
		// Copy channel names
		channels := make([]string, 0, len(topic.channelMap))
		topic.RLock()
		for chName := range topic.channelMap {
			channels = append(channels, chName)
		}
		topic.RUnlock()
		n.RUnlock()

		// Sort channels for deterministic output
		sort.Strings(channels)

		topicStats := TopicStats{Name: topicName}
		if detailed {
			topicStats.Channels = make([]ChannelStats, 0, len(channels))
			for _, chName := range channels {
				n.RLock()
				topic, exists := n.topicMap[topicName]
				if !exists {
					n.RUnlock()
					continue
				}
				topic.RLock()
				channel, exists := topic.channelMap[chName]
				if !exists {
					topic.RUnlock()
					n.RUnlock()
					continue
				}
				// Collect channel stats (simplified)
				chStats := ChannelStats{Name: chName}
				topicStats.Channels = append(topicStats.Channels, chStats)
				topic.RUnlock()
				n.RUnlock()
			}
		}
		data.Topics = append(data.Topics, topicStats)
	}

	// Log slow operations
	if dur := time.Since(start); dur > 500*time.Millisecond {
		n.logf("WARNING: GetStats took %v", dur)
	}

	return data, nil
}