# AMQP consumers

Python and Java reference consumers for the [AMQP guide](../../docs/api/amqp-guide.md).
They share the same suspension, join, version, heartbeat and result-storage rules.
Use the `price()` gate in an application: cached rows alone do not mean prices are usable.
Their stdout is a demonstration JSON event stream; a `status` with `joined:false`
suspends **every** previously printed price until the next completed join.

## Run

Python 3.11 or later:

```sh
python3 -m venv .venv
.venv/bin/pip install --require-hashes -r examples/amqp/python/requirements.txt
.venv/bin/python examples/amqp/python/consumer.py
```

Java 21, Maven 3.9 or later:

```sh
mvn -f examples/amqp/java/pom.xml package
java -jar examples/amqp/java/target/amqp-consumer-1.0.0.jar
```

Both read the same environment. Set secrets through the application's secret manager
or hidden prompts, not command-line arguments or checked-in environment files.

| Variable | Meaning |
|---|---|
| `FEED_AMQP_URL` | `amqps://<broker address>:5671/feed`, without embedded credentials |
| `FEED_AMQP_USER`, `FEED_AMQP_PASSWORD` | Assigned broker account |
| `FEED_AMQP_CA_FILE` | Supplied private authority's public PEM certificate; required |
| `FEED_CLIENT` | Assigned client name |
| `FEED_QUEUE_SET` | Assigned queue set, defaults to `1` |
| `FEED_API_URL` | HTTPS API base URL |
| `FEED_API_KEY` | API key with AMQP access |
| `FEED_API_CA_FILE` | Optional pinned API authority for a private API certificate |
| `FEED_BINDINGS` | Comma-separated supported keys; defaults to `#` |
| `FEED_UNBIND` | Old keys to remove from both price/event exchanges before joining |
| `FEED_EVENTS` | `1` to keep event objects; defaults to `0` |
| `FEED_RESULTS_FILE` | Durable JSON-lines result journal; defaults to `results.jsonl` |

Keep the result journal across restarts. Each consumer needs its own journal;
active and standby processes must not concurrently write the same file. The example
fsyncs a result and its journal's directory before acknowledging, and loads prior
versions on restart. Malformed result messages remain unacknowledged. A malformed or
torn journal line stops startup, and a failed journal write stops the consumer:
preserve the file and reconcile it with durable
results before restarting; never discard it automatically. The journal requires a
Linux filesystem that supports directory fsync. Production
applications can replace this journal with a durable database transaction. The
broker retains results for seven days; longer outages need REST reconciliation.

The consumers deliberately repeat the whole join procedure after a lost channel or
connection. Java library automatic recovery is disabled because recovering the
connection alone cannot establish a current price snapshot. See the official
[Java client guide](https://www.rabbitmq.com/client-libraries/java-api-guide).
Both trust only the configured broker authority and check the broker address.

## Verify

```sh
bash examples/amqp/test.sh python
bash examples/amqp/test.sh java
# or both:
bash examples/amqp/test.sh all
```

The script creates one disposable RabbitMQ 4.4 container with no published ports,
copies only these examples into it, installs pinned client/runtime dependencies,
then runs the same scripts as CI. It removes only its own named container. Allow
about 1 GiB for the image/tools and several minutes per language.

Each language runs nine scenarios: join with durable results/corrections, feed
restart and crash, real broker restart with persistent result recovery, feed
reconnection, resync and heartbeat gap, moving events, delayed moves, standby takeover, and a slow
consumer. A synthetic feed serves gzip joins and event pages while publishing
changes; the harness compares consumer rows/results with it and checks suspension,
monotonic versions, acknowledgements and absence of credentials in output. Python
also has focused regressions for join ordering, stale moves, removal versions,
suspension, heartbeat faults and storage failure before acknowledgement.

The test broker recreates the client-facing production topology and permissions:
durable classic queues, single active consumer, queue bounds, user limits and
pinned TLS. Its temporary CA and passwords are generated for each run. The test
API uses loopback HTTP; deployed clients use HTTPS.

CI runs the broker **inside the disposable job container**, rather than as a GitLab
service: this allows testing actual broker stops/restarts and generated TLS without
a privileged job or a Docker socket. The pinned broker image is shared by both
jobs. No external service or production account is used. Job reports contain only
scenario outcomes; certificate/private-key directories are never uploaded.

The stale-move regression clarifies API-660: compare versions before removing an
event, and remember its removal version until the next join. Otherwise an old
queued move can erase a newer snapshot or restore a removed row. The contract
clarification is included in this change.