rabbitmq/send.go

package rabbitmq

import (
	"github.com/streadway/amqp"
)

/**
 * 发送消息
 * @param exchange 交换机名
 * @param extype 消息类型
 * @param route 路由key
 * @param body 内容
 */
func Send(extype string, exchange string, route string, body []byte, extend ...bool) error {

	var durable, auto_delete, no_wait, internal bool = true, false, false, false

	if len(extend) > 0 {
		durable = extend[0]
	}
	if len(extend) > 1 {
		auto_delete = extend[1]
	}
	if len(extend) > 2 {
		no_wait = extend[2]
	}
	if len(extend) > 3 {
		internal = extend[3]
	}
	var err error
	err = Chan.ExchangeDeclare(
		exchange,    // name
		extype,      // type
		durable,     // durable
		auto_delete, // auto-deleted
		internal,    // internal
		no_wait,     // no-wait
		nil,         // arguments
	)
	if err != nil {
		return err
	}

	err = Chan.Publish(
		exchange, // exchange
		route,    // routing key
		false,    // mandatory
		false,    // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        body,
		})
	return err
}

/**
 * 发送消息
 * @param exchange 交换机名
 * @param extype 消息类型
 * @param route 路由key
 * @param body 内容
 * @param delay 延迟多少毫秒
 */
func SendDelay(extype string, exchange string, route string, body []byte, delay int64, extend ...bool) error {

	var durable, auto_delete, no_wait, internal bool = true, false, false, false

	if len(extend) > 0 {
		durable = extend[0]
	}
	if len(extend) > 1 {
		auto_delete = extend[1]
	}
	if len(extend) > 2 {
		no_wait = extend[2]
	}
	if len(extend) > 3 {
		internal = extend[3]
	}
	var err error
	err = Chan.ExchangeDeclare(
		exchange,                             // name
		"x-delayed-message",                  // type
		durable,                              // durable
		auto_delete,                          // auto-deleted
		internal,                             // internal
		no_wait,                              // no-wait
		amqp.Table{"x-delayed-type": extype}, // arguments
	)
	if err != nil {
		return err
	}

	err = Chan.Publish(
		exchange, // exchange
		route,    // routing key
		false,    // mandatory
		false,    // immediate
		amqp.Publishing{
			Headers:     amqp.Table{"x-delay": delay},
			ContentType: "text/plain",
			Body:        body,
		})
	return err
}