/usr/share/gocode/src/github.com/streadway/amqp/_examples/simple-producer/producer.go is in golang-github-streadway-amqp-dev 0.0~git20150820.0.f4879ba-6.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | // This example declares a durable Exchange, and publishes a single message to
// that Exchange with a given routing key.
//
package main
import (
"flag"
"fmt"
"log"
"github.com/streadway/amqp"
)
var (
uri = flag.String("uri", "amqp://guest:guest@localhost:5672/", "AMQP URI")
exchangeName = flag.String("exchange", "test-exchange", "Durable AMQP exchange name")
exchangeType = flag.String("exchange-type", "direct", "Exchange type - direct|fanout|topic|x-custom")
routingKey = flag.String("key", "test-key", "AMQP routing key")
body = flag.String("body", "foobar", "Body of message")
reliable = flag.Bool("reliable", true, "Wait for the publisher confirmation before exiting")
)
func init() {
flag.Parse()
}
func main() {
if err := publish(*uri, *exchangeName, *exchangeType, *routingKey, *body, *reliable); err != nil {
log.Fatalf("%s", err)
}
log.Printf("published %dB OK", len(*body))
}
func publish(amqpURI, exchange, exchangeType, routingKey, body string, reliable bool) error {
// This function dials, connects, declares, publishes, and tears down,
// all in one go. In a real service, you probably want to maintain a
// long-lived connection as state, and publish against that.
log.Printf("dialing %q", amqpURI)
connection, err := amqp.Dial(amqpURI)
if err != nil {
return fmt.Errorf("Dial: %s", err)
}
defer connection.Close()
log.Printf("got Connection, getting Channel")
channel, err := connection.Channel()
if err != nil {
return fmt.Errorf("Channel: %s", err)
}
log.Printf("got Channel, declaring %q Exchange (%q)", exchangeType, exchange)
if err := channel.ExchangeDeclare(
exchange, // name
exchangeType, // type
true, // durable
false, // auto-deleted
false, // internal
false, // noWait
nil, // arguments
); err != nil {
return fmt.Errorf("Exchange Declare: %s", err)
}
// Reliable publisher confirms require confirm.select support from the
// connection.
if reliable {
log.Printf("enabling publishing confirms.")
if err := channel.Confirm(false); err != nil {
return fmt.Errorf("Channel could not be put into confirm mode: %s", err)
}
confirms := channel.NotifyPublish(make(chan amqp.Confirmation, 1))
defer confirmOne(confirms)
}
log.Printf("declared Exchange, publishing %dB body (%q)", len(body), body)
if err = channel.Publish(
exchange, // publish to an exchange
routingKey, // routing to 0 or more queues
false, // mandatory
false, // immediate
amqp.Publishing{
Headers: amqp.Table{},
ContentType: "text/plain",
ContentEncoding: "",
Body: []byte(body),
DeliveryMode: amqp.Transient, // 1=non-persistent, 2=persistent
Priority: 0, // 0-9
// a bunch of application/implementation-specific fields
},
); err != nil {
return fmt.Errorf("Exchange Publish: %s", err)
}
return nil
}
// One would typically keep a channel of publishings, a sequence number, and a
// set of unacknowledged sequence numbers and loop until the publishing channel
// is closed.
func confirmOne(confirms <-chan amqp.Confirmation) {
log.Printf("waiting for confirmation of one publishing")
if confirmed := <-confirms; confirmed.Ack {
log.Printf("confirmed delivery with delivery tag: %d", confirmed.DeliveryTag)
} else {
log.Printf("failed delivery of delivery tag: %d", confirmed.DeliveryTag)
}
}
|