AMQP Crystal Client

The recommended Crystal client is amqp-client.cr. It’s fast, depends only on the amq-protocol shard, and comes from the team behind LavinMQ, which is also written in Crystal. Add it to shard.yml and run shards install:

dependencies:
  amqp-client:
    github: cloudamqp/amqp-client.cr

Publish and consume

The example reads the broker URL from BROKER_URL and falls back to a local LavinMQ at amqp://guest:guest@localhost.

Save the connection helper as src/example.cr:

require "amqp-client"

module Example
  # Connects to BROKER_URL. TLS connections (amqps://) verify the server certificate.
  # The client doesn't reconnect: restart the process or reconnect yourself if the connection is lost.
  def self.connect : AMQP::Client::Connection
    AMQP::Client.new(ENV.fetch("BROKER_URL", "amqp://guest:guest@localhost")).connect
  end

  # Publishes a persistent message and blocks until the broker has confirmed it
  def self.publish(ch : AMQP::Client::Channel, queue : String, message : String) : Nil
    q = ch.queue(queue, durable: true)
    props = AMQP::Client::Properties.new(content_type: "text/plain", delivery_mode: 2_u8)
    q.publish_confirm(message, props: props) || raise "Message was not confirmed"
    puts " [x] Sent '#{message}'"
  end

  # Consumes with a prefetch of 10, acking each message after the block returns
  def self.consume(ch : AMQP::Client::Channel, queue : String, &handler : String ->) : Nil
    ch.prefetch(10)
    ch.queue(queue, durable: true).subscribe(no_ack: false) do |msg|
      handler.call(msg.body_io.to_s)
      msg.ack
    end
  end
end

Save the publisher as src/publisher.cr:

require "./example"

conn = Example.connect
Example.publish(conn.channel, "hello", "Hello world!")
conn.close

Save the consumer as src/worker.cr in the same directory:

require "./example"

conn = Example.connect
Example.consume(conn.channel, "hello") { |message| puts " [x] Received '#{message}'" }
puts " [*] Waiting for messages. To exit press CTRL+C"

Process.on_terminate do
  conn.close
  exit
end
sleep

The shard.yml with the publisher and worker targets lives in the example repo. Run shards build, then bin/worker in one terminal and bin/publisher in another.

The full, runnable example lives in crystal/amqp-client in cloudamqp/examples, and CI tests it against LavinMQ.

Good to know

The worker acks manually. subscribe acks on delivery by default, so the example passes no_ack: false and acks after the block returns. If the block raises, the client closes the channel, LavinMQ requeues the unacked messages, and that consumer stops.
Set a prefetch limit. The client sets no prefetch, so LavinMQ sends as many messages as it can. The example calls ch.prefetch(10) before subscribing.
The client doesn’t reconnect. Register conn.on_disconnect to notice a lost connection, then reconnect or let the process exit and restart.
Use amqps:// for TLS. The client verifies the server certificate and hostname.

The amqp-client.cr README covers publisher confirms, stream queues and the rest of the channel API.

Ready to take the next steps?

Managed LavinMQ instance via CloudAMQP

LavinMQ has been built with performance and ease of use in mind - we've benchmarked a throughput of about 1,000,000 messages/sec. You can try LavinMQ without any installation hassle by creating a free instance on CloudAMQP. Signing up is a breeze.

Get started with CloudAMQP ->

Help and feedback

We welcome your feedback and are eager to address any questions you may have about this piece or using LavinMQ. Join our Slack channel to connect with us directly. You can also find LavinMQ on GitHub.

Can’t find what you’re looking for? Let us know

Was this helpful?