Ears

Ears is a small, simple library for writing RabbitMQ consumers and publishers.

Installation

Add this line to your application's Gemfile:

gem'ears'

And then execute:

$ bundle install

Or install it yourself as:

$ gem install ears

Usage

Publishing Messages

Ears provides a thread-safe publisher for sending messages to RabbitMQ exchanges with automatic retry and connection recovery capabilities.

Basic Publisher Usage

To publish messages, create an Ears::Publisher instance and call publish:

require'ears'# Configure Ears (same configuration is shared by consumers and publishers)Ears.configuredo |config|
config.rabbitmq_url='amqp://user:password@myrmq:5672'config.connection_name='My Publisher'end# Create a publisher for a topic exchangepublisher=Ears::Publisher.new('my_exchange',:topic,durable: true)# Publish a messagedata={user_id: 123,action: 'login',timestamp: Time.now.iso8601}publisher.publish(data,routing_key: 'user.login')

Exchange Types and Options

Publishers support all RabbitMQ exchange types:

# Topic exchange (default)topic_publisher=Ears::Publisher.new('events',:topic)# Direct exchangedirect_publisher=Ears::Publisher.new('commands',:direct)# Fanout exchangefanout_publisher=Ears::Publisher.new('broadcasts',:fanout)# Headers exchangeheaders_publisher=Ears::Publisher.new('complex_routing',:headers)# Custom exchange optionspublisher=Ears::Publisher.new('my_exchange',:topic,durable: true,auto_delete: false,arguments: {'x-message-ttl'=>60_000,},)

Message Options

The publish method accepts various message options:

