Getting Started

Minimal Setup

Installation

composer require dmstr/symfony-job-queue-bundle

return [
    // ...
    Dmstr\JobQueueBundle\DmstrJobQueueBundle::class => ['all' => true],
];

Database Migration Run migrations to create the job table:

php bin/console doctrine:migrations:diff
php bin/console doctrine:migrations:migrate

First Job Dispatch Define a job handler:

namespace App\MessageHandler;

use Dmstr\JobQueueBundle\Message\JobMessage;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
class ProcessUserDataHandler
{
    public function __invoke(JobMessage $message): void
    {
        // Your business logic here
        $message->setResult(['status' => 'processed']);
    }
}

Dispatch a job via JobQueueService:

use Dmstr\JobQueueBundle\Service\JobQueueService;

$jobQueue = $this->container->get(JobQueueService::class);
$job = $jobQueue->dispatch(
    'process_user_data',
    ['user_id' => 123],
    ProcessUserDataHandler::class
);

Monitor Jobs Access the API endpoint at /admin/jobs (if API Platform is configured).

Implementation Patterns

Core Workflow

Job Creation & Dispatch Use JobQueueService::dispatch() to create a persisted job and enqueue it:

$job = $jobQueue->dispatch(
    'job_name',          // Unique identifier
    ['data' => 'payload'], // Serializable payload
    ProcessJobHandler::class, // Handler class
    ['priority' => 5]    // Optional options
);

Validation: Input data is validated against JobMessage constraints.
Persistence: Job is saved to the database before dispatching.

Handler Design Extend JobMessage or use #[AsMessageHandler]:

#[AsMessageHandler]
class MyJobHandler
{
    public function __invoke(JobMessage $message): void
    {
        try {
            $result = $this->process($message->getData());
            $message->setResult($result);
        } catch (\Exception $e) {
            $message->setError($e->getMessage());
        }
    }

    private function process(array $data): array
    {
        // Business logic
    }
}

Status Transitions Jobs automatically transition through states: pending → running (when dispatched) → done/failed (after handler execution). Override transitions via JobMessage methods:
```
$message->setStatus(Job::STATUS_FAILED);
```

Retry Logic Configure retries in config/packages/messenger.yaml:

framework:
    messenger:
        transports:
            async: '%env(MESSENGER_TRANSPORT_DSN)%'
        routing:
            'App\Message\JobMessage': async
        failure_transport: failed
        transports:
            failed: 'doctrine://default?queue_name=failed'
        retry_strategy:
            max_retries: 3
            delay: 1000

Bulk Operations Use JobQueueService for batch dispatch:

$jobs = [];
foreach ($userIds as $id) {
    $jobs[] = $jobQueue->dispatch('process_user', ['id' => $id], UserHandler::class);
}

Integration Tips

Symfony Messenger: Leverage existing transports (e.g., doctrine, amqp, redis).

Doctrine Events: Listen to job lifecycle events:

$dispatcher->addListener(JobEvents::JOB_CREATED, function (JobEvent $event) {
    // Log or notify on job creation
});

API Platform: Expose jobs via Job entity (configure in config/api_platform/resources.yaml):

resources:
    Dmstr\JobQueueBundle\Entity\Job:
        collectionOperations:
            get:
                method: 'GET'
                path: '/admin/jobs'
                normalization_context:
                    groups: ['job:read']

Testing: Use JobQueueService in tests:

$job = $this->createMock(Job::class);
$jobQueue = $this->createMock(JobQueueService::class);
$jobQueue->expects($this->once())
         ->method('dispatch')
         ->willReturn($job);

Gotchas and Tips

Pitfalls

Payload Serialization

Ensure job data is serializable (use arrays, not objects with circular references).

Customize serialization with JobMessage::setData() or override JobMessage:

class CustomJobMessage extends JobMessage
{
    public function __construct(array $data)
    {
        parent::__construct(json_encode($data)); // Custom serialization
    }
}

Handler Execution Context
- Handlers run in a separate process (e.g., Symfony Messenger worker). Avoid:
  - Direct dependency injection of non-serializable services.
  - Stateful operations (use stateless handlers or external storage).
- Fix: Use #[AsMessageHandler] with constructor injection for services:
```
#[AsMessageHandler]
class MyHandler
{
    public function __construct(private MyService $service) {}
}
```
Database Locking
- High concurrency may cause Job table locks. Optimize with:
  - Indexes on status, created_at, updated_at.
  - Batch processing for bulk jobs.
Status Polling
- Avoid polling /admin/jobs for real-time updates. Use:
  - Symfony Messenger failure_transport for retries.
  - WebSocket notifications (e.g., Mercure) for status changes.

API Platform CORS

If accessing /admin/jobs from a frontend, configure CORS in config/packages/nelmio_cors.yaml:

nelmio_cors:
    paths:
        '^/admin/jobs':
            allow_origin: ['*']
            allow_methods: ['GET']
            allow_headers: ['Authorization']

Debugging Tips

Job Stuck in running?

Check Symfony Messenger worker logs:

php bin/console messenger:consume async -vv

Manually update status via Doctrine:

$job = $entityManager->getRepository(Job::class)->find($id);
$job->setStatus(Job::STATUS_FAILED);
$entityManager->flush();

Handler Not Triggered?

Verify the handler is registered in config/packages/messenger.yaml:

framework:
    messenger:
        transports:
            async: '%env(MESSENGER_TRANSPORT_DSN)%'
        routing:
            'App\Message\JobMessage': async

Check for routing conflicts with #[AsMessageHandler].

Payload Corruption

Inspect serialized data in the job table:

SELECT payload FROM job WHERE id = 123;

Use json_decode() to debug:

$payload = json_decode($job->getPayload(), true);

Extension Points

Custom Job Entity Extend Job entity to add fields:

namespace App\Entity;

use Dmstr\JobQueueBundle\Entity\Job as BaseJob;
use Doctrine\ORM\Mapping as ORM;

#[ORM\Entity]
class Job extends BaseJob
{
    #[ORM\Column(type: 'string', length: 255, nullable: true)]
    private ?string $customField = null;

    // Getters/setters
}

Custom Statuses Add new status constants in Job:

final class Job extends BaseJob
{
    public const STATUS_RETRYING = 'retrying';

    // ...
}

Pre/Post Dispatch Hooks Subscribe to events:

use Dmstr\JobQueueBundle\Event\JobEvent;
use Dmstr\JobQueueBundle\Event\JobEvents;
use Symfony\Component\EventDispatcher\Attribute\AsEventListener;

#[AsEventListener(JobEvents::JOB_DISPATCHED)]
public function onJobDispatch(JobEvent $event): void
{
    $job = $event->getJob();
    // Add logic (e.g., send notification)
}

Custom Transport Integrate with non-Messenger transports (e.g., AWS

Symfony Job Queue Bundle Laravel Package