func (t *Topic) getOrCreateChannel(channelName string) (*Channel, bool) {
	channel, ok := t.channelMap[channelName]
	if !ok {
		deleteCallback := func(c *Channel) {
			t.DeleteExistingChannel(c.name)
		}
		readEnd := t.backend.GetQueueReadEnd()
		curCommit := t.GetCommitted()
		if curCommit != nil && readEnd.Offset() > curCommit.Offset() {
			if t.tpLog.Level() >= levellogger.LOG_DEBUG {
				t.tpLog.Logf("channel %v, end to commit: %v, read end: %v", channelName, curCommit, readEnd)
			}
			readEnd = curCommit
		}
		var ext int32
		if t.IsExt() {
			ext = 1
		} else {
			ext = 0
		}
		start := t.backend.GetQueueReadStart()
		channel = NewChannel(t.GetTopicName(), t.GetTopicPart(), t.IsOrdered(), channelName, readEnd,
			t.option, deleteCallback, t.flushForChannelMoreData, atomic.LoadInt32(&t.writeDisabled),
			t.nsqdNotify, ext, start, t.metaStorage, t.kvTopic, !t.IsDataNeedFix())
		channel.UpdateQueueEnd(readEnd, false)
		channel.SetDelayedQueue(t.GetDelayedQueue())
		if t.IsWriteDisabled() {
			channel.DisableConsume(true)
		}
		t.channelMap[channelName] = channel
		t.tpLog.Logf("TOPIC new channel(%s), end: %v",
			channel.name, channel.GetChannelEnd())
		return channel, true
	}
	return channel, false
}
this expects the caller to handle locking
package nsqd
import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"os"
	"path"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/youzan/nsq/internal/ext"
	"github.com/youzan/nsq/internal/levellogger"
	"github.com/youzan/nsq/internal/protocol"
	"github.com/youzan/nsq/internal/quantile"
	"github.com/youzan/nsq/internal/util"
	"github.com/youzan/nsq/nsqd/engine"
)
const (
	MAX_TOPIC_PARTITION    = 1023
	HISTORY_STAT_FILE_NAME = ".stat.history.dat"
	slowCost               = time.Millisecond * 50
)
var (
	ErrInvalidMessageID           = errors.New("message id is invalid")
	ErrWriteOffsetMismatch        = errors.New("write offset mismatch")
	ErrOperationInvalidState      = errors.New("the operation is not allowed under current state")
	ErrMessageInvalidDelayedState = errors.New("the message is invalid for delayed")
	PubQueue                      = 500
	errChannelNotExist            = errors.New("channel does not exist")
)
func writeMessageToBackend(writeExt bool, buf *bytes.Buffer, msg *Message, bq *diskQueueWriter) (BackendOffset, int32, diskQueueEndInfo, error) {
	buf.Reset()
	_, err := msg.WriteTo(buf, writeExt)
	if err != nil {
		return 0, 0, diskQueueEndInfo{}, err
	}
	return bq.PutV2(buf.Bytes())
}
func writeMessageToBackendWithCheck(writeExt bool, buf *bytes.Buffer, msg *Message, checkSize int64, bq *diskQueueWriter) (BackendOffset, int32, diskQueueEndInfo, error) {
	buf.Reset()
	wsize, err := msg.WriteTo(buf, writeExt)
	if err != nil {
		return 0, 0, diskQueueEndInfo{}, err
	}
	if checkSize > 0 && wsize+4 != checkSize {
		return 0, 0, diskQueueEndInfo{}, fmt.Errorf("message write size mismatch %v vs %v", checkSize, wsize+4)
	}
	return bq.PutV2(buf.Bytes())
}
type MsgIDGenerator interface {
	NextID() uint64
	Reset(uint64)
}
type TopicDynamicConf struct {
	AutoCommit               int32
	RetentionDay             int32
	SyncEvery                int64
	OrderedMulti             bool
	MultiPart                bool
	DisableChannelAutoCreate bool
	Ext                      bool
}
type PubInfo struct {
	Done       chan struct{}
	MsgBody    []byte
	ExtContent ext.IExtContent
	StartPub   time.Time
	Err        error
}
type PubInfoChan chan *PubInfo
type MPubInfo struct {
	Done     chan struct{}
	Msgs     []*Message
	StartPub time.Time
	Err      error
}
type MPubInfoChan chan *MPubInfo
type ChannelMetaInfo struct {
	Name           string `json:"name"`
	Paused         bool   `json:"paused"`
	Skipped        bool   `json:"skipped"`
	ZanTestSkipped bool   `json:"zanTestSkipped"`
}
func (cm *ChannelMetaInfo) IsZanTestSkipepd() bool {
	return cm.ZanTestSkipped
}
type Topic struct {
	sync.Mutex
	tname       string
	fullName    string
	partition   int
	channelMap  map[string]*Channel
	channelLock sync.RWMutex
	backend     *diskQueueWriter
	dataPath    string
	flushChan   chan int
	exitFlag    int32
	ephemeral bool
	deleter   sync.Once
	nsqdNotify      INsqdNotify
	option          *Options
	msgIDCursor     MsgIDGenerator
	defaultIDSeq    uint64
	needFlush       int32
	EnableTrace     int32
	lastSyncCnt     int64
	putBuffer       bytes.Buffer
	bp              sync.Pool
	writeDisabled   int32
	dynamicConf     *TopicDynamicConf
	isOrdered       int32
	magicCode       int64
	committedOffset atomic.Value
	detailStats     *DetailStatsInfo
	needFixData     int32
	pubWaitingChan  PubInfoChan
	mpubWaitingChan MPubInfoChan
	quitChan        chan struct{}
	pubLoopFunc     func(v *Topic)
	wg              sync.WaitGroup
	delayedQueue                 atomic.Value
	isExt                        int32
	isChannelAutoCreatedDisabled int32
	pubFailedCnt                 int64
	metaStorage                  IMetaStorage
	pubWaitingBytes int64
	tpLog           *levellogger.LevelLogger
	kvTopic         *KVTopic
}
func (t *Topic) setExt() {
	atomic.StoreInt32(&t.isExt, 1)
}
func (t *Topic) IsExt() bool {
	return atomic.LoadInt32(&t.isExt) == 1
}
func (t *Topic) DisableChannelAutoCreate() {
	atomic.StoreInt32(&t.isChannelAutoCreatedDisabled, 1)
}
func (t *Topic) EnableChannelAutoCreate() {
	atomic.StoreInt32(&t.isChannelAutoCreatedDisabled, 0)
}
func (t *Topic) IsChannelAutoCreateDisabled() bool {
	return atomic.LoadInt32(&t.isChannelAutoCreatedDisabled) == 1
}
func (t *Topic) IncrPubWaitingBytes(sz int64) int64 {
	return atomic.AddInt64(&t.pubWaitingBytes, sz)
}
func (t *Topic) IncrPubFailed() {
	atomic.AddInt64(&t.pubFailedCnt, 1)
	TopicPubFailedCnt.With(prometheus.Labels{
		"topic":     t.GetTopicName(),
		"partition": strconv.Itoa(t.GetTopicPart()),
	}).Inc()
}
func (t *Topic) PubFailed() int64 {
	return atomic.LoadInt64(&t.pubFailedCnt)
}
func GetTopicFullName(topic string, part int) string {
	return topic + "-" + strconv.Itoa(part)
}
func NewTopicForDelete(topicName string, part int, opt *Options,
	writeDisabled int32, metaStorage IMetaStorage,
	notify INsqdNotify, loopFunc func(v *Topic)) *Topic {
	return NewTopicWithExt(topicName, part, false, false, opt, writeDisabled, metaStorage, nil, notify, loopFunc)
}
func NewTopicWithExt(topicName string, part int, ext bool, ordered bool, opt *Options,
	writeDisabled int32, metaStorage IMetaStorage,
	kvEng engine.KVEngine,
	notify INsqdNotify, loopFunc func(v *Topic)) *Topic {
	return NewTopicWithExtAndDisableChannelAutoCreate(topicName, part, ext, ordered, false, opt, writeDisabled, metaStorage, kvEng, notify, loopFunc)
}
func NewTopicWithExtAndDisableChannelAutoCreate(topicName string, part int, ext bool, ordered bool,
	disbaleChannelAutoCreate bool, opt *Options,
	writeDisabled int32, metaStorage IMetaStorage,
	kvEng engine.KVEngine,
	notify INsqdNotify, loopFunc func(v *Topic)) *Topic {
	if part > MAX_TOPIC_PARTITION {
		return nil
	}
	t := &Topic{
		tname:           topicName,
		partition:       part,
		channelMap:      make(map[string]*Channel),
		flushChan:       make(chan int, 10),
		option:          opt,
		dynamicConf:     &TopicDynamicConf{SyncEvery: opt.SyncEvery, AutoCommit: 1},
		putBuffer:       bytes.Buffer{},
		nsqdNotify:      notify,
		writeDisabled:   writeDisabled,
		pubWaitingChan:  make(PubInfoChan, PubQueue),
		mpubWaitingChan: make(MPubInfoChan, PubQueue/2),
		quitChan:        make(chan struct{}),
		pubLoopFunc:     loopFunc,
		metaStorage:     metaStorage,
		kvTopic:         NewKVTopicWithEngine(topicName, part, ext, kvEng),
	}
	t.fullName = GetTopicFullName(t.tname, t.partition)
	t.tpLog = nsqLog.WrappedWithPrefix("["+t.fullName+"]", 0)
	if metaStorage == nil {
		t.metaStorage = &fileMetaStorage{}
	}
	if ext {
		t.setExt()
	}
	if ordered {
		atomic.StoreInt32(&t.isOrdered, 1)
	}
	if t.dynamicConf.SyncEvery < 1 {
		t.dynamicConf.SyncEvery = 1
	}
	if disbaleChannelAutoCreate {
		t.DisableChannelAutoCreate()
	}
	t.bp.New = func() interface{} {
		return &bytes.Buffer{}
	}
	t.dataPath = path.Join(opt.DataPath, topicName)
	err := os.MkdirAll(t.dataPath, 0755)
	if err != nil {
		t.tpLog.LogErrorf("failed to create directory: %v ", err)
		return nil
	}
	backendName := getBackendName(t.tname, t.partition)
	queue, err := newDiskQueueWriterWithMetaStorage(backendName,
		t.dataPath,
		opt.MaxBytesPerFile,
		int32(minValidMsgLength),
		int32(opt.MaxMsgSize)+minValidMsgLength,
		opt.SyncEvery,
		t.metaStorage,
	)
	if err != nil {
		t.tpLog.LogErrorf("failed to init disk queue: %v ", err)
		if err == ErrNeedFixQueueStart || err == ErrNeedFixQueueEnd {
			t.SetDataFixState(true)
		} else {
			return nil
		}
	}
	t.backend = queue.(*diskQueueWriter)
	if err != ErrNeedFixQueueEnd {
		t.UpdateCommittedOffset(t.backend.GetQueueWriteEnd())
	}
	if _, err := t.tryFixKVTopic(); err != nil {
		t.tpLog.LogErrorf("failed to auto fix kv topic while init: %s", err)
		t.SetDataFixState(true)
	}
	err = t.loadMagicCode()
	if err != nil {
		t.tpLog.LogErrorf("failed to load magic code: %v", err)
		return nil
	}
	t.detailStats = NewDetailStatsInfo(t.tname, strconv.Itoa(t.partition), t.TotalDataSize(), t.getHistoryStatsFileName())
	t.nsqdNotify.NotifyStateChanged(t, true)
	t.tpLog.LogDebugf("new topic created: %v", t.tname)
	if t.pubLoopFunc != nil {
		t.wg.Add(1)
		go func() {
			defer t.wg.Done()
			t.pubLoopFunc(t)
		}()
	}
	t.LoadChannelMeta()
	return t
}
func (t *Topic) GetDelayedQueue() *DelayQueue {
	if t.IsOrdered() {
		return nil
	}
	if t.delayedQueue.Load() == nil {
		return nil
	}
	return t.delayedQueue.Load().(*DelayQueue)
}
func (t *Topic) GetOrCreateDelayedQueueForReadNoLock() (*DelayQueue, error) {
	if t.delayedQueue.Load() == nil {
		delayedQueue, err := NewDelayQueueForRead(t.tname, t.partition, t.dataPath, t.option, nil, t.IsExt())
		if err == nil {
			t.delayedQueue.Store(delayedQueue)
		} else {
			t.tpLog.LogWarningf("init delayed queue error %v", err)
			return nil, err
		}
	}
	return t.delayedQueue.Load().(*DelayQueue), nil
}
func (t *Topic) GetOrCreateDelayedQueueNoLock(idGen MsgIDGenerator) (*DelayQueue, error) {
	if t.delayedQueue.Load() == nil {
		delayedQueue, err := NewDelayQueue(t.tname, t.partition, t.dataPath, t.option, idGen, t.IsExt())
		if err == nil {
			t.delayedQueue.Store(delayedQueue)
			t.channelLock.RLock()
			for _, ch := range t.channelMap {
				ch.SetDelayedQueue(delayedQueue)
			}
			t.channelLock.RUnlock()
		} else {
			t.tpLog.LogWarningf("init delayed queue error %v", err)
			return nil, err
		}
	}
	return t.delayedQueue.Load().(*DelayQueue), nil
}
func (t *Topic) GetWaitChan() PubInfoChan {
	return t.pubWaitingChan
}
func (t *Topic) IsWaitChanFull() bool {
	return len(t.pubWaitingChan) >= cap(t.pubWaitingChan)-1
}
func (t *Topic) IsMWaitChanFull() bool {
	return len(t.mpubWaitingChan) >= cap(t.mpubWaitingChan)-1
}
func (t *Topic) GetMWaitChan() MPubInfoChan {
	return t.mpubWaitingChan
}
func (t *Topic) QuitChan() <-chan struct{} {
	return t.quitChan
}
func (t *Topic) IsDataNeedFix() bool {
	return atomic.LoadInt32(&t.needFixData) == 1
}
func (t *Topic) SetDataFixState(needFix bool) {
	if needFix {
		atomic.StoreInt32(&t.needFixData, 1)
	} else {
		atomic.StoreInt32(&t.needFixData, 0)
	}
}
func (t *Topic) getMagicCodeFileName() string {
	return path.Join(t.dataPath, "magic"+strconv.Itoa(t.partition))
}
func (t *Topic) saveMagicCode() error {
	var f *os.File
	var err error
	fileName := t.getMagicCodeFileName()
	f, err = os.OpenFile(fileName, os.O_RDWR|os.O_CREATE, 0644)
	if err != nil {
		return err
	}
	defer f.Close()
	_, err = fmt.Fprintf(f, "%d\n",
		atomic.LoadInt64(&t.magicCode))
	if err != nil {
		return err
	}
	f.Sync()
	t.tpLog.Infof("saved as magic code: %v", atomic.LoadInt64(&t.magicCode))
	return nil
}
func (t *Topic) removeMagicCode() {
	fileName := t.getMagicCodeFileName()
	err := os.Remove(fileName)
	if err != nil {
		t.tpLog.Infof("remove the magic file %v failed:%v", fileName, err)
	}
}
func (t *Topic) loadMagicCode() error {
	var f *os.File
	var err error
	fileName := t.getMagicCodeFileName()
	f, err = os.OpenFile(fileName, os.O_RDONLY, 0644)
	if err != nil {
		if os.IsNotExist(err) {
			return nil
		}
		return err
	}
	defer f.Close()
	var code int64
	_, err = fmt.Fscanf(f, "%d\n",
		&code)
	if err != nil {
		return err
	}
	atomic.StoreInt64(&t.magicCode, code)
	t.tpLog.Infof("loading topic as magic code: %v", code)
	return nil
}
func (t *Topic) removeHistoryStat() {
	fileName := t.getHistoryStatsFileName()
	err := os.Remove(fileName)
	if err != nil {
		t.tpLog.Infof("remove file %v failed:%v", fileName, err)
	}
}
func (t *Topic) MarkAsRemoved() (string, error) {
	t.Lock()
	defer t.Unlock()
	atomic.CompareAndSwapInt32(&t.exitFlag, 0, 1)
	t.tpLog.Logf("TOPIC(%s): deleting", t.GetFullName())
	t.nsqdNotify.NotifyStateChanged(t, true)
	t.channelLock.Lock()
	for _, channel := range t.channelMap {
		t.channelMap[channel.name] = nil
		delete(t.channelMap, channel.name)
		channel.Delete()
	}
	t.channelLock.Unlock()
	renamePath := t.dataPath + "-removed-" + strconv.Itoa(int(time.Now().Unix()))
	t.tpLog.Warningf("mark the topic as removed: %v", renamePath)
	os.MkdirAll(renamePath, 0755)
	err := t.backend.RemoveTo(renamePath)
	if err != nil {
		t.tpLog.Errorf("failed to mark the topic as removed %v failed: %v", renamePath, err)
	}
	util.AtomicRename(t.getMagicCodeFileName(), path.Join(renamePath, "magic"+strconv.Itoa(t.partition)))
	t.removeHistoryStat()
	t.RemoveChannelMeta()
	t.removeMagicCode()
	if t.GetDelayedQueue() != nil {
		t.GetDelayedQueue().Delete()
	}
	return renamePath, err
}
func (t *Topic) SetMagicCode(code int64) error {
	if t.magicCode == code {
		return nil
	} else if t.magicCode > 0 {
		t.tpLog.Errorf("magic code set more than once :%v, %v", t.magicCode, code)
		return errors.New("magic code can not be changed.")
	}
	t.magicCode = code
	return t.saveMagicCode()
}
func (t *Topic) GetMagicCode() int64 {
	return t.magicCode
}
func (t *Topic) SetTrace(enable bool) {
	if enable {
		atomic.StoreInt32(&t.EnableTrace, 1)
	} else {
		atomic.StoreInt32(&t.EnableTrace, 0)
	}
	if t.GetDelayedQueue() != nil {
		t.GetDelayedQueue().SetTrace(enable)
	}
}
func (t *Topic) getChannelMetaFileName() string {
	return path.Join(t.dataPath, "channel_meta"+strconv.Itoa(t.partition))
}
func (t *Topic) LoadChannelMeta() error {
	fn := t.getChannelMetaFileName()
	channels, err := t.metaStorage.LoadChannelMeta(fn)
	if err != nil {
		if IsMetaNotFound(err) {
			return nil
		}
		t.tpLog.LogWarningf("failed to load metadata - %s", err)
		return err
	}
	for _, ch := range channels {
		channelName := ch.Name
		if !protocol.IsValidChannelName(channelName) {
			t.tpLog.LogWarningf("skipping creation of invalid channel %s", channelName)
			continue
		}
		t.channelLock.Lock()
		channel, _ := t.getOrCreateChannel(channelName)
		t.channelLock.Unlock()
		if ch.Paused {
			channel.Pause()
		}
		if ch.Skipped {
			channel.Skip()
		}
		if !ch.IsZanTestSkipepd() {
			channel.UnskipZanTest()
		}
	}
	return nil
}
func (t *Topic) UpdateChannelMeta(ch *Channel, paused int, skipped int, zanTestSkipped int) error {
	switch paused {
	case 1:
		ch.Pause()
	case 0:
		ch.UnPause()
	}
	switch skipped {
	case 1:
		ch.Skip()
	case 0:
		ch.UnSkip()
	}
	switch int32(zanTestSkipped) {
	case ZanTestSkip:
		ch.SkipZanTest()
	case ZanTestUnskip:
		ch.UnskipZanTest()
	}
	return t.SaveChannelMeta()
}
func (t *Topic) GetChannelMeta() []ChannelMetaInfo {
	t.channelLock.RLock()
	channels := make([]ChannelMetaInfo, 0, len(t.channelMap))
	for _, channel := range t.channelMap {
		channel.RLock()
		if !channel.ephemeral {
			meta := ChannelMetaInfo{
				Name:           channel.name,
				Paused:         channel.IsPaused(),
				Skipped:        channel.IsSkipped(),
				ZanTestSkipped: channel.IsZanTestSkipped(),
			}
			channels = append(channels, meta)
		}
		channel.RUnlock()
	}
	t.channelLock.RUnlock()
	return channels
}
func (t *Topic) SaveChannelMeta() error {
	channels := make([]*ChannelMetaInfo, 0)
	t.channelLock.RLock()
	for _, channel := range t.channelMap {
		channel.RLock()
		if !channel.ephemeral {
			meta := &ChannelMetaInfo{
				Name:           channel.name,
				Paused:         channel.IsPaused(),
				Skipped:        channel.IsSkipped(),
				ZanTestSkipped: channel.IsZanTestSkipped(),
			}
			channels = append(channels, meta)
		}
		channel.RUnlock()
	}
	t.channelLock.RUnlock()
	fileName := t.getChannelMetaFileName()
	err := t.metaStorage.SaveChannelMeta(fileName, t.option.UseFsync, channels)
	if err != nil {
		return err
	}
	return nil
}
func (t *Topic) RemoveChannelMeta() {
	fileName := t.getChannelMetaFileName()
	t.metaStorage.RemoveChannelMeta(fileName)
}
func (t *Topic) getHistoryStatsFileName() string {
	return path.Join(t.dataPath, t.fullName+HISTORY_STAT_FILE_NAME)
}
func (t *Topic) GetDetailStats() *DetailStatsInfo {
	return t.detailStats
}
func (t *Topic) SaveHistoryStats() error {
	if t.Exiting() {
		return ErrExiting
	}
	t.Lock()
	defer t.Unlock()
	return t.detailStats.SaveHistory(t.getHistoryStatsFileName())
}
func (t *Topic) LoadHistoryStats() error {
	return t.detailStats.LoadHistory(t.getHistoryStatsFileName())
}
func (t *Topic) GetCommitted() BackendQueueEnd {
	l := t.committedOffset.Load()
	if l == nil {
		return nil
	}
	return l.(BackendQueueEnd)
}
func (t *Topic) UpdateCommittedOffset(offset BackendQueueEnd) {
	if offset == nil {
		return
	}
	cur := t.GetCommitted()
	if cur != nil && offset.Offset() < cur.Offset() {
		t.tpLog.LogDebugf("committed is rollbacked: %v, %v", cur, offset)
	}
	t.committedOffset.Store(offset)
	syncEvery := atomic.LoadInt64(&t.dynamicConf.SyncEvery)
	if syncEvery == 1 ||
		offset.TotalMsgCnt()-atomic.LoadInt64(&t.lastSyncCnt) >= syncEvery {
		if !t.IsWriteDisabled() {
			t.flushBuffer(true)
		}
	} else {
		t.notifyChEndChanged(false)
	}
}
func (t *Topic) GetDiskQueueSnapshot(checkCommit bool) *DiskQueueSnapshot {
	e := t.backend.GetQueueReadEnd()
	if checkCommit {
		commit := t.GetCommitted()
		if commit != nil && e.Offset() > commit.Offset() {
			e = commit
		}
	}
	start := t.backend.GetQueueReadStart()
	d := NewDiskQueueSnapshot(getBackendName(t.tname, t.partition), t.dataPath, e)
	d.SetQueueStart(start)
	return d
}
func (t *Topic) BufferPoolGet(capacity int) *bytes.Buffer {
	b := t.bp.Get().(*bytes.Buffer)
	b.Reset()
	b.Grow(capacity)
	return b
}
func (t *Topic) BufferPoolPut(b *bytes.Buffer) {
	t.bp.Put(b)
}
func (t *Topic) GetChannelMapCopy() map[string]*Channel {
	tmpMap := make(map[string]*Channel)
	t.channelLock.RLock()
	for k, v := range t.channelMap {
		tmpMap[k] = v
	}
	t.channelLock.RUnlock()
	return tmpMap
}
func (t *Topic) Exiting() bool {
	return atomic.LoadInt32(&t.exitFlag) == 1
}
func (t *Topic) GetDynamicInfo() TopicDynamicConf {
	t.Lock()
	info := *t.dynamicConf
	t.Unlock()
	return info
}
func (t *Topic) IsOrdered() bool {
	return atomic.LoadInt32(&t.isOrdered) == 1
}
func (t *Topic) SetDynamicInfo(dynamicConf TopicDynamicConf, idGen MsgIDGenerator) {
	t.Lock()
	if idGen != nil {
		t.msgIDCursor = idGen
	}
	atomic.StoreInt64(&t.dynamicConf.SyncEvery, dynamicConf.SyncEvery)
	atomic.StoreInt32(&t.dynamicConf.AutoCommit, dynamicConf.AutoCommit)
	atomic.StoreInt32(&t.dynamicConf.RetentionDay, dynamicConf.RetentionDay)
	t.dynamicConf.OrderedMulti = dynamicConf.OrderedMulti
	if dynamicConf.OrderedMulti {
		atomic.StoreInt32(&t.isOrdered, 1)
	} else {
		atomic.StoreInt32(&t.isOrdered, 0)
	}
	t.dynamicConf.MultiPart = dynamicConf.MultiPart
	dq := t.GetDelayedQueue()
	if dq != nil {
		atomic.StoreInt64(&dq.SyncEvery, dynamicConf.SyncEvery)
		if dynamicConf.Ext {
			dq.setExt()
		}
	}
	t.dynamicConf.Ext = dynamicConf.Ext
	if dynamicConf.Ext {
		t.setExt()
		if t.kvTopic != nil {
			t.kvTopic.setExt()
		}
	}
	t.dynamicConf.DisableChannelAutoCreate = dynamicConf.DisableChannelAutoCreate
	channelAutoCreateDisabled := t.IsChannelAutoCreateDisabled()
	if dynamicConf.DisableChannelAutoCreate && !channelAutoCreateDisabled {
		t.DisableChannelAutoCreate()
	} else if !dynamicConf.DisableChannelAutoCreate && channelAutoCreateDisabled {
		t.EnableChannelAutoCreate()
	}
	t.tpLog.Logf("topic dynamic configure changed to %v", dynamicConf)
	t.channelLock.RLock()
	for _, ch := range t.channelMap {
		ext := dynamicConf.Ext
		ch.SetExt(ext)
	}
	t.channelLock.RUnlock()
	t.Unlock()
	t.nsqdNotify.NotifyStateChanged(t, true)
}
func (t *Topic) nextMsgID() MessageID {
	id := uint64(0)
	if t.msgIDCursor != nil {
		id = t.msgIDCursor.NextID()
	} else {
		id = atomic.AddUint64(&t.defaultIDSeq, 1)
	}
	return MessageID(id)
}
func (t *Topic) GetFullName() string {
	return t.fullName
}
func (t *Topic) GetTopicName() string {
	return t.tname
}
func (t *Topic) GetTopicPart() int {
	return t.partition
}
func (t *Topic) GetChannel(channelName string) *Channel {
	t.channelLock.Lock()
	channel, isNew := t.getOrCreateChannel(channelName)
	t.channelLock.Unlock()
	if isNew {
		if !channel.IsEphemeral() {
			t.SaveChannelMeta()
		}
		t.NotifyReloadChannels()
	}
	return channel
}
func (t *Topic) NotifyReloadChannels() {
}
func (t *Topic) GetTopicChannelDebugStat(channelName string) string {
	statStr := ""
	t.channelLock.RLock()
	for n, channel := range t.channelMap {
		if channelName == "" || channelName == n {
			statStr += channel.GetChannelDebugStats()
		}
	}
	t.channelLock.RUnlock()
	dq := t.GetDelayedQueue()
	if dq != nil {
		t.tpLog.Logf("delayed queue stats: %v", dq.Stats())
	}
	return statStr
}