Method chanSub

queue/queue.go:178–191

(topic string)

}

func (q *queue) chanSub(topic string) *chanSub {
	q.mu.Lock()
	defer q.mu.Unlock()
	_, ok := q.chanSubs[topic]
	if !ok {
		q.chanSubs[topic] = &chanSub{
			high:    make(chan *Message, defaultChanBuffer),
			low:     make(chan *Message, defaultLowChanBuffer),
			isClose: 0,
			done:    make(chan struct{}),
		}
	}
	return q.chanSubs[topic]
}

func (q *queue) closeTopic(topic string) {
	q.mu.Lock()