func (bus *Bus) Push(topicName string, message interface{}) error {
	if atomic.LoadInt32(&bus.stopFlag) == _CLOSED {
		return ErrClosed
	}
	bus.rwmut.RLock()
	ch, found := bus.topics[topicName]
	bus.rwmut.RUnlock()
	if !found {
		return ErrUnvaliableTopic
	}
	ch.PostUserMessage(message)
	return nil
}
Push  push a message to topic mailbox
package eventbus
import (
	"errors"
	"sync"
	"sync/atomic"
)
const (
	_RUNNING int32 = 0
	_CLOSED        = 1
)
var (
	ErrUnvaliableTopic = errors.New("topic unvaliable.")
	ErrDuplicateTopic  = errors.New("topic duplicated.")
	ErrClosed          = errors.New("eventbus closed.")
)
type CallFunc func(message interface{})
type cmdLoadFilter struct {
	topic  string
	filter Filter
}
type cmdUnloadFilter struct {
	topic  string
	filter Filter
}
type cmdAddConsumer struct {
	topic       string
	consumerSeq uint64
	callFn      CallFunc
}
type cmdRmConsumer struct {
	consumerSeq uint64
}
type cmdStop struct {
	wg *sync.WaitGroup
}
type cmdStopGracefull struct {
	wg *sync.WaitGroup
}
var (
	defaultBus *Bus
	oncedo     sync.Once
)
func Default() *Bus {
	oncedo.Do(initDefaultBus)
	return defaultBus
}
func initDefaultBus() {
	defaultBus = New()
}
type Bus struct {
	rwmut       sync.RWMutex
	topics      map[string]*Topic  
	consumerSeq uint64
	stopFlag    int32
}
func New() *Bus {
	bus := &Bus{
		topics: make(map[string]*Topic),
	}
	return bus
}
func (bus *Bus) Subscribe(handle CallFunc, topicNames ...string) (Subscribe, error) {
	if atomic.LoadInt32(&bus.stopFlag) == _CLOSED {
		return nil, ErrClosed
	}
	var subscribe = &defaultSubscribe{
		topics: make(map[string]uint64),
		bus:    bus,
	}
	for _, topicname := range topicNames {
		seq := atomic.AddUint64(&bus.consumerSeq, 1)
		msg := cmdAddConsumer{
			topic:       topicname,
			consumerSeq: seq,
			callFn:      handle,
		}
		bus.rwmut.RLock()
		ch, found := bus.topics[topicname]
		bus.rwmut.RUnlock()
		if !found {
			ch = NewTopic(topicname, nil)
			bus.rwmut.Lock()
			bus.topics[topicname] = ch
			bus.rwmut.Unlock()
		}
		ch.PostCmdMessage(msg)
		subscribe.topics[topicname] = seq
	}
	return subscribe, nil
}
func (bus *Bus) InitTopic(topicName string, statics Statistics) error {
	if atomic.LoadInt32(&bus.stopFlag) == _CLOSED {
		return ErrClosed
	}
	bus.rwmut.RLock()
	ch, found := bus.topics[topicName]
	bus.rwmut.RUnlock()
	if !found {
		ch = NewTopic(topicName, statics)
		bus.rwmut.Lock()
		bus.topics[topicName] = ch
		bus.rwmut.Unlock()
	} else {
		return ErrDuplicateTopic
	}
	return nil
}
func (bus *Bus) unsubscribe(topicName string, id uint64) error {
	if atomic.LoadInt32(&bus.stopFlag) == _CLOSED {
		return ErrClosed
	}
	ch, found := bus.topics[topicName]
	if !found {
		return ErrUnvaliableTopic
	}
	msg := cmdRmConsumer{
		consumerSeq: id,
	}
	ch.PostCmdMessage(msg)
	return nil
}