Gorabbit

<p align="center"> <img src="assets/gorabbit-logo-sm.jpg" alt="logo" width="256" style="border-radius:50%"/> </p>

Gorabbit is a wrapper that provides high level and robust RabbitMQ operations through a client or a manager.

This wrapper depends on the official Go RabbitMQ plugin.

• Gorabbit
  • Installation
    • Go module
    • Environment variables
  • Always-on mechanism
  • Client
    • Client initialization
    • Client options
    • Client with default options
    • Client with options from environment variables
    • Client with custom options
      • Client options using the builder
      • Client options using struct initialization
    • Client disconnection
    • Publishing
    • Consuming
    • Ready and Health checks
  • Manager
    • Manager initialization
    • Manager options
    • Manager with default options
    • Manager with options from environment variables
    • Manager with custom options
      • Manager options using the builder
      • Manager options using struct initialization
    • Manager disconnection
    • Manager operations
      • Exchange creation
      • Queue creation
      • Binding creation
      • Queue messages count
      • Push message
      • Pop message
      • Delete binding
      • Purge queue
      • Delete queue
      • Delete exchange
      • Setup from schema definition file
  • Launch Local RabbitMQ Server
  • License

Installation

Go module

go get github.com/KardinalAI/gorabbit

Environment variables

The client's and manager's Mode can also be set via an environment variable that will override the manually entered value.

GORABBIT_MODE: debug    # possible values: release or debug

The client and manager can also be completely disabled via the following environment variable:

GORABBIT_DISABLED: true     # possible values: true, false, 1, or 0 

Always-on mechanism

Here is a visual representation of the always-on mechanism of a connection and channel when the KeepAlive flag is set to true.

Client

The gorabbit client offers 2 main functionalities:

• Publishing
• Consuming

Additionally, the client also provides a ready check and a health check.

Client initialization

A client can be initialized via the constructor NewClient. This constructor takes ClientOptions as an optional parameter.

Client options

| Property | Description | Default Value | |---------------------|---------------------------------------------------------------|---------------| | Host | The hostname of the RabbitMQ server | 127.0.0.1 | | Port | The port of the RabbitMQ server | 5672 | | Username | The plain authentication username | guest | | Password | The plain authentication password | guest | | Vhost | The specific vhost to use when connection to CloudAMQP | | | UseTLS | The flag that activates the use of TLS (amqps) | false | | ConnectionName | The desired connection name | Gorabbit | | KeepAlive | The flag that activates retry and re-connect mechanisms | true | | RetryDelay | The delay between each retry and re-connection | 3 seconds | | MaxRetry | The max number of message retry if it failed to process | 5 | | PublishingCacheTTL | The time to live for a failed publish when set in cache | 60 seconds | | PublishingCacheSize | The max number of failed publish to add into cache | 128 | | Mode | The mode defines whether logs are shown or not | Release | | Marshaller | The content type used for messages and how they're marshalled | JSON |

Client with default options

Passing nil options will trigger the client to use default values (host, port, credentials, etc...) via DefaultClientOptions().

client := gorabbit.NewClient(nil)

You can also explicitly pass DefaultClientOptions() for a cleaner initialization.

client := gorabbit.NewClient(gorabbit.DefaultClientOptions())

Finally, passing a NewClientOptions() method also initializes default values if not overwritten.

client := gorabbit.NewClient(gorabbit.NewClientOptions())

Client with options from environment variables

You can instantiate a client from environment variables, without the need of manually specifying options in the code.

client := gorabbit.NewClientFromEnv()

Here are the following supported environment variables:

• RABBITMQ_HOST: Defines the host,
• RABBITMQ_PORT: Defines the port,
• RABBITMQ_USERNAME: Defines the username,
• RABBITMQ_PASSWORD: Defines the password,
• RABBITMQ_VHOST: Defines the vhost,
• RABBITMQ_USE_TLS: Defines whether to use TLS or no.
• RABBITMQ_CONNECTION_NAME: Defines the desired connection name.

Note that environment variables are all optional, so missing keys will be replaced by their corresponding default.

Client with custom options

We can input custom values for a specific property, either via the built-in builder or via direct struct initialization.

Client options using the builder

NewClientOptions() and DefaultClientOptions() both return an instance of *ClientOptions that can act as a builder.

options := gorabbit.NewClientOptions().
    SetMode(gorabbit.Debug).
    SetCredentials("root", "password").
    SetRetryDelay(5 * time.Second)

client := gorabbit.NewClient(options)

:information_source: There is a setter method for each property.

Client options using struct initialization

ClientOptions is an exported type, so it can be used directly.

options := gorabbit.ClientOptions {
    Host:     "localhost",
    Port:     5673,
    Username: "root",
    Password: "password",
    ...
}

client := gorabbit.NewClient(&options)

:warning: Direct initialization via the struct does not use default values on missing properties, so be sure to fill in every property available.

Client disconnection

When a client is initialized, to prevent a leak, always disconnect it when no longer needed.

client := gorabbit.NewClient(gorabbit.DefaultClientOptions())
defer client.Disconnect()

Publishing

To send a message, the client offers two simple methods: Publish and PublishWithOptions. The required arguments for publishing are:

• Exchange (which exchange the message should be sent to)
• Routing Key
• Payload (interface{}, the object will be marshalled internally)

Example of sending a simple string

err := client.Publish("events_exchange", "event.foo.bar.created", "foo string")

Example of sending an object

type foo struct {
    Action string
}

err := client.Publish("events_exchange", "event.foo.bar.created", foo{Action: "bar"})

Optionally, you can set the message's Priority, DeliveryMode and Expiration via the PublishWithOptions method.

options := gorabbit.SendOptions().
    SetPriority(gorabbit.PriorityMedium).
    SetDeliveryMode(gorabbit.Persistent).
    SetTTL(5*time.Second)

err := client.PublishWithOptions("events_exchange", "event.foo.bar.created", "foo string", options)

:information_source: If the KeepAlive flag is set to true when initializing the client, failed publishing will be cached once and re-published as soon as the channel is back up.

Consuming

To consume messages, gorabbit offers a very simple asynchronous consumer method Consume that takes a MessageConsumer as argument. Error handling, acknowledgement, negative acknowledgement and rejection are all done internally by the consumer.

err := client.RegisterConsumer(gorabbit.MessageConsumer{
    Queue:             "events_queue",
    Name:              "toto_consumer",
    PrefetchSize:      0,
    PrefetchCount:     10,
    AutoAck:           false,
    ConcurrentProcess: false,
    Handlers: gorabbit.MQTTMessageHandlers{
        "event.foo.bar.created": func (payload []byte) error {
            fmt.Println(string(payload))

            return nil
        },
    },
})

• Queue: The queue to consume messages from
• Name: Unique identifier for the consumer
• PrefetchSize: The maximum size of messages that can be processed at the same time
• PrefetchCount: The maximum number of messages that can be processed at the same time
• AutoAck: Automatic acknowledgement of messages upon reception
• ConcurrentProcess: Asynchronous handling of deliverie