Method send

queue/queue.go:211–251

(msg *Message, timeout time.Duration)

}

func (q *queue) send(msg *Message, timeout time.Duration) (err error) {
	if q.isClosed() {
		return types.ErrChannelClosed
	}
	sub := q.chanSub(msg.Topic)
	if sub.isClose == 1 {
		return types.ErrChannelClosed
	}
	if timeout == -1 {
		select {
		case sub.high <- msg:
			return nil
		case <-sub.done:
			return types.ErrChannelClosed
		}
	}
	defer func() {
		res := recover()
		if res != nil {
			err = res.(error)
		}
	}()
	if timeout == 0 {
		select {
		case sub.high <- msg:
			return nil
		default:
			qlog.Error("send chainfull", "msg", msg, "topic", msg.Topic, "sub", sub)
			return ErrQueueChannelFull
		}
	}
	t := time.NewTimer(timeout)
	defer t.Stop()
	select {
	case sub.high <- msg:
	case <-t.C:
		qlog.Error("send timeout", "msg", msg, "topic", msg.Topic, "sub", sub)
		return ErrQueueTimeout
	}
	return nil
}

func (q *queue) sendAsyn(msg *Message) error {
	if q.isClosed() {