Gem for happn

Happn connects a RabbitMQ exchange and listens for CREPE events (possibly generated using flu-rails) sequentially. Happn helps developers to create "Projectors" that define how to match and consume events.

This gem connects a single RabbitMQ queue and bind it automatically to its exchange. These bindings are defined by developers through "matchers" when loading projectors.

Requirements

Ruby >= 3.2
Tested with RabbitMQ 3.5.8
happn-ruby works with or without Rails (tested with Rails 4 and 5).

Installation

Add the gem to your project's Gemfile:

```ruby
gem "happn"
```

Then, configure Happn. If you use Rails, you can create an initializer into your Rails app (config/initializers/happn.rb). This code can be called anywhere before starting Happn.init.

Happn.configure do |config|
  config.logger                     = Rails.logger
  config.rabbitmq_host              = ENV["RABBITMQ_HOST"]
  config.rabbitmq_port              = ENV["RABBITMQ_PORT"]
  config.rabbitmq_management_port   = ENV["RABBITMQ_MANAGEMENT_PORT"]
  config.rabbitmq_user              = ENV["RABBITMQ_USER"]
  config.rabbitmq_password          = ENV["RABBITMQ_PASSWORD"]
  config.rabbitmq_queue_name        = ENV["CONSUMER_QUEUE_NAME"]
  config.rabbitmq_exchange_name     = ENV["RABBITMQ_EXCHANGE_NAME"]
  config.rabbitmq_exchange_durable  = ENV["RABBITMQ_EXCHANGE_DURABLE"] == "true"
  config.projector_classes          = [LoggerProjector]
  config.bunny_options              = {}
end

Each configuration is detailed below.

About queues

Happn consumes a single queue through the RabbitMQ's Topic Exchange Model.
If the queue does not exist when Happn starts, it is created automatically.
When connecting a queue, please be careful that each connection parameter must match the existing queue's parameters. For instance, the value of x-queue-mode must match to avoid a PRECONDITION FAILED error.
The bindings between the queue and the exchange Happn consumes are reset when starting Happn. Based on all the projectors that have been registered (option projector_classes), Happn detects which events must be consumed, binds its queue to the exchange depending on these event matchers, and unbinds the routing keys that no longer match any projector. Bindings the same queue may have to other exchanges are left untouched.
Happn reads those existing bindings through the RabbitMQ management API, on the vhost its broker connection uses. To consume a vhost other than the default one, declare it in bunny_options (vhost or virtual_host, as Bunny accepts both): Happn follows it on both the AMQP connection and the management API.

About projectors

A projector:

defines one or multiple matchers to detect which events must be consumed. Matchers can be declared using 4 event properties
- emitter: e.g. Facebook or MyInternalApi (:all or no value mean "all emitters")
- kind: e.g. entity_change or kind (:all or no value mean "all kinds")
- name: e.g. create country or request to destroy bunnies (:all or no value mean "all names")
- status: e.g. :new or :replayed (:all or no value mean "all statuses")
defines how to consume these events.

When a projector raises an Error, Happn stops.

Usage

Start Up

Starting Happn consumes events sequentially. For instance, it can be started from a Rake tasks:

```ruby
namespace :events do
desc "Listen all events and consume them."
task consume: :environment do
  Happn.init
  Happn.start
end
end
```

Shutting Down

Happn.start blocks the calling thread for as long as the consumption lasts. Happn.stop releases it: it cancels the consumer, lets the messages already being handled run to completion, and closes the broker connection. It is meant to be called from another thread than the one blocked in Happn.start.

Happn installs no signal handler of its own, because only the application knows what a graceful shutdown means for it. Wiring one is up to you:

```ruby
namespace :events do
desc "Listen all events and consume them."
task consume: :environment do
  Happn.init
  Signal.trap("TERM") { Thread.new { Happn.stop } }
  Signal.trap("INT")  { Thread.new { Happn.stop } }
  Happn.start
end
end
```

Signal handlers run in a context where very little is allowed, hence the Thread.new.

Without it, an orchestrator stopping the process interrupts the handler in flight. Nothing is lost, the message was never acknowledged, so the broker redelivers it, but it is consumed twice, once partially.

Define a Projector

A projector is a class that defines how to consume one or multiple types of events. This class must:

extend Happn::Projector.
use its on method to declare which events to match and how to consume them. This must be done in a define_handlers method.

class LoggerProjector < Happn::Projector
  def define_handlers
    on emitter: "MyApplication", name: "create country" do |event|
      Rails.logger("A country has been created and generated an event with id #{event.id}")
    end

    on kind: "request", status: :new do |event|
      Rails.logger("This is a new request to the controller: #{event.data[:controller_name]}")
    end
  end
end

Registering all projectors automatically

If all subclass of Happn::Projector should be registered seamlessly, the configuration can declare the following code (using Rails):

