Laravel Amqp Laravel Package

Getting Started

This is a drop-in, opt-in release: no migrations, no config rewrites, and existing 3.4.0 publish/consume code keeps working. Treat it as "the AMQP client grew an RPC layer and a microservice toolkit" — start with what you already have, then adopt one layer at a time.

Minimal setup

1. composer require bschmitt/laravel-amqp:^3.4 (PHP 7.3–8.5, Laravel 8–13, Lumen 8+).
2. Publish and review the config: php artisan vendor:publish --tag=amqp-config, then set your connection, exchange, and queue in config/amqp.php / .env.
3. Run a worker as an Artisan command under Supervisor — consume loops are long-lived processes, not request-scoped.

Where to look first

• config/amqp.php — connection, exchange, queue, and the new broadcast_laravel_events switch.
• src/Rpc/RpcService.php + src/Rpc/RpcDispatcher.php — the contract and the call/serve machinery.
• src/Support/ — DeadLetterManager, MonitoringDashboard, RetryPolicy, CorrelationContext, InMemoryMessageStore.
• README / changelog for the full new public surface (the release notes are truncated mid-list, so verify class names against the tag).

First use case (pick one)

If you only publish/consume today: stay on TypedMessage. Define a DTO, call TypedMessage::dispatch([...]) from a controller or job, and consume it with your existing handler. Zero new concepts.

If you have request/reply today: move it to gRPC-lite. Define a request DTO extending RpcRequest (override responseClass() for typed replies), a service extending RpcService with queue() and methods(), and a handler class. Register and serve it in a worker (Rpc::register(UserService::class, UserServiceHandler::class)->serve(UserService::class)), then call Rpc::call(UserService::class, GetUserRequest::make(['id' => 1])) from anywhere. That single path exercises DTO hydration, typed responses, timeout typing, and tracing headers — the highest-value 30 minutes in this release.

Implementation Patterns

Publish / consume (the baseline you keep using)

• Define DTOs extending TypedMessage; use make() for construction, dispatch() for immediate publish, dispatchLater($payload, $delayMs) for delayed work.
• Resolve the Amqp singleton from the container in Laravel; TypedMessage::dispatch does this for you and throws a clear RuntimeException outside Laravel — useful signal in tests.
• Keep handlers idempotent. At-least-once delivery is unchanged; the new retry/dead-letter tooling makes redelivery more common, not less.

gRPC-lite RPC

• One service class per bounded context; methods() maps request DTO → handler method name. Use name() / exchange() / routingKey() overrides only when you need non-default topology.
• Give each request a responseClass() so Rpc::call() hydrates the reply — this is what makes the RPC layer feel typed rather than array-shaped.
• Branch on failure mode explicitly: RpcTimeoutException = no reply, RpcException = remote handler threw (it carries the original exception class and message — log both).
• Set a sane global Rpc::defaultTimeout(5) at boot; override per call with the third argument and pass extra publish properties as the fourth (['exchange' => 'rpc.svc']) for tenant/routing variants.

Service discovery

• Register short names once in a service provider: Rpc::services()->register('payments', PaymentsService::class). Add public static function alias() to services and use autodiscover([...]) if you prefer convention over a registry list.
• Prefer the fluent caller for call sites that need policy: Rpc::service('payments')->timeout(2)->withProperties([...])->call($request). It keeps timeout/transport concerns out of the call signature.

Sagas

• Use Saga::make()->step('reserve', $reserve)->compensate($release) — the compensation reads next to the step it undoes, which matters during review.
• The legacy step($name, $action, $compensation) signature still works; don't mix styles in one saga.

Retries

• Put #[Retry(attempts: 5, strategy: RetryStrategy::EXPONENTIAL, delayMs: 250, maxDelayMs: 10000, jitter: true)] on the handler method, and build the policy with RetryPolicy::fromAttribute($class, $method) where you wire consumers. One source of truth, no config duplication.
• Use RetryStrategy::NONE for poison-message-prone handlers and route them to dead-letter management instead.

Ops surface

• Amqp::deadLetters()->for($queue)->count() / messages($limit) for inspection; replayTo($target, $limit) and purge() for remediation. Wrap replay in your own audit logging — it is a write operation.
• php artisan amqp:monitor --queue=orders --json in CI/cron/scrape targets; Amqp::dashboard($queues)->snapshot() when you need the array inline (health endpoint, custom widget).

Tracing

• Call CorrelationContext::inheritFromMessage($incoming) at the top of every consumer before publishing downstream. That is what turns message_id into x-causation-id and gives you "this happened because of that" chains.
• applyToPublishProperties() is already part of the publish path; just make sure the context is populated.

Audit / replay

• Attach a store with Amqp::setMessageStore($store). The in-memory default is for tests only — implement MessageStoreInterface (append / find / all / count / purge) against Eloquent or Redis for production, and index the fields you filter on.
• Both publish and consume record automatically when a store is attached; if volume is a concern, attach it selectively per connection/environment.

Laravel events → AMQP

• Implement ShouldPublishToAmqpInterface on the event, set amqp.broadcast_laravel_events => true, and let event(new OrderCreated(...)) publish. Override amqpRouting() / amqpPayload() / amqpExchange() on the event rather than creating a listener — the wildcard AmqpEventListener handles the rest.
• Keep these events small and serialization-safe; they are now wire contracts, not just in-process objects.

Gotchas and Tips

• Nothing is on by default. The event bridge requires broadcast_laravel_events => true; services must be registered or aliased; message stores must be attached. If a feature "does nothing," check opt-in first.
• InMemoryMessageStore is not durable. It vanishes with the worker and is per-process. Fine for tests, wrong for audit trails.
• Dead-letter split behavior. count()/messages() use the RabbitMQ Management API (plugin/credentials required); replayTo()/purge() use the AMQP channel and work without it. A half-configured environment is a common source of "inspection returns empty but replay works."
• RPC error typing is the point. Catch RpcTimeoutException and RpcException separately; retrying a timeout is often correct, retrying a deterministic remote error usually is not. Surface the embedded original exception class in logs.
• #[Retry] is PHP 8+ only. On PHP 7.x the attribute parses as a comment and is silently ignored — tests can pass on 7.3 while production on 8.x behaves differently. Pin your CI matrix accordingly, and remember the package itself must keep parsing as 7.3 (scripts/check-php73-compat-all-src.php); don't introduce 8-only syntax into src/.
• Causation is manual. inheritFromMessage() must be called; it is not wired into the consumer automatically. Forgetting it yields correlation IDs with no causal chain.
• autodiscover() needs public static function alias(). Classes without it are skipped without noise.
• Per-call properties merge, they don't replace. The fourth Rpc::call() argument overrides individual publish properties; verify exchange/routing keys in a staging topology before relying on overrides.
• Handler resolution is container-based. Constructor injection works — avoid mutable state in handlers, since workers are long-lived.
• Dashboard process metrics are per-worker. MetricsCollector counters are in-process; aggregate across workers yourself or rely on the Management API section for queue depth.
• Config caching. Changing amqp.php (especially broadcast_laravel_events) requires php artisan config:clear in cached environments.
• Lumen needs manual wiring. The Rpc and Saga facades are auto-registered via composer.json for Laravel; register providers/facades explicitly under Lumen.
• The release note's public-surface list is truncated. Verify the exact names of ServiceCaller, the new Rpc\* classes, and the Artisan command against the tagged source or IDE autocompletion before writing docs for your team.
• Upgrade is safe, adoption is the risk. Because 3.4.1 is purely additive, the failure mode is architectural drift (three ways to do RPC, two ways to do retries), not breakage. Pick one pattern per concern and document it internally.