AMQP

Golang AMQP wrapper is a library that wraps amqp.

Check out the docs.

Features

• Auto-reconnect to brocker and auto redeclare exchanges and queues.
• Control channels lifecycle: open new on high load and close unused.
• Declarative style.

  client, err:=amqp.NewClient(
  conn.DefaultConnector("amqp://localhost:5672",
  conn.WithLogger(lg), // We want to know connection status and errors.
  ),
  amqp.TemporaryExchange("example-exchange"), // Declare exchanges and queues.amqp.PersistentExchanges(
  "exchange-one",
  "exchange-two",
  "exchange-three",
  ),
  amqp.PersistentQueues(
  "queue for one",
  "queue for two",
  "second queue for two",
  ),
  amqp.Exchange{
  Name: "declare directly",
  },
  amqp.Queue{
  Name: "", // left empty, broker generates name for you.
  },
  amqp.Binding{ // do not forget to bind queue to exchange.Exchange: "exchange-one",
  Queue: "queue for one",
  },
  amqp.WithLogger{Logger: lg}, // We want to know AMQP protocol errors.
  )

• Encoding and decoding hiden inside.
  • Use Codec interface for your format.
  • XML, JSON and Protocol Buffers (protobuf) registered yet.
• Tons of options.
  • Min and max opened channels per publisher/subscriber.
  • Limit receiving messages.
  • Any amount of data formats.
  • Fill all message fields as you wish.
  • And more others...
• Everything from AMQP may be used directly.

Contributing

We are waiting for your issue or pull request.

Example

package main
import (
"context""fmt""strconv""time""github.com/devimteam/amqp""github.com/devimteam/amqp/conn""github.com/devimteam/amqp/logger"
)
// Data, that we want to deal with.typeCommentstruct {
IdstringMessagestring
}
funcmain() {
ch:=make(chan []interface{})
// Listens errors and writes them to stdout.gofunc() {
forl:=rangech {
fmt.Println(l...)
}
}()
lg:=logger.NewChanLogger(ch) // Logger interface identical to go-kit Logger.client, err:=amqp.NewClient(
conn.DefaultConnector("amqp://localhost:5672",
conn.WithLogger(lg), // We want to know connection status and errors.
),
amqp.TemporaryExchange("example-exchange"), // Declare exchanges and queues.
amqp.WithLogger{Logger:lg}, // We want to know AMQP protocol errors.
)
iferr!=nil {
panic(err)
}
subscr:=client.Subscriber()
// context used here as closing mechanism.eventChan:=subscr.SubscribeToExchange(context.Background(),"example-exchange", Comment{}, amqp.Consumer{})
gofunc() {
forevent:=rangeeventChan {
fmt.Println(event.Data) // do something with events
}
}()
pubsr:=client.Publisher()
fori:=0; i<10; i++ {
// Prepare your data before publishingcomment:=Comment{
Id: strconv.Itoa(i),
Message: "message "+strconv.Itoa(i),
}
// Context used here for passing data to `before` functions.err:=pubsr.Publish(context.Background(), "example-exchange", comment, amqp.Publish{})
iferr!=nil {
panic(err)
}
time.Sleep(time.Millisecond*500)
}
time.Sleep(time.Second*5) // wait for delivering all messages.
}