Download the PHP package tsterker/hopper without Composer

tsterker/hopper

Opinionated RabbitMQ client (Hopper) intended to facilitate the implementation of "at least once" semantics bundled with a RabitMQ Management Interface client (Warren).

:construction: This library (and this README) is the rushed and incoherent result out of "brainstorming" and "coding by wishful thinking" session with very specific use-cases in mind and is bound to change or even be completely reimagined in a later iteration.

Some guiding principles/thoughts were:

• Pragmatic, purpose-focused; keeping interface minimal/simple
• Facilitate "at least once" semantics, also for chains of consumers, where incoming messages should only be ACKed after publish of outgoing message was confirmed
• Make it easy to send regular heartbeats (e.g. by leveraging declare(ticks=N))
• Don't try to generalize the solution yet (e.g. currently only support for fanout exchanges), but get a feeling for a general architecture/core concepts.
• Don't leak underlying AMQP library (too much), but internally fully commit to it.
• Consider extending library to also support AMQP libraries (bunny?) or switch to it all together. Main requirement is that publisher confirms are supported.

Some opinions, which might be loosened as the package matures:

• Enforce prefetch count
• Use publisher confirms
• Only declare durable queues/exchanges
• Only send persistent messages (delivery_mode=persistent)
• Only use lazy queues (x-queue-mode=lazy)
• Only raw json messages
• Only declare FANOUT exchanges
• ...

1. Table of Contents

• 1. Table of Contents
• 2. Getting Started
• 3. Core Concepts
  • 3.1. Subscriber
  • 3.3. Piper
• 4. Optimization / Parameter Tuning
  • 4.1. Optimize Message Publishing
  • 4.2. Publisher Confirms
  • 4.2.1. ACK latency for persisted messages
• 5. Troubleshooting
  • 5.1. Framing error
• 6. TODO

2. Getting Started

:warning: The library currently depends on an AMQPStreamConnection. The goal is to move this dependency inside Hopper.

Create an AMQP connection.

Start using hopper

3. Core Concepts

Below a rought outline of the core concepts that govern this library and how they are intended to be leveraged for different use cases.

3.1. Subscriber

 * "LOW LEVEL" (e.g. not Handler classes, plain callbacks. But convenience of idle handling)
 * - Register one or more message handler callback(s) to queue(s)
 * - Register idle handler
 * - Consume

• Subscribe to multiple queues
• Register idle handler that should be called when no messages are received for configured idle timeout

3.3. Piper

The Piper builds on top of Subscribers and supports "at least once" semantics in message pipelines, by ensuring that each incoming message is only ACKed after an outgoing message was confirmed.

• Connect input and output queue with a Tranformer
• Transformer receives message from input queue and returns a message that should be published to output queue
• Piper will take care of ACKing incoming messages only once outgoing messages were successfully published
• Supports message buffering & batch publish
• TODO: Don't mix concepts of onFlush and idle callbacks?

4. Optimization / Parameter Tuning

This secion is a brain-dump of things to consider for potential optimizations.

4.1. Optimize Message Publishing

see https://github.com/php-amqplib/php-amqplib#optimized-message-publishing

4.2. Publisher Confirms

4.2.1. ACK latency for persisted messages

see https://www.rabbitmq.com/confirms.html#publisher-confirms-latency

basic.ack for a persistent message routed to a durable queue will be sent after persisting the message to disk. The RabbitMQ message store persists messages to disk in batches after an interval (a few hundred milliseconds) to minimise the number of fsync(2) calls, or when a queue is idle.

This means that under a constant load, latency for basic.ack can reach a few hundred milliseconds. To improve throughput, applications are strongly advised to process acknowledgements asynchronously (as a stream) or publish batches of messages and wait for outstanding confirms. The exact API for this varies between client libraries.

5. Troubleshooting

5.1. Framing error

• Seems I only ran into this when attempting to wait for pending publisher acknowledgements (wait_for_pending_acks).
• Seems to happen when the read_write_timeout is too low (e.g. 0-10) when creating a connection
• Setting read_write_timeout to null (or ommiting it) seems to prevent this error, but might come with implications if rabbitmq would stop responding?
• Currently I settled with setting it to a reasonably high 120. If I would run into such a timeout, I'd be happy for the process to fail.

6. TODO

• Allow more configuration for what types of queues/exchanges to declare
• ...