publisher.publish({message: 'Hello World'},routing_key: 'greeting.hello',persistent: true,# Persist message to disk (default: true)headers: {version: '1.0',},# Custom headerstimestamp: Time.now.to_i,# Message timestamp (default: current time)message_id: SecureRandom.uuid,# Unique message identifiercorrelation_id: 'abc-123',# Correlation ID for request/response patternsreply_to: 'response_queue',# Queue for responsesexpiration: '60000',# Message TTL in millisecondspriority: 5,# Message priority (0-9)type: 'user_event',# Message typeapp_id: 'my_application',# Application identifieruser_id: 'system',# User identifier (verified by RabbitMQ))

Thread-Safe Publishing

Publishers use a connection pool for thread-safe operation, making them suitable for concurrent use:

# Single publisher can be safely used across multiple threadspublisher=Ears::Publisher.new('events',:topic)# Example with multiple threadsthreads=10.times.mapdo |i|
Thread.newdo100.timesdo |j|
publisher.publish({thread: i,message: j},routing_key: "thread.#{i}")endendendthreads.each(&:join)

Publisher Configuration

Publisher behavior can be fine-tuned through configuration options:

Ears.configuredo |config|
# Connection settingsconfig.rabbitmq_url='amqp://user:password@myrmq:5672'config.connection_name='My Application'# Publisher-specific settingsconfig.publisher_pool_size=32# Channel pool size (default: 32)config.publisher_pool_timeout=2# Pool checkout timeout in seconds (default: 2)# Connection retry settingsconfig.publisher_connection_attempts=30# Connection retry attempts (default: 30)config.publisher_connection_base_delay=1# Base delay between connection attempts (default: 1s)config.publisher_connection_backoff_factor=1.5# Connection backoff multiplier (default: 1.5)# Publish retry settingsconfig.publisher_max_retries=3# Max publish retry attempts (default: 3)config.publisher_retry_base_delay=0.1# Base delay between publish retries (default: 0.1s)config.publisher_retry_backoff_factor=2# Publish retry backoff multiplier (default: 2)end

Fault Tolerance and Recovery

Publishers automatically handle connection failures and provide several recovery mechanisms:

Automatic Retry with Exponential Backoff

# Publishers automatically retry failed operationspublisher=Ears::Publisher.new('events',:topic)# This will automatically retry with exponential backoff if the connection failspublisher.publish({event: 'user_signup'},routing_key: 'user.signup')

Manual Recovery

If you need to manually reset the connection pool (e.g., after detecting connection issues):

publisher=Ears::Publisher.new('events',:topic)# Reset the channel pool to force new connectionspublisher.reset!# Subsequent publishes will use fresh channelspublisher.publish({event: 'recovery_test'},routing_key: 'system.recovery')

Error Handling

Publishers raise specific exceptions that you can handle:

require'ears'publisher=Ears::Publisher.new('events',:topic)beginpublisher.publish({data: 'test'},routing_key: 'test.message')rescueEars::PublisherRetryHandler::PublishError=>e# Handle publish failures (after all retries exhausted)logger.error"Failed to publish message: #{e.message}"# Consider queuing message for later retry or alertingrescue=>e# Handle other unexpected errorslogger.error"Unexpected error: #{e.message}"end

Publisher Confirms

Basic Usage

For guaranteed message delivery, use publish_with_confirmation which waits for broker acknowledgment:

publisher=Ears::Publisher.new('events',:topic)# Publish with confirmation - blocks until broker acknowledgespublisher.publish_with_confirmation({user_id: 123,action: 'payment_processed'},routing_key: 'payment.processed',)

Configuration

Publisher confirms use a separate channel pool with configurable settings:

Ears.configuredo |config|
# Confirms-specific channel pool size (default: 32)config.publisher_confirms_pool_size=32# Timeout for waiting for confirms in seconds (default: 5.0)config.publisher_confirms_timeout=5.0# Cleanup timeout after confirmation failure (default: 1.0)config.publisher_confirms_cleanup_timeout=1.0end

Error Handling

Publisher confirms raise specific exceptions that are NOT automatically retried:

beginpublisher.publish_with_confirmation(data,routing_key: 'important.event')rescueEars::PublishConfirmationTimeout=>e# Message may or may not have reached brokerlogger.error"Confirmation timed out: #{e.message}"rescueEars::PublishNacked=>e# Broker explicitly rejected the messagelogger.error"Message was nacked: #{e.message}"end

Note: Unlike regular publishing, confirmation errors are not retried to avoid message duplication.

Basic consumer usage

First, you should configure Ears.

require'ears'Ears.configuredo |config|
config.rabbitmq_url='amqp://user:password@myrmq:5672'config.connection_name='My Consumer'config.recover_from_connection_close=false# optional configuration, defaults to true if not setconfig.recovery_attempts=3# optional configuration, defaults to 10, Bunny::Session would have been nil# Publisher configuration (optional)config.publisher_pool_size=32# Thread pool size for publishers (default: 32)config.publisher_pool_timeout=2# Timeout for pool checkout in seconds (default: 2)config.publisher_connection_attempts=30# Connection retry attempts (default: 30)config.publisher_connection_base_delay=1# Base delay between connection attempts in seconds (default: 1)config.publisher_connection_backoff_factor=1.5# Connection retry backoff multiplier (default: 1.5)config.publisher_max_retries=3# Max publish retry attempts (default: 3)config.publisher_retry_base_delay=0.1# Base delay between publish retries in seconds (default: 0.1)config.publisher_retry_backoff_factor=2# Publish retry backoff multiplier (default: 2)end

Note: connection_name is a mandatory setting!

Next, you can define your exchanges, queues, and consumers in 2 ways:

1. consumer specific configuration method (recommended)

1. Pass your consumer classes to Ears.setup:

Ears.setupdoEars.setup_consumers(Consumer1,Consumer2, ...)end

2. Implement your consumers by subclassing Ears::Consumer. and call the configure method.

classConsumer1 < Ears::Consumerconfigure(queue: 'queue_name',exchange: 'exchange',routing_keys: %w[routing_key1routing_key2],retry_queue: true,# optional configuration, defaults to false, Adds a retry queueerror_queue: true,# optional configuration, defaults to false, Adds an error queue)defwork(delivery_info,metadata,payload)message=JSON.parse(payload)do_stuff(message)ackendend

2. Generic configuration method

Ears.setupdo# define a durable topic exchangemy_exchange=exchange('my_exchange',:topic,durable: true)# define a queuemy_queue=queue('my_queue',durable: true)# bind your queue to the exchangemy_queue.bind(my_exchange,routing_key: 'my.routing.key')# define a consumer that handles messages for that queueconsumer(my_queue,MyConsumer)end

Finally, you need to implement MyConsumer by subclassing Ears::Consumer. and call the configure method.

classMyConsumer < Ears::Consumerdefwork(delivery_info,metadata,payload)message=JSON.parse(payload)do_stuff(message)ackendend

Run your consumers

Note: Be prepared that unhandled errors will be reraised. So, take care of cleanup work.

beginEars.run!ensure# all your cleanup work goes here...end

At the end of the #work method, you must always return ack, reject, or requeue to signal what should be done with the message.

Middlewares

Ears supports middlewares that you can use for recurring tasks that you don't always want to reimplement. It comes with some built-in middlewares:

• Ears::JSON for automatically parsing JSON payloads
• Ears::Appsignal for automatically wrapping #work in an Appsignal transaction

You can use a middleware by just calling use with the middleware you want to register in your consumer.

require'ears/middlewares/json'classMyConsumer < Ears::Consumer# register the JSON middleware and don't symbolize keys (this can be omitted, the default is true)# and nack the message on parsing error. This defaults to Proc.new { :reject }.useEars::Middlewares::JSON,on_error: Proc.new{:nack},symbolize_keys: falsedefwork(delivery_info,metadata,payload)returnackunlesspayload['data'].nil?# this now just worksendend

If you want to implement your own middleware, just subclass Ears::Middleware and implement #call (and if you need it #initialize).

classMyMiddleware < Ears::Middlewaredefinitialize(opts={})@my_option=opts.fetch(:my_option,nil)enddefcall(delivery_info,metadata,payload,app)do_stuff# always call the next middleware in the chain or your consumer will never be calledapp.call(delivery_info,metadata,payload)endend

Multiple threads

If you need to handle a lot of messages, you might want to have multiple instances of the same consumer all working on a dedicated thread. This is supported out of the box. You just have to define how many consumers you want when calling consumer in Ears.setup.

Ears.setupdomy_exchange=exchange('my_exchange',:topic,durable: true)my_queue=queue('my_queue',durable: true)my_queue.bind(my_exchange,routing_key: 'my.routing.key')# this will instantiate MyConsumer 10 times and run every instance on a dedicated threadconsumer(my_queue,MyConsumer,10)end

It may also be interesting for you to increase the prefetch amount. The default prefetch amount is 1, but if you have a lot of very small, fast to process messages, a higher prefetch is a good idea. Just set it when defining your consumer.

Ears.setupdomy_exchange=exchange('my_exchange',:topic,durable: true)my_queue=queue('my_queue',durable: true)my_queue.bind(my_exchange,routing_key: 'my.routing.key')# this will instantiate one consumer but with a prefetch value of 10consumer(my_queue,MyConsumer,1,prefetch: 10)end

Setting arbitrary exchange/queue parameters

If you need some custom arguments on your exchange or queue, you can just pass these to queue or exchange inside Ears.setup. These are then just passed on to Bunny::Queue and Bunny::Exchange.

Ears.setupdomy_queue=queue('my_queue',durable: true,arguments: {'x-message-ttl'=>10_000})end

Implementing a retrying queue

Sometimes you want to automatically retry processing a message, in case it just failed due to temporary problems. In that case, you can set the retry_queue and retry_delay parameters when creating the queue OR pass it to the configure method in your consumer.

classMyConsumer < Ears::Consumerconfigure(queue: 'queue_name',exchange: 'exchange',routing_keys: %w[routing_key1routing_key2],retry_queue: true,)defwork(delivery_info,metadata,payload)message=JSON.parse(payload)do_stuff(message)ackendend

my_queue=queue('my_queue',durable: true,retry_queue: true,retry_delay: 5000)

This will automatically create a queue named my_queue.retry and use the arguments x-dead-letter-exchange and x-dead-letter-routing-key to route rejected messages to it. When routed to the retry queue, messages will wait there for the number of milliseconds specified in retry_delay, after which they will be redelivered to the original queue. Note that this will not automatically catch unhandled errors. You still have to catch any errors yourself and reject your message manually for the retry mechanism to work.

This will happen indefinitely, so if you want to bail out of this cycle at some point, it is best to use the error_queue option to create an error queue and then use the MaxRetries middleware to route messages to this error queue after a certain amount of retries.

Implementing an error queue

You can set the error_queue parameter to automatically create an error queue, or add it to the configure method in your consumer.

classMyConsumer < Ears::Consumerconfigure(queue: 'queue_name',exchange: 'exchange',routing_keys: %w[routing_key1routing_key2],error_queue: true,)defwork(delivery_info,metadata,payload)message=JSON.parse(payload)do_stuff(message)ackendend

my_queue=queue('my_queue',durable: true,retry_queue: true,retry_delay: 5000,error_queue: true,)

This will automatically create a queue named my_queue.error. It does not have any special properties, the helper's main purpose is to enforce naming conventions. In your consumer, you should then use the MaxRetries middleware to route messages to the error queue after a certain amount of retries.

classMyConsumer < Ears::ConsumeruseEars::Middlewares::MaxRetries,retries: 3,error_queue: 'my_queue.error'defwork(delivery_info,metadata,payload)# ...endend

This will automatically route messages to my_queue.error after they have been re-tried three times. This prevents you from infinitely retrying a faulty message.

Stopping any running consumers

When you are running Ears in a non-blocking way (e.g. in a Thread), it might be cumbersome to remove the running consumers without restarting the whole app.

For this use case, there is a stop! method:

Ears.stop!

It will close and reset the current Bunny connection, leading to all consumers being shut down. Also, it will reset the channel.

Complete Example: Consumer and Publisher

Here's a complete example showing both consumer and publisher usage:

require'ears'# Shared configurationEars.configuredo |config|
config.rabbitmq_url='amqp://guest:guest@localhost:5672'config.connection_name='Order Processing Service'config.publisher_pool_size=16end# Consumer that processes orders and publishes eventsclassOrderProcessor < Ears::Consumerconfigure(queue: 'orders',exchange: 'ecommerce',routing_keys: %w[order.createdorder.updated],retry_queue: true,error_queue: true,)definitializesuper@event_publisher=Ears::Publisher.new('events',:topic,durable: true)enddefwork(delivery_info,metadata,payload)order=JSON.parse(payload)# Process the orderprocess_order(order)# Publish success event@event_publisher.publish({order_id: order['id'],status: 'processed',processed_at: Time.now.iso8601,},routing_key: 'order.processed',)ackrescue=>error# Publish error event@event_publisher.publish({order_id: order&.dig('id'),error: error.message,failed_at: Time.now.iso8601,},routing_key: 'order.failed',)reject# Send to error queueendprivatedefprocess_order(order)# Order processing logic heresleep(0.1)# Simulate processing timeendend# Setup and runEars.setup{Ears.setup_consumers(OrderProcessor)}beginEars.run!ensure# Cleanup code hereend

Testing

Ears provides testing helpers to easily test your message publishing without connecting to RabbitMQ.

Basic Setup

Include the test helper in your RSpec tests and mock the exchanges you want to test:

require'ears/testing'RSpec.describeMyServicedoincludeEars::Testing::TestHelperbeforedo# Mock exchanges that your code will publish tomock_ears('events','notifications')endafterdo# Clean up mocks and captured messagesears_reset!endend

Capturing and Inspecting Messages

Use the helper methods to inspect published messages:

it'publishes user creation event'doservice=UserService.newservice.create_user(name: 'John',email: 'john@example.com')# Get all messages published to 'events' exchangemessages=published_messages('events')expect(messages.size).toeq(1)# Inspect the messagemessage=messages.firstexpect(message.routing_key).toeq('user.created')expect(message.data).toinclude(name: 'John')expect(message.options[:headers]).toinclude(version: '1.0')end

Available Helper Methods

• published_messages(exchange_name = nil) - Get messages for a specific exchange or all messages
• last_published_message(exchange_name = nil) - Get the most recent message
• clear_published_messages - Clear captured messages during a test

Message Properties

Each captured message has the following properties:

• exchange_name - Name of the exchange
• routing_key - Message routing key
• data - The message payload
• options - Publishing options (headers, persistent, etc.)
• timestamp - When the message was captured
• thread_id - Thread that published the message

Custom RSpec Matcher: have_been_published

To make tests more expressive, Ears provides a custom RSpec matcher that allows you to easily assert that a specific message was published to a mocked exchange.

Usage

Include the matcher by requiring ears/testing in your RSpec tests and including the helper module:

require'ears/testing/matchers'RSpec.describeMyPublisherdoincludeEars::Testing::Matchersbefore{mock_ears('events')}after{ears_reset!}it'publishes a user.created message'dopublisher=Ears::Publisher.new('events',:topic)publisher.publish({user_id: 1},routing_key: 'user.created')expect(exchange_name: 'events',routing_key: 'user.created',data: {user_id: 1,},).tohave_been_publishedend# also works with negative assertionsit'does not publish a user.deleted message'dopublisher=Ears::Publisher.new('events',:topic)publisher.publish({user_id: 1},routing_key: 'user.created')expect(exchange_name: 'events',routing_key: 'user.deleted',data: {user_id: 1,},).not_tohave_been_publishedendend

Supported Keys

You can match on any or all of the following attributes:

Key	Description	Example
:exchange_name	The exchange where the message was published	'events'
:routing_key	The routing key used for the message	'user.created'
:data	The message payload (exact match)	{ user_id: 1 }
:options	Message options such as headers or persistence	{ persistent: true, headers: { version: '1.0' } }

If a key is omitted, it will not be checked — allowing partial matches (for example, matching only on exchange_name and routing_key).

Example: Matching with Options

Note: When matching :options, you only need to specify the options you want to verify — the matcher will ignore any additional options present in the published message.

expect(exchange_name: 'events',routing_key: 'user.created',data: {id: 42,name: 'Alice',},options: {persistent: true,},).tohave_been_published

Error Handling

By default, publishing to unmocked exchanges raises an error:

it'raises error for unmocked exchanges'dopublisher=Ears::Publisher.new('unmocked_exchange')expect{publisher.publish({data: 'test'},routing_key: 'test')}.toraise_error(Ears::Testing::UnmockedExchangeError)end

Complete Example

require'ears/testing'RSpec.describeOrderProcessordoincludeEars::Testing::TestHelperbefore{mock_ears('events','notifications')}after{ears_reset!}it'publishes events when processing order'doprocessor=OrderProcessor.neworder={id: 123,items: ['item1'],total: 99.99}processor.process(order)# Check event was publishedevents=published_messages('events')expect(events.size).toeq(1)expect(events.first.routing_key).toeq('order.processed')expect(events.first.data[:order_id]).toeq(123)# Check notification was sentnotifications=published_messages('notifications')expect(notifications.size).toeq(1)expect(notifications.first.routing_key).toeq('email.order_confirmation')endend

Documentation

If you need more in-depth information, look at our API documentation.

Contributing

Bug reports and pull requests are welcome on GitHub at https://github.com/ivx/ears. This project is intended to be a safe, welcoming space for collaboration, and contributors are expected to adhere to the code of conduct.

License

The gem is available as open-source under the terms of the MIT License.

Code of Conduct

Everyone interacting in the Ears project's codebases, issue trackers, chat rooms, and mailing lists is expected to follow the code of conduct.