PHPackages                             yangusik/thrun-laravel - PHPackages - PHPackages  [Skip to content](#main-content)[PHPackages](/)[Directory](/)[Categories](/categories)[Trending](/trending)[Leaderboard](/leaderboard)[Changelog](/changelog)[Analyze](/analyze)[Collections](/collections)[Log in](/login)[Sign up](/register)

1. [Directory](/)
2. /
3. [Queues & Workers](/categories/queues)
4. /
5. yangusik/thrun-laravel

ActiveLibrary[Queues & Workers](/categories/queues)

yangusik/thrun-laravel
======================

Laravel adapter for Thrun async queue worker

v0.2.5(4w ago)115[3 issues](https://github.com/YanGusik/thrun_laravel/issues)MITPHPPHP ^8.4

Since Jun 5Pushed 4w ago2 watchersCompare

[ Source](https://github.com/YanGusik/thrun_laravel)[ Packagist](https://packagist.org/packages/yangusik/thrun-laravel)[ RSS](/packages/yangusik-thrun-laravel/feed)WikiDiscussions master Synced 1w ago

READMEChangelog (6)Dependencies (8)Versions (9)Used By (0)

Thrun Laravel
=============

[](#thrun-laravel)

Laravel adapter for the async queue worker [**Thrun**](https://github.com/yangusik/thrun). Built on real OS threads (TrueAsync) and provides two workflow styles: **clean architecture** (recommended) and **self-handling jobs** (for simple tasks).

---

Benchmarks
----------

[](#benchmarks)

Measured on WSL2, 8GB RAM, PHP 8.6 TrueAsync fork:

ScenarioConfigJobsTimeThroughputRSSHorizon IO12 workers1,00012.1s83/s872 MBThrun IO1 thread, 100 coroutines1,0002.3s434/s80 MBHorizon IO12 workers10,00055.0s182/s1019 MBThrun IO1 thread, 100 coroutines10,0006.3s1580/s84 MBHorizon CPU12 workers10018.4s5.4/s1022 MBThrun CPU12 threads10016.3s6.1/s100 MBHorizon CPU12 workers1,000162.6s6.2/s1023 MBThrun CPU12 threads1,000139.5s7.2/s101 MBHorizon NOOP12 workers1,0005.0s198/s656 MBThrun NOOP12 threads1,0002.3s434/s103 MBTrueAsync 12x10 uses **17x less RSS** than Horizon 12 workers, **11x more IO throughput**.

---

Installation
------------

[](#installation)

```
composer require yangusik/thrun-laravel
```

Publish the configuration file:

```
php artisan vendor:publish --tag=thrun-config
```

---

Configuration (`config/thrun.php`)
----------------------------------

[](#configuration-configthrunphp)

### Redis

[](#redis)

```
'redis' => [
    'host'    => env('THRUN_REDIS_HOST', '127.0.0.1'),
    'port'    => (int) env('THRUN_REDIS_PORT', 6379),
    'prefix'  => env('THRUN_REDIS_PREFIX', 'thrun:queue'),
    'timeout' => 1.0,
],
```

> **Note:** `prefix` affects Redis keys. Default is `thrun:queue`, but you can change it when multiple environments share the same Redis instance.

### Queues

[](#queues)

Each queue is a separate transport. Currently supported: `redis` and `memory`.

```
'queues' => [
    'emails'        => ['transport' => 'redis'],
    'notifications' => ['transport' => 'memory'],
],
```

`memory` transport holds jobs in process memory with no external dependencies. Primarily useful for local development and testing when you don't want to run Redis.

All `memory` queues are automatically exposed on the RPC server — external processes can push jobs into them over a socket.

> **Note:** `memory` queues are designed for up to `queue_size` jobs in flight. Sending significantly more jobs than `queue_size` allows provides no delivery guarantees. Use `redis` for workloads that require guaranteed delivery.

> FOR MEMORY USE: `$bus->dispatchViaRpc();`

### Worker configuration

[](#worker-configuration)

```
'worker' => [
    'threads'     => env('THRUN_WORKER_THREADS', 2),
    'concurrency' => env('THRUN_WORKER_CONCURRENCY', 100),
    'queue_size'  => env('THRUN_WORKER_QUEUE_SIZE', 1000),
],
```

- `threads` — number of OS threads. For I/O-bound workloads `1` is enough; increase for CPU-bound tasks.
- `concurrency` — number of coroutines per thread.
- `queue_size` — internal job buffer size. For `redis` queues this is a soft throughput limiter — Redis keeps the rest. For `memory` queues the buffer is the only storage, so tune this to match your expected peak load.

### Supervisors

[](#supervisors)

Each supervisor is an isolated worker group with its own queues, strategy, and policies.

```
'supervisors' => [
    'default' => [
        'queues'    => ['emails', 'notifications'],
        'worker'    => ['threads' => 2, 'concurrency' => 100, 'queue_size' => 1000],
        'supervisor'=> ['max_crashes' => 3, 'restart_window' => 300, 'restart_backoff' => 1.0],
        'strategy'  => ['class' => PriorityStrategy::class, 'priorities' => ['emails' => 3, 'notifications' => 1]],
        'policy'    => ['enabled' => false, 'class' => MaxConcurrencyPolicy::class, 'options' => ['max_per_partition' => 5]],
        'handlers'  => [],              // manual routing
    ],
],
```

### RPC Server

[](#rpc-server)

The RPC server runs alongside the worker and enables cross-process job dispatch and event broadcasting.

```
'rpc' => [
    'enabled'     => env('THRUN_RPC_ENABLED', true),
    'transport'   => env('THRUN_RPC_TRANSPORT', 'unix'), // unix | tcp
    'socket_path' => env('THRUN_RPC_SOCKET', '/tmp/thrun_rpc.sock'),
    'host'        => env('THRUN_RPC_HOST', '127.0.0.1'),
    'port'        => (int) env('THRUN_RPC_PORT', 9000),
],
```

### Failed Jobs

[](#failed-jobs)

When a message exhausts all retries, it is sent to the global failed job store (configured separately from supervisors):

```
'failed' => [
    'driver' => env('THRUN_FAILED_DRIVER', 'redis'),
    'redis' => [
        'prefix' => env('THRUN_FAILED_PREFIX', 'thrun:failed'),
    ],
],
```

Supported drivers: `redis` (default) and `null` (no-op).

You can register custom failed job drivers:

```
use Thrun\Laravel\Transport\TransportFactory;

$factory = app(TransportFactory::class);
$factory->extendFailed('database', function (array $config) {
    return new DatabaseFailedJobSender(
        table: $config['database']['table'] ?? 'thrun_failed_jobs',
    );
});
```

### Auto-discover

[](#auto-discover)

```
'auto_discover' => [
    'App\\Handlers',   // regular Handler classes
    'App\\Jobs',       // self-handling Job classes
    'App\\Events',     // event listeners
],
```

---

Two Workflow Styles
-------------------

[](#two-workflow-styles)

### 1. Clean Architecture (recommended) — Message + Handler

[](#1-clean-architecture-recommended--message--handler)

Message is a DTO with data. Handler is an invokable class with business logic. Pure Symfony-style separation of concerns.

**Message:**

```
namespace App\Messages;

use Thrun\Laravel\Handler\Attribute\Delay;
use Thrun\Laravel\Handler\Attribute\Queue;
use Thrun\Laravel\Handler\Attribute\Retry;

#[Queue('emails')]
#[Retry(backoff: [1000, 2000, 4000], maxAttempts: 3)]
#[Delay(5000)]
final readonly class SendEmailMessage
{
    public function __construct(
        public string $to,
        public string $subject,
    ) {}
}
```

**Handler:**

```
namespace App\Handlers;

use App\Messages\SendEmailMessage;
use Thrun\Laravel\Handler\AsThrunHandler;
use Thrun\Worker\Acknowledger;

#[AsThrunHandler] // auto-wired to SendEmailMessage
final class SendEmailHandler
{
    public function __construct(private MailerInterface $mailer) {}

    public function __invoke(SendEmailMessage $message, Acknowledger $ack): void
    {
        $this->mailer->send($message->to, $message->subject);
        $ack->ack();
    }
}
```

**Dispatch:**

```
use Thrun\Laravel\Bus\ThrunMessageBus;
use Thrun\Laravel\Bus\DispatchOptions;

$bus->dispatch(new SendEmailMessage('user@test.com', 'Hello'));

// or with option override:
$bus->dispatch(
    new SendEmailMessage('user@test.com', 'Hello'),
    'emails',
    new DispatchOptions(delayMs: 10_000, messageId: 'email-42'),
);
```

---

### 2. Alternative — Self-handling Job

[](#2-alternative--self-handling-job)

A single class acts as both Message and Handler. Data goes in the constructor, logic in `__invoke()`. For simple tasks when you don't want extra files.

```
namespace App\Jobs;

use Thrun\Laravel\Handler\Attribute\Queue;
use Thrun\Laravel\Handler\Attribute\Retry;
use Thrun\Laravel\Handler\Attribute\ThrunJob;
use Thrun\Worker\Acknowledger;

#[ThrunJob]
#[Queue('emails')]
#[Retry(backoff: [1000, 2000, 4000], maxAttempts: 3)]
final readonly class SendEmailJob
{
    public function __construct(
        public string $to,
        public string $subject,
    ) {}

    public function __invoke(MailerInterface $mailer, Acknowledger $ack): void
    {
        $mailer->send($this->to, $this->subject);
        $ack->ack();
    }
}
```

```
// queue is taken from #[Queue] attribute
$bus->dispatch(new SendEmailJob('user@test.com', 'Hello'));

// or override:
$bus->dispatch(new SendEmailJob('user@test.com', 'Hello'), 'urgent-emails');
```

> **Important:** the constructor accepts **scalar data only** (`int`, `string`, `array`, etc.) — these values get serialized to Redis. Never inject services or objects into the constructor. All services are injected via DI in `__invoke()` — Laravel `Container::call()` resolves them automatically.

---

Attributes
----------

[](#attributes)

AttributePurposeApplies to`#[ThrunJob]`Self-handling job markerJob`#[Queue('emails')]`Default queue for the messageJob / Message`#[Retry(backoff: [...], maxAttempts: 3)]`Retry policyJob / Message`#[Delay(5000)]`Delay in msJob / Message`#[Timeout(30000)]`Hard execution timeoutJob / Message`#[AsThrunHandler(messageClass: ...)]`Explicit Handler → Message bindingHandler`#[ThrunEventListener('order.completed')]`Event listener bindingEvent Listener**Dispatch priority:**

1. `dispatch()` argument (explicit)
2. `#[Queue]` attribute on class
3. `'default'`

---

Acknowledger
------------

[](#acknowledger)

`Acknowledger` is the explicit processing acknowledgement object.

```
public function __invoke(MyMessage $message, Acknowledger $ack): void
{
    // ... logic ...
    $ack->ack();   // confirm success
    // $ack->nack(); // reject (goes to retry or failure transport)
}
```

> **Recommendation:** always accept `Acknowledger $ack` explicitly and call `$ack->ack()`. This gives you full control over the message lifecycle.

---

Events
------

[](#events)

Thrun provides a lightweight pub/sub event system over the RPC socket. Events are **ephemeral** (fire-and-forget, no persistence) — use jobs for guaranteed delivery.

### Emitting an event from a handler

[](#emitting-an-event-from-a-handler)

```
use Thrun\Laravel\Rpc\RpcPublisher;

final class ProcessOrderHandler
{
    public function __construct(private RpcPublisher $rpc) {}

    public function __invoke(ProcessOrderMessage $message, Acknowledger $ack): void
    {
        // ... business logic ...
        $this->rpc->emit('order.completed', ['order_id' => $message->orderId]);
        $ack->ack();
    }
}
```

### Registering a listener

[](#registering-a-listener)

```
namespace App\Events;

use Thrun\Laravel\Event\Attribute\ThrunEventListener;

#[ThrunEventListener('order.completed')]
final class OrderCompletedListener
{
    public function __construct(private readonly MailService $mail) {}

    public function __invoke(array $payload): void
    {
        $this->mail->sendConfirmation($payload['order_id']);
    }
}
```

Wildcard subscriptions:

```
#[ThrunEventListener('*')]                // all events
#[ThrunEventListener('payment.*')]        // all payment.* events
#[ThrunEventListener]                     // uses class name as identifier (PHP-only)
```

### Running the listener process

[](#running-the-listener-process)

```
php artisan thrun:event

# subscribe to specific patterns
php artisan thrun:event --subscribe=order.completed --subscribe=payment.*
```

### Emitting from another language (Go example)

[](#emitting-from-another-language-go-example)

```
payload := map[string]any{
    "event": "order.completed",
    "data":  map[string]any{"order_id": 123},
}
body, _ := json.Marshal(payload)

// FrameType::Event = 0x02
frame := make([]byte, 5+len(body))
binary.BigEndian.PutUint32(frame[:4], uint32(len(body)))
frame[4] = 0x02
copy(frame[5:], body)
conn.Write(frame)
```

---

Wire Protocol
-------------

[](#wire-protocol)

All RPC communication uses a simple length-prefixed binary framing: \[4 bytes BE uint32 — payload length\]\[1 byte — frame type\]\[N bytes — JSON payload\]

TypeByteDirectionPurpose`Job``0x01`client → serverPush job into a local memory queue`Event``0x02`client → serverBroadcast event to subscribers`Subscribe``0x03`client → serverRegister interest in an event name/pattern`RpcRequest``0x04`client → serverSynchronous call, expects `RpcReply``RpcReply``0x05`server → clientResponse to `RpcRequest``Error``0x06`server → clientError response---

---

Auto-discover
-------------

[](#auto-discover-1)

ConventionResult`#[AsThrunHandler]` on handler classExplicit binding to `messageClass``SendEmailHandler` → `SendEmailMessage`Naming convention auto-wire`#[ThrunJob]` on invokable classClass registers as its own handler`#[ThrunEventListener('event.name')]`Binds listener to event name/pattern---

Middleware
----------

[](#middleware)

You can register worker middleware per supervisor via config. Classes are resolved through the Laravel container (constructor injection is supported).

```
'supervisors' => [
    'default' => [
        // ...
        'middleware' => [
            \App\Middleware\LogMiddleware::class,
            \App\Middleware\MetricsMiddleware::class,
        ],
    ],
],
```

A middleware must implement `WorkerMiddlewareInterface` from the core `thrun` package:

```
namespace App\Middleware;

use Thrun\Worker\Acknowledger;
use Thrun\Contract\WorkerMiddlewareInterface;

final class LogMiddleware implements WorkerMiddlewareInterface
{
    public function handle(object $message, Acknowledger $ack, \Closure $next): void
    {
        try {
            $next($message, $ack);
        } catch (\Throwable $e) {
            // log, metrics, etc.
            throw $e;
        }
    }
}
```

---

Failed Jobs & CLI
---------------------

[](#failed-jobs--cli)

When a message exhausts all retries, it is persisted to the failed job store (configured in `config/thrun.php` under `failed`).

### List failed jobs

[](#list-failed-jobs)

```
php artisan thrun:failed
php artisan thrun:failed --queue=emails
php artisan thrun:failed --limit=100
```

### Show details of a failed job

[](#show-details-of-a-failed-job)

```
php artisan thrun:failed:show 019e9c83-c3d7-7216-b37d-04b1c154a5c8
```

Shows: type, queue, exception, message, file, line, full trace, payload, stamps.

### Retry a failed job

[](#retry-a-failed-job)

```
php artisan thrun:retry 019e9c83-c3d7-7216-b37d-04b1c154a5c8
php artisan thrun:retry --all
```

Retry creates a **new** message with a fresh `JobIdStamp` but preserves `MessageIdStamp`.

### Flush failed jobs

[](#flush-failed-jobs)

```
php artisan thrun:failed:flush
```

### Flush queues

[](#flush-queues)

```
# Flush a specific queue (ready, processing, delayed)
php artisan thrun:flush emails

# Flush all configured queues
php artisan thrun:flush

# Flush queues + failed jobs
php artisan thrun:flush --failed
```

---

Running the Worker
------------------

[](#running-the-worker)

```
php artisan thrun:work                    # all supervisors + RPC server
php artisan thrun:work --supervisor=X     # single supervisor
php artisan thrun:work --stats            # with real-time throughput stats
php artisan thrun:work --no-rpc           # without RPC server
```

---

---

Commands Reference
------------------

[](#commands-reference)

CommandDescription`thrun:work`Start all supervisors + RPC server`thrun:work --supervisor=X`Start a single supervisor + RPC server`thrun:work --stats`Real-time throughput stats`thrun:work --no-rpc`Disable RPC server for this process`thrun:event`Listen and dispatch incoming events`thrun:event --subscribe=X`Subscribe to specific pattern`thrun:failed`List failed jobs`thrun:failed:show {id}`Show details of a failed job`thrun:retry {id}`Retry a specific failed job`thrun:retry --all`Retry all failed jobs`thrun:failed:flush`Delete all failed jobs`thrun:flush {queue?}`Flush queue(s)`thrun:flush --failed`Flush queues + failed jobs---

DispatchOptions
---------------

[](#dispatchoptions)

Explicit stamp control when dispatching (overrides class attributes):

```
use Thrun\Laravel\Bus\DispatchOptions;

$bus->dispatch($message, 'emails', new DispatchOptions(
    messageId: 'uuid-42',
    delayMs: 5000,
    retryBackoff: [1000, 2000, 4000],
    maxAttempts: 3,
    timeoutMs: 30000,
));
```

For edge cases you can use `dispatchCustom()` with a ready-made `Envelope`:

```
use Thrun\Envelope\Envelope;
use Thrun\Envelope\Stamp\QueueStamp;

$bus->dispatchCustom(
    Envelope::wrap($message, new QueueStamp('custom')),
    'emails',
);
```

---

Message IDs
-----------

[](#message-ids)

You can generate dynamic message IDs directly from the message payload using `IdentifiableMessage`:

```
use Thrun\Laravel\Contract\IdentifiableMessage;

#[Queue('emails')]
final readonly class SendEmailMessage implements IdentifiableMessage
{
    public function __construct(
        public string $to,
        public string $subject,
        public int $userId,
        public int $productId,
    ) {}

    public function getId(): string
    {
        return "{$this->userId}-{$this->productId}";
    }
}
```

```
$bus->dispatch(new SendEmailMessage('a@b.com', 'Hi', 42, 7));
// messageId = "42-7" automatically
```

Priority:

1. `DispatchOptions->messageId`
2. `IdentifiableMessage->getId()`
3. `null` (no ID)

---

Fluent Builder
--------------

[](#fluent-builder)

For quick one-off overrides without creating a `DispatchOptions` object:

```
$bus->builder()
    ->id('custom-42')
    ->retry([1000, 2000], 3)
    ->delay(5000)
    ->timeout(30000)
    ->send($message, 'emails');
```

---

Extending Transports
--------------------

[](#extending-transports)

You can register custom transports (e.g. RabbitMQ) via closures or config drivers.

### Closure driver

[](#closure-driver)

```
use Thrun\Laravel\Transport\TransportFactory;

$factory = app(TransportFactory::class);
$factory->extend('rabbitmq', function (string $name, array $config) {
    return new RabbitMQTransport(
        host: $config['host'],
        port: $config['port'],
        queue: $name,
    );
});
```

### Config-based driver

[](#config-based-driver)

```
'queues' => [
    'orders' => [
        'transport' => 'custom',
        'driver'    => \App\Transport\RabbitMQTransport::class,
        'host'      => 'localhost',
    ],
],
```

The factory will try to resolve the class via Laravel `Container::make()` with `['name' => ..., 'config' => ...]`, or fall back to `new $driverClass($name, $config)`.

---

Requirements
------------

[](#requirements)

- PHP (TrueAsync Core) ^8.6
- Laravel ^11.0
- [ext-async](https://github.com/true-async/php-async) (TrueAsync extension)
- [ext-phpredis](https://github.com/true-async/phpredis) (TrueAsync fork) (if using Redis transport)

###  Health Score

41

—

FairBetter than 87% of packages

Maintenance88

Actively maintained with recent releases

Popularity12

Limited adoption so far

Community8

Small or concentrated contributor base

Maturity46

Maturing project, gaining track record

 Bus Factor1

Top contributor holds 100% of commits — single point of failure

How is this calculated?**Maintenance (25%)** — Last commit recency, latest release date, and issue-to-star ratio. Uses a 2-year decay window.

**Popularity (30%)** — Total and monthly downloads, GitHub stars, and forks. Logarithmic scaling prevents top-heavy scores.

**Community (15%)** — Contributors, dependents, forks, watchers, and maintainers. Measures real ecosystem engagement.

**Maturity (30%)** — Project age, version count, PHP version support, and release stability.

###  Release Activity

Cadence

Every ~3 days

Total

7

Last Release

29d ago

### Community

Maintainers

![](https://www.gravatar.com/avatar/ca757be9e125f187e3fc6f17f8f71463c29e0b083ef8dfb1c1616adc71381f50?d=identicon)[YanGusik](/maintainers/YanGusik)

---

Top Contributors

[![YanGusik](https://avatars.githubusercontent.com/u/28189620?v=4)](https://github.com/YanGusik "YanGusik (21 commits)")

### Embed Badge

![Health badge](/badges/yangusik-thrun-laravel/health.svg)

```
[![Health](https://phpackages.com/badges/yangusik-thrun-laravel/health.svg)](https://phpackages.com/packages/yangusik-thrun-laravel)
```

###  Alternatives

[laravel/ai

The official AI SDK for Laravel.

1.0k3.2M246](/packages/laravel-ai)[illuminate/queue

The Illuminate Queue package.

20432.6M1.7k](/packages/illuminate-queue)[laravel/mcp

Rapidly build MCP servers for your Laravel applications.

77922.3M186](/packages/laravel-mcp)[mike-bronner/laravel-model-caching

Automatic caching for Eloquent models.

2.4k96.5k1](/packages/mike-bronner-laravel-model-caching)[psalm/plugin-laravel

Psalm plugin for Laravel

3345.3M347](/packages/psalm-plugin-laravel)[aedart/athenaeum

Athenaeum is a mono repository; a collection of various PHP packages

255.2k](/packages/aedart-athenaeum)

PHPackages © 2026

[Directory](/)[Categories](/categories)[Trending](/trending)[Changelog](/changelog)[Analyze](/analyze)