func (n *NSQD) GetExistingTopic(topicName string, part int) (*Topic, error) {
	var err error
	var topic *Topic
	n.RLock()
	topics, ok := n.topicMap[topicName]
	if ok {
		topic, ok = topics[part]
		if !ok {
			err = ErrTopicNotExist
		}
	} else {
		err = ErrTopicNotExist
	}
	n.RUnlock()
	return topic, err
}
GetExistingTopic gets a topic only if it exists
package nsqd
import (
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"math/rand"
	"net"
	"os"
	"path"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/bitly/go-simplejson"
	"github.com/twmb/murmur3"
	"github.com/youzan/nsq/internal/clusterinfo"
	"github.com/youzan/nsq/internal/dirlock"
	"github.com/youzan/nsq/internal/http_api"
	"github.com/youzan/nsq/internal/levellogger"
	"github.com/youzan/nsq/internal/protocol"
	"github.com/youzan/nsq/internal/statsd"
	"github.com/youzan/nsq/internal/util"
	"github.com/youzan/nsq/internal/version"
	"github.com/youzan/nsq/nsqd/engine"
)
const (
	TLSNotRequired = iota
	TLSRequiredExceptHTTP
	TLSRequired
)
const (
	jobQueueChLen = 100
)
type errStore struct {
	err error
}
var (
	ErrTopicPartitionMismatch = errors.New("topic partition mismatch")
	ErrTopicNotExist          = errors.New("topic does not exist")
)
var DEFAULT_RETENTION_DAYS = 3
var EnableDelayedQueue = int32(1)
const (
	FLUSH_DISTANCE = 4
)
type INsqdNotify interface {
	NotifyDeleteTopic(*Topic)
	NotifyStateChanged(v interface{}, needPersist bool)
	ReqToEnd(*Channel, *Message, time.Duration) error
	NotifyScanChannel(c *Channel, wait bool) bool
	PushTopicJob(string, func())
}
type ReqToEndFunc func(*Channel, *Message, time.Duration) error
type NSQD struct {
	sync.RWMutex
	opts atomic.Value
	dl        *dirlock.DirLock
	isLoading int32
	errValue  atomic.Value
	startTime time.Time
	topicMap       map[string]map[int]*Topic
	magicCodeMutex sync.Mutex
	poolSize         int
	topicJobPoolSize int
	topicJobChList   [16]chan func()
	MetaNotifyChan       chan interface{}
	OptsNotificationChan chan struct{}
	exitChan             chan int
	waitGroup            util.WaitGroupWrapper
	ci               *clusterinfo.ClusterInfo
	exiting          bool
	pubLoopFunc      func(t *Topic)
	reqToEndCB       ReqToEndFunc
	scanTriggerChan  chan *Channel
	persistNotifyCh  chan struct{}
	persistClosed    chan struct{}
	persistWaitGroup util.WaitGroupWrapper
	metaStorage      IMetaStorage
	kvTopicStorage   engine.KVEngine
	sharedCfg        engine.SharedRockConfig
}
func New(opts *Options) (*NSQD, error) {
	dataPath := opts.DataPath
	if opts.DataPath == "" {
		cwd, _ := os.Getwd()
		dataPath = cwd
		opts.DataPath = dataPath
	}
	err := os.MkdirAll(dataPath, 0755)
	if err != nil {
		nsqLog.LogErrorf("failed to create directory: %v ", err)
		return nil, err
	}
	if opts.RetentionSizePerDay > 0 {
		DEFAULT_RETENTION_DAYS = int(opts.RetentionDays)
	}
	if opts.QueueReadBufferSize > 0 {
		readBufferSize = int(opts.QueueReadBufferSize)
	}
	if opts.QueueWriteBufferSize > 0 {
		writeBufSize = int(opts.QueueWriteBufferSize)
	}
	if opts.PubQueueSize > 0 {
		PubQueue = opts.PubQueueSize
	}
	n := &NSQD{
		startTime:            time.Now(),
		topicMap:             make(map[string]map[int]*Topic),
		exitChan:             make(chan int),
		MetaNotifyChan:       make(chan interface{}, 128),
		OptsNotificationChan: make(chan struct{}, 1),
		ci:                   clusterinfo.New(opts.Logger, http_api.NewClient(nil)),
		dl:                   dirlock.New(dataPath),
		scanTriggerChan:      make(chan *Channel, 100),
		persistNotifyCh:      make(chan struct{}, 2),
		persistClosed:        make(chan struct{}),
	}
	for i := 0; i < len(n.topicJobChList); i++ {
		n.topicJobChList[i] = make(chan func(), jobQueueChLen)
	}
	n.SwapOpts(opts)
	n.errValue.Store(errStore{})
	err = n.dl.Lock()
	if err != nil {
		nsqLog.LogErrorf("FATAL: --data-path=%s in use (possibly by another instance of nsqd: %v", dataPath, err)
		return nil, err
	}
	if opts.MaxDeflateLevel < 1 || opts.MaxDeflateLevel > 9 {
		nsqLog.LogErrorf("FATAL: --max-deflate-level must be [1,9]")
		return nil, errors.New("configure invalid")
	}
	if opts.ID < 0 || opts.ID >= MAX_NODE_ID {
		nsqLog.LogErrorf("FATAL: --worker-id must be [0,%d)", MAX_NODE_ID)
		return nil, errors.New("configure invalid")
	}
	nsqLog.Logf("broadcast option: %s, %s", opts.BroadcastAddress, opts.BroadcastInterface)
	n.metaStorage, err = NewShardedDBMetaStorage(path.Join(dataPath, "shared_meta"))
	if err != nil {
		nsqLog.LogErrorf("FATAL: init shared meta storage failed: %v", err.Error())
		return nil, err
	}
	if opts.KVEnabled {
		kvPath := path.Join(dataPath, "shared_kvdata")
		err = os.MkdirAll(kvPath, 0755)
		if err != nil {
			nsqLog.LogErrorf("failed to create directory %s: %s ", kvPath, err)
			return nil, err
		}
		cfg := engine.NewRockConfig()
		cfg.DisableWAL = true
		cfg.UseSharedCache = true
		cfg.UseSharedRateLimiter = true
		if opts.KVMaxWriteBufferNumber > 0 {
			cfg.MaxWriteBufferNumber = int(opts.KVMaxWriteBufferNumber)
		}
		if opts.KVWriteBufferSize > 0 {
			cfg.WriteBufferSize = int(opts.KVWriteBufferSize)
		}
		if opts.KVBlockCache > 0 {
			cfg.BlockCache = opts.KVBlockCache
		}
		sharedCfg, err := engine.NewSharedEngConfig(cfg.RockOptions)
		if err != nil {
			nsqLog.LogErrorf("failed to init engine config %v: %s ", cfg, err)
		}
		n.sharedCfg = sharedCfg
		cfg.SharedConfig = sharedCfg
		cfg.DataDir = kvPath
		engine.FillDefaultOptions(&cfg.RockOptions)
		eng, err := engine.NewKVEng(cfg)
		if err != nil {
			nsqLog.LogErrorf("failed to create engine: %s ", err)
			return nil, err
		}
		err = eng.OpenEng()
		if err != nil {
			nsqLog.LogErrorf("failed to open engine: %s ", err)
			return nil, err
		}
		n.kvTopicStorage = eng
	}
	if opts.StatsdPrefix != "" {
		var port string
		if opts.ReverseProxyPort != "" {
			port = opts.ReverseProxyPort
		} else {
			_, port, err = net.SplitHostPort(opts.HTTPAddress)
			if err != nil {
				nsqLog.LogErrorf("failed to parse HTTP address (%s) - %s", opts.HTTPAddress, err)
				return nil, err
			}
		}
		statsdHostKey := statsd.HostKey(net.JoinHostPort(opts.BroadcastAddress, port))
		prefixWithHost := strings.Replace(opts.StatsdPrefix, "%s", statsdHostKey, -1)
		if prefixWithHost[len(prefixWithHost)-1] != '.' {
			prefixWithHost += "."
		}
		opts.StatsdPrefix = prefixWithHost
		nsqLog.Infof("using the stats prefix: %v", opts.StatsdPrefix)
	}
	if opts.TLSClientAuthPolicy != "" && opts.TLSRequired == TLSNotRequired {
		opts.TLSRequired = TLSRequired
	}
	return n, nil
}
func (n *NSQD) SetReqToEndCB(reqToEndCB ReqToEndFunc) {
	n.Lock()
	n.reqToEndCB = reqToEndCB
	n.Unlock()
}
func (n *NSQD) SetPubLoop(loop func(t *Topic)) {
	n.Lock()
	n.pubLoopFunc = loop
	n.Unlock()
}
func (n *NSQD) GetOpts() *Options {
	if n.opts.Load() == nil {
		return nil
	}
	return n.opts.Load().(*Options)
}
func (n *NSQD) SwapOpts(opts *Options) {
	nsqLog.SetLevel(opts.LogLevel)
	n.opts.Store(opts)
}
func (n *NSQD) TriggerOptsNotification() {
	select {
	case n.OptsNotificationChan <- struct{}{}:
	default:
	}
}
func (n *NSQD) SetHealth(err error) {
	n.errValue.Store(errStore{err: err})
}
func (n *NSQD) IsHealthy() bool {
	return n.GetError() == nil
}
func (n *NSQD) GetError() error {
	errValue := n.errValue.Load()
	return errValue.(errStore).err
}
func (n *NSQD) GetHealth() string {
	err := n.GetError()
	if err != nil {
		return fmt.Sprintf("NOK - %s", err)
	}
	return "OK"
}
func (n *NSQD) GetStartTime() time.Time {
	return n.startTime
}
func (n *NSQD) GetTopicMapRef() map[string]map[int]*Topic {
	return n.topicMap
}
func (n *NSQD) GetTopicPartitions(topicName string) map[int]*Topic {
	tmpMap := make(map[int]*Topic)
	n.RLock()
	parts, ok := n.topicMap[topicName]
	if ok {
		for p, t := range parts {
			tmpMap[p] = t
		}
	}
	n.RUnlock()
	return tmpMap
}
func (n *NSQD) GetTopicMapCopy() []*Topic {
	n.RLock()
	tmpList := make([]*Topic, 0, len(n.topicMap)*2)
	for _, topics := range n.topicMap {
		for _, t := range topics {
			tmpList = append(tmpList, t)
		}
	}
	n.RUnlock()
	return tmpList
}
func (n *NSQD) Start() {
	n.waitGroup.Wrap(func() { n.queueScanLoop() })
	n.waitGroup.Wrap(func() { n.queueTopicJobLoop() })
	n.persistWaitGroup.Wrap(func() { n.persistLoop() })
}
func (n *NSQD) LoadMetadata(disabled int32) {
	atomic.StoreInt32(&n.isLoading, 1)
	defer atomic.StoreInt32(&n.isLoading, 0)
	fn := fmt.Sprintf(path.Join(n.GetOpts().DataPath, "nsqd.%d.dat"), n.GetOpts().ID)
	data, err := ioutil.ReadFile(fn)
	if err != nil {
		if !os.IsNotExist(err) {
			nsqLog.LogErrorf("failed to read channel metadata from %s - %s", fn, err)
		}
		return
	}
	js, err := simplejson.NewJson(data)
	if err != nil {
		nsqLog.LogErrorf("failed to parse metadata - %s", err)
		return
	}
	topics, err := js.Get("topics").Array()
	if err != nil {
		nsqLog.LogErrorf("failed to parse metadata - %s", err)
		return
	}
	for ti := range topics {
		topicJs := js.Get("topics").GetIndex(ti)
		topicName, err := topicJs.Get("name").String()
		if err != nil {
			nsqLog.LogErrorf("failed to parse metadata - %s", err)
			return
		}
		if !protocol.IsValidTopicName(topicName) {
			nsqLog.LogWarningf("skipping creation of invalid topic %s", topicName)
			continue
		}
		part, err := topicJs.Get("partition").Int()
		if err != nil {
			nsqLog.LogErrorf("failed to parse metadata - %s", err)
			return
		}
		ext, err := topicJs.Get("ext").Bool()
		if err != nil {
			nsqLog.Infof("failed to parse topic extend metadata, set to false - %s", err)
			ext = false
		}
		ordered, err := topicJs.Get("ordered").Bool()
		if err != nil {
			ordered = false
		}
		disableChannelAutoCreate, err := topicJs.Get("disbale_channel_auto_create").Bool()
		if err != nil {
			disableChannelAutoCreate = false
		}
		topic := n.internalGetTopic(topicName, part, ext, ordered, disableChannelAutoCreate, disabled)
		if topic == nil {
			nsqLog.LogErrorf("failed to init topic %v - %s", topicName, err)
			continue
		}
		channels, err := topicJs.Get("channels").Array()
		if err != nil {
			nsqLog.LogErrorf("failed to parse metadata - %s", err)
			return
		}
		for ci := range channels {
			channelJs := topicJs.Get("channels").GetIndex(ci)
			channelName, err := channelJs.Get("name").String()
			if err != nil {
				nsqLog.LogErrorf("failed to parse metadata - %s", err)
				return
			}
			if !protocol.IsValidChannelName(channelName) {
				nsqLog.LogWarningf("skipping creation of invalid channel %s", channelName)
				continue
			}
			topic.channelLock.Lock()
			channel, _ := topic.getOrCreateChannel(channelName)
			topic.channelLock.Unlock()
			paused, _ := channelJs.Get("paused").Bool()
			if paused {
				channel.Pause()
			}
			skipped, _ := channelJs.Get("skipped").Bool()
			if skipped {
				channel.Skip()
			}
			zanTestSkipped, _ := channelJs.Get("zanTestSkipped").Bool()
			if !zanTestSkipped {
				channel.UnskipZanTest()
			}
		}
		topic.LoadChannelMeta()
	}
}
func (n *NSQD) persistLoop() {
	for {
		select {
		case <-n.persistClosed:
			tmpMap := n.GetTopicMapCopy()
			n.persistMetadata(tmpMap)
			return
		case <-n.persistNotifyCh:
			tmpMap := n.GetTopicMapCopy()
			n.persistMetadata(tmpMap)
		}
	}
}
func (n *NSQD) NotifyPersistMetadata() {
	select {
	case n.persistNotifyCh <- struct{}{}:
	default:
	}
}
func (n *NSQD) persistMetadata(currentTopics []*Topic) error {
	fileName := fmt.Sprintf(path.Join(n.GetOpts().DataPath, "nsqd.%d.dat"), n.GetOpts().ID)
	nsqLog.Logf("NSQ: persisting topic/channel metadata to %s", fileName)
	defer nsqLog.Logf("NSQ: persisted metadata")
	js := make(map[string]interface{})
	topics := []interface{}{}
	for _, topic := range currentTopics {
		if topic.ephemeral {
			continue
		}
		topicData := make(map[string]interface{})
		topicData["name"] = topic.GetTopicName()
		topicData["partition"] = topic.GetTopicPart()
		topicData["ext"] = topic.IsExt()
		topicData["ordered"] = topic.IsOrdered()
		channels := []interface{}{}
		topicData["channels"] = channels
		topicData["disbale_channel_auto_create"] = topic.IsChannelAutoCreateDisabled()
		topics = append(topics, topicData)
	}
	js["version"] = version.Binary
	js["enabled_delayedqueue"] = atomic.LoadInt32(&EnableDelayedQueue)
	js["topics"] = topics
	data, err := json.Marshal(&js)
	if err != nil {
		return err
	}
	tmpFileName := fmt.Sprintf("%s.%d.tmp", fileName, rand.Int())
	f, err := os.OpenFile(tmpFileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
	if err != nil {
		return err
	}
	_, err = f.Write(data)
	if err != nil {
		f.Close()
		return err
	}
	f.Sync()
	f.Close()
	err = util.AtomicRename(tmpFileName, fileName)
	if err != nil {
		return err
	}
	return nil
}
func (n *NSQD) Exit() {
	n.Lock()
	if n.exiting {
		n.Unlock()
		return
	}
	n.exiting = true
	n.Unlock()
	close(n.persistClosed)
	n.persistWaitGroup.Wait()
	topics := n.GetTopicMapCopy()
	nsqLog.Logf("NSQ: closing topics")
	for _, topic := range topics {
		topic.Close()
	}
	close(n.exitChan)
	n.waitGroup.Wait()
	n.metaStorage.Close()
	if n.kvTopicStorage != nil {
		n.kvTopicStorage.CloseAll()
	}
	if n.sharedCfg != nil {
		n.sharedCfg.Destroy()
	}
	n.dl.Unlock()
	nsqLog.Logf("NSQ: exited")
}
func (n *NSQD) GetTopicDefaultPart(topicName string) int {
	n.RLock()
	topics, ok := n.topicMap[topicName]
	if ok {
		if len(topics) > 0 {
			for _, t := range topics {
				n.RUnlock()
				return t.GetTopicPart()
			}
		}
	}
	n.RUnlock()
	return -1
}
func (n *NSQD) GetTopicIgnPart(topicName string) *Topic {
	n.RLock()
	topics, ok := n.topicMap[topicName]
	if ok {
		if len(topics) > 0 {
			for _, t := range topics {
				n.RUnlock()
				return t
			}
		}
	}
	n.RUnlock()
	return n.GetTopic(topicName, 0, false)
}
func (n *NSQD) GetTopicWithDisabled(topicName string, part int, ext bool, ordered bool, disableChannelAutoCreate bool) *Topic {
	return n.internalGetTopic(topicName, part, ext, ordered, disableChannelAutoCreate, 1)
}
func (n *NSQD) GetTopic(topicName string, part int, ordered bool) *Topic {
	return n.internalGetTopic(topicName, part, false, false, false, 0)
}
func (n *NSQD) GetTopicWithExt(topicName string, part int, ordered bool) *Topic {
	return n.internalGetTopic(topicName, part, true, ordered, false, 0)
}
func (n *NSQD) getKVStorageForTopic(topicName string) engine.KVEngine {
	return n.kvTopicStorage
}
func (n *NSQD) internalGetTopic(topicName string, part int, ext bool, ordered bool, disableChannelAutoCreate bool, disabled int32) *Topic {
	if part > MAX_TOPIC_PARTITION || part < 0 {
		return nil
	}
	if topicName == "" {
		nsqLog.Logf("TOPIC name is empty")
		return nil
	}
	n.RLock()
	topics, ok := n.topicMap[topicName]
	if ok {
		t, ok2 := topics[part]
		if ok2 {
			n.RUnlock()
			return t
		}
	}
	n.RUnlock()
	n.Lock()
	defer n.Unlock()
	topics, ok = n.topicMap[topicName]
	if ok {
		t, ok := topics[part]
		if ok {
			return t
		}
	} else {
		topics = make(map[int]*Topic)
		n.topicMap[topicName] = topics
	}
	var t *Topic
	t = NewTopicWithExtAndDisableChannelAutoCreate(topicName, part, ext, ordered,
		disableChannelAutoCreate, n.GetOpts(), disabled, n.metaStorage,
		n.getKVStorageForTopic(topicName), n,
		n.pubLoopFunc)
	if t == nil {
		nsqLog.Errorf("TOPIC(%s): create failed", topicName)
	} else {
		topics[part] = t
		nsqLog.Logf("TOPIC(%s): created", t.GetFullName())
	}
	if t != nil {
		t.NotifyReloadChannels()
	}
	return t
}