func subscribe(sessions chan chan session, messages chan<- message) {
	queue := identity()
	for session := range sessions {
		sub := <-session
		if _, err := sub.QueueDeclare(queue, false, true, true, false, nil); err != nil {
			log.Printf("cannot consume from exclusive queue: %q, %v", queue, err)
			return
		}
		routingKey := "application specific routing key for fancy toplogies"
		if err := sub.QueueBind(queue, routingKey, exchange, false, nil); err != nil {
			log.Printf("cannot consume without a binding to exchange: %q, %v", exchange, err)
			return
		}
		deliveries, err := sub.Consume(queue, "", false, true, false, false, nil)
		if err != nil {
			log.Printf("cannot consume from: %q, %v", queue, err)
			return
		}
		log.Printf("subscribed...")
		for msg := range deliveries {
			messages <- message(msg.Body)
			sub.Ack(msg.DeliveryTag, false)
		}
	}
}
subscribe consumes deliveries from an exclusive queue from a fanout exchange and sends to the application specific messages chan.
package main
import (
	"bufio"
	"crypto/sha1"
	"flag"
	"fmt"
	"io"
	"log"
	"os"
	"github.com/streadway/amqp"
	"golang.org/x/net/context"
)
var url = flag.String("url", "amqp:///", "AMQP url for both the publisher and subscriber")
const exchange = "pubsub"
type message []byte
type session struct {
	*amqp.Connection
	*amqp.Channel
}
func (s session) Close() error {
	if s.Connection == nil {
		return nil
	}
	return s.Connection.Close()
}
func redial(ctx context.Context, url string) chan chan session {
	sessions := make(chan chan session)
	go func() {
		sess := make(chan session)
		defer close(sessions)
		for {
			select {
			case sessions <- sess:
			case <-ctx.Done():
				log.Println("shutting down session factory")
				return
			}
			conn, err := amqp.Dial(url)
			if err != nil {
				log.Fatalf("cannot (re)dial: %v: %q", err, url)
			}
			ch, err := conn.Channel()
			if err != nil {
				log.Fatalf("cannot create channel: %v", err)
			}
			if err := ch.ExchangeDeclare(exchange, "fanout", false, true, false, false, nil); err != nil {
				log.Fatalf("cannot declare fanout exchange: %v", err)
			}
			select {
			case sess <- session{conn, ch}:
			case <-ctx.Done():
				log.Println("shutting down new session")
				return
			}
		}
	}()
	return sessions
}
func publish(sessions chan chan session, messages <-chan message) {
	var (
		running bool
		reading = messages
		pending = make(chan message, 1)
		confirm = make(chan amqp.Confirmation, 1)
	)
	for session := range sessions {
		pub := <-session
		if err := pub.Confirm(false); err != nil {
			log.Printf("publisher confirms not supported")
			close(confirm)  
		} else {
			pub.NotifyPublish(confirm)
		}
		log.Printf("publishing...")
		for {
			var body message
			select {
			case confirmed := <-confirm:
				if !confirmed.Ack {
					log.Printf("nack message %d, body: %q", confirmed.DeliveryTag, string(body))
				}
				reading = messages
			case body = <-pending:
				routingKey := "ignored for fanout exchanges, application dependent for other exchanges"
				err := pub.Publish(exchange, routingKey, false, false, amqp.Publishing{
					Body: body,
				})
				if err != nil {
					pending <- body
					pub.Close()
					break
				}
			case body, running = <-reading:
				if !running {
					return
				}
				pending <- body
				reading = nil
			}
		}
	}
}
func identity() string {
	hostname, err := os.Hostname()
	h := sha1.New()
	fmt.Fprint(h, hostname)
	fmt.Fprint(h, err)
	fmt.Fprint(h, os.Getpid())
	return fmt.Sprintf("%x", h.Sum(nil))
}