Happn.configure do |config|
  Rails.application.eager_load!
  config.projector_classes = Happn::Projector.descendants
end

When Zeitwerk is defined and when Rails.application.eager_load! returns false, you should call Zeitwerk::Loader.eager_load_all.

Running the tests

The test suite is written with RSpec and needs no running RabbitMQ: the broker is stubbed.

```bash
bundle install
bundle exec rake     # or: bundle exec rspec
```

Running the integration tests

The test suite above needs no running RabbitMQ. A second suite, under spec_integration/, exercises Happn against a real broker instead: queue and binding lifecycle, message routing, consumption and acknowledgement, and the reject-and-exit contract when a projector handler raises.

```bash
docker compose up -d --wait   # starts RabbitMQ with the management plugin
bundle exec rake integration  # or: bundle exec rspec spec_integration
docker compose down
```

Overall configuration options

All options have a default value. However, all of them can be changed in your Happn.configure block.

Option	Default Value	Type	Required?	Description	Example
`logger`	`Logger.new(STDOUT)`	Logger	Optional	The logger used by `happn`	`Rails.logger`
`rabbitmq_host`	`"localhost"`	String	Required	RabbitMQ exchange's host.	`"192.168.42.42"`
`rabbitmq_port`	`5672`	Integer	Required	RabbitMQ exchange's port.	`1234`
`rabbitmq_user`	`""`	String	Required	RabbitMQ exchange's username.	`"root"`
`rabbitmq_password`	`""`	String	Required	RabbitMQ exchange's password.	`"pouet"`
`rabbitmq_exchange_name`	`"events"`	String	Required	RabbitMQ exchange's name.	`"myproject"`
`rabbitmq_management_scheme`	`"http"`	String	Required	RabbitMQ exchange's management scheme. This scheme is used when `happn` must access metadata information about queues, messages, etc. This port is used to create/delete bindings between the queue and its exchange.	`"https"`
`rabbitmq_management_port`	`15672`	Integer	Required	RabbitMQ exchange's management port. This port is used when `happn` must access metadata information about queues, messages, etc. This port is used to create/delete bindings between the queue and its exchange.	`15671`
`rabbitmq_queue_name`	`"happn-queue"`	String	Required	The RabbitMQ queue to create, bind and consume. If the queue does not exist, it will be created at startup.	`"my-queue"`
`rabbitmq_exchange_durable`	`true`	Boolean	Optional	Make the RabbitMQ's exchange durable or not. From RabbitMQ's documentation: "Durable exchanges survive broker restart whereas transient exchanges do not (they have to be redeclared when broker comes back online)."	`false`
`rabbitmq_queue_mode`	`nil`	String	Optional	When creating the queue, this option can be passed to set `x-queue-mode`. For instance, a queue can be made "lazy" by passing `"lazy"` as a value. See RabbitMQ's documentation for more details.	`lazy`
`rabbitmq_prefetch_size`	`10`	Integer	Optional	Also known as RabbitMQ's QOS. From the RabbitMQ's documentation: "AMQP specifies the basic.qos method to allow you to limit the number of unacknowledged messages on a channel (or connection) when consuming (aka "prefetch count")."	`1000`
`projector_classes`	`[]`	Array of constants	Required	All Projector classes to register. This value can be generated by reading all descendant classes from `Happn::Projector`.	`[MyProjector]`
`bunny_options`	`{}`	Hash of symbols	Optional	Additional options to add when connecting the RabbitMQ broker. This overrides the existing options with the same name. A `vhost` (or `virtual_host`) declared here is also the vhost `Happn` queries through the management API.	`{ verify_peer: true }`
`management_options`	`{}`	Hash of symbols	Optional	Additional options to add when accessing the RabbitMQ Managmement. This overrides the existing options with the same name. The options are defined at https://github.com/ruby-amqp/rabbitmq_http_api_client	`{ verify: false }`
`on_error`	`nil`	`block` with an argument `exception`	false	When the consumption of an event raises an Error, the consumption exits. However, this block can be called before exiting the consumption execution.	`lambda {	exception	Raven.capture_exception(exception) }` (see Sentry's documentation

How to release a new version

Releases are published to RubyGems by GitHub Actions (.github/workflows/release.yml), through Trusted Publishing: no API key is stored in this repository, the workflow exchanges a short-lived GitHub OIDC token for a scoped RubyGems credential.

Update Happn::VERSION in lib/happn/version.rb and the CHANGELOG.md
Commit and push these changes to main
Tag the commit and push the tag:

  $ git tag -a v1.1.8 -m "Version 1.1.8"
  $ git push origin v1.1.8

The workflow then checks that the tag matches Happn::VERSION, runs the tests, builds the gem and pushes it. It only publishes tags starting with v.