func (a *amqpMessageQueue) Publish(topic string, message []byte) (err error) {
	ch, errChannel := a.conn.Channel()
	if errChannel != nil {
		return errChannel
	}
	err = a.declareFanoutExchange(ch)
	if err != nil {
		return err
	}
	err = ch.Publish(
		a.settings.getName("fanout"),  
		topic,  
		false,  
		false,  
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        message,
		})
	if err != nil {
		return err
	}
	return err
}
publishes message to fanout exchange
package ergoq
import (
	"fmt"
	"net/url"
	"github.com/streadway/amqp"
)
const (
	RELIABLE_URL_PARAM = "reliable"
	AUTO_ACK_URL_PARAM = "auto_ack"
	EXCHANGE_URL_PARAM = "exchange"
	DEF_AMQP_RELIABLE = true
	DEF_AMQP_AUTO_ACK = false
	DEF_AMQP_PREFIX   = "errgoq"
	DEF_EXCHANGE      = "errgo"
)
func init() {
	Register("amqp", &AMQPDriver{})
}
type AMQPDriver struct{}
func (a *AMQPDriver) Open(dsn string) (MessageQueuer, error) {
	pd, err := ParseAMQPDSN(dsn)
	if err != nil {
		return nil, err
	}
	conn, errDial := amqp.Dial(pd.dsn)
	if errDial != nil {
		return nil, errDial
	}
	d := amqpMessageQueue{
		conn:     conn,
		settings: pd.settings,
	}
	return &d, nil
}
func (a *AMQPDriver) OpenConnection(connection interface{}, settings string) (MessageQueuer, error) {
	conn, ok := connection.(*amqp.Connection)
	if !ok {
		return nil, fmt.Errorf("ergoq: amqp connection %+v not recognized.", connection)
	}
	s, err := url.ParseQuery(settings)
	if err != nil {
		return nil, err
	}
	d := amqpMessageQueue{
		conn:     conn,
		settings: NewAMQPURLSettings(s),
	}
	return &d, nil
}
type amqpMessageQueue struct {
	conn     *amqp.Connection
	settings *amqpURLSettings
}
func (a *amqpMessageQueue) declareFanoutExchange(channel *amqp.Channel) error {
	return channel.ExchangeDeclare(
		a.settings.getName("fanout"),  
		"direct",                      
		false,                         
		true,                          
		false,                         
		false,                         
		nil,                           
	)
}
func (a *amqpMessageQueue) Push(queue string, message []byte) (err error) {
	ch, err := a.conn.Channel()
	if err != nil {
		return err
	}
	defer ch.Close()
	q, errQueue := ch.QueueDeclare(
		a.settings.getName(queue),
		true,
		true,
		false,
		false,
		nil)
	if errQueue != nil {
		return errQueue
	}
	err = ch.Publish(
		"",      
		q.Name,  
		false,   
		false,   
		amqp.Publishing{
			DeliveryMode: amqp.Persistent,
			ContentType:  "text/plain",
			Body:         message,
		})
	if err != nil {
		return err
	}
	return nil
}
func (a *amqpMessageQueue) Pop(queue string) (QueueMessage, error) {
	ch, errChannel := a.conn.Channel()
	if errChannel != nil {
		return nil, errChannel
	}
	if a.settings.autoAck {
		defer ch.Close()
	}
	q, errQueue := ch.QueueDeclare(
		a.settings.getName(queue),
		true,
		true,
		false,
		false,
		nil)
	if errQueue != nil {
		return nil, errQueue
	}
	d, ok, e := ch.Get(q.Name, a.settings.autoAck)
	if e != nil {
		return nil, e
	}
	if !ok {
		return nil, fmt.Errorf("not found")
	}
	message := &amqpQueueMessage{
		settings: a.settings,
		delivery: &d,
		channel:  ch,
	}
	return message, nil
}