Bob's RabbitMQ Example

consumer.go

package main

import (
	"log"
	"os"
	"os/signal"

	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	// ch.Confirm(false)

	q, err := ch.QueueDeclare(
		"hello1", // name
		false,    // durable
		false,    // delete when unused
		false,    // exclusive
		false,    // no-wait
		nil,      // arguments
	)
	failOnError(err, "Failed to declare a queue")

	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		false,  // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	failOnError(err, "Failed to register a consumer")

	chanInt := make(chan os.Signal, 1)
	signal.Notify(chanInt, os.Interrupt)
	go func() {
		for range chanInt {
			ch.Close()
			log.Println("Channel closed")
			os.Exit(0)
		}
	}()

	go func() {
		for d := range msgs {
			log.Printf("Received a message on queue %s: %s", q.Name, d.Body)
			// break
			// d.Nack(false, true)
			d.Ack(false)
		}
	}()

	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	<-chanInt
}

publisher.go

package main

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

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	ch.Confirm(false)

	ack, nack := ch.NotifyConfirm(make(chan uint64, 1), make(chan uint64, 1))
	// _, _ = ch.NotifyConfirm(make(chan uint64, 1), make(chan uint64, 1))

	q, err := ch.QueueDeclare(
		"hello1", // name
		false,    // durable
		false,    // delete when unused
		false,    // exclusive
		false,    // no-wait
		nil,      // arguments
	)
	failOnError(err, "Failed to declare a queue")
	log.Printf("Before publishing q has %v unacked messages", q.Messages)
	body := "hello"
	err = ch.Publish(
		"",     // exchange
		q.Name, // routing key
		true,   // mandatory
		false,  // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	log.Printf(" [x] Sent %s to routing key %s", body, q.Name)
	failOnError(err, "Failed to publish a message")
	select {
	case tag := <-ack:
		log.Println("Acked ", tag)
	case tag := <-nack:
		log.Println("Nack alert! ", tag)
	}
	q2, _ := ch.QueueInspect(q.Name)
	log.Printf("After publishing q has %v unacked messages", q2.Messages)
}