Job Worker Bundle Laravel Package

Getting Started

Minimal Setup

1. Install the Bundle

   composer require abc/job-worker-bundle
   

   Enable the bundle in config/bundles.php:

   return [
       // ...
       Abc\JobWorkerBundle\AbcJobWorkerBundle::class => ['all' => true],
   ];
   

2. Configure the Bundle Add the required configuration in config/packages/abc_job_worker.yaml:

   abc_job_worker:
       server_baseUrl: "http://your-job-server/api/"
       default_queue: "default"
       default_replyTo: "reply"
   

3. Set Up Transport Layer Ensure php-enqueue is configured to match AbcJobServerBundle (e.g., RabbitMQ, Redis, or Doctrine). Example for Redis:

   enqueue:
       client: "redis://localhost"
   

4. First Job Processor Create a processor implementing ProcessorInterface:

   namespace App\Job;
   
   use Abc\JobWorkerBundle\Processor\Context;
   use Abc\JobWorkerBundle\Processor\ProcessorInterface;
   
   class SayHelloProcessor implements ProcessorInterface
   {
       public function process(?string $input, Context $context): void
       {
           echo "Hello, " . $input . "!\n";
       }
   }
   

5. Register the Processor Tag the service in config/services.yaml:

   services:
       App\Job\SayHelloProcessor:
           tags:
               - { name: 'abc.job.processor', jobName: 'say_hello' }
   

6. Run the Worker Start the worker via CLI:

   php bin/console abc:job:worker
   

Implementation Patterns

Workflow: Processing Jobs

1. Job Dispatch Dispatch jobs from your application to AbcJobServerBundle (e.g., via HTTP API or SDK). Example:

   $client = new \GuzzleHttp\Client();
   $response = $client->post('http://your-job-server/api/jobs', [
       'json' => [
           'jobName' => 'say_hello',
           'input' => 'World',
       ],
   ]);
   

2. Worker Consumption The worker (abc:job:worker) polls the configured transport (e.g., Redis queue) for jobs. It routes each job to the corresponding processor based on jobName.

3. Context Utilization Leverage the Context object in processors for:

   • Job metadata (e.g., getJobId(), getReplyTo()).
   • Replying to jobs (e.g., reply(string $output)):

     $context->reply("Processed: " . $input);
     

4. Error Handling Throw exceptions in processors to mark jobs as failed. Use Context::fail(string $message) for custom failure messages.

Integration Tips

1. Dependency Injection Inject services into processors via constructor:

   class EmailProcessor implements ProcessorInterface
   {
       private $mailer;
   
       public function __construct(\Swift_Mailer $mailer)
       {
           $this->mailer = $mailer;
       }
   
       public function process(?string $email, Context $context): void
       {
           $this->mailer->send(...);
       }
   }
   

2. Dynamic Queues Override the default queue per job by adding queueName to the job payload:

   # config/packages/abc_job_worker.yaml
   abc_job_worker:
       queues:
           high_priority: "high_priority_queue"
   

   Then dispatch with:

   { "jobName": "say_hello", "input": "World", "queueName": "high_priority" }
   

3. Retry Logic Configure retries in AbcJobServerBundle (e.g., max retries, delay). The worker respects these settings automatically.

4. Monitoring Log job processing in processors:

   public function process(?string $input, Context $context): void
   {
       \Monolog\Logger::getInstance()->info("Processing job", ['input' => $input]);
   }
   

Gotchas and Tips

Pitfalls

1. Transport Mismatch

   • Issue: Jobs dispatched to AbcJobServerBundle must use the same transport as configured in php-enqueue.
   • Fix: Verify enqueue.client in config/packages/enqueue.yaml matches the server’s transport (e.g., redis://localhost for Redis).

2. JobName Case Sensitivity

   • Issue: jobName in processor tags and job payloads must match exactly (including case).
   • Fix: Use constants or lowercase all job names:

     final class JobNames {
         public const SAY_HELLO = 'say_hello';
     }
     

3. Worker Stuck on Startup

   • Issue: Worker fails silently if no processors are registered or transport is misconfigured.
   • Fix: Enable debug mode and check logs:

     php bin/console debug:container Abc\JobWorkerBundle
     

     Ensure the abc.job.processor tag is correctly applied.

4. Context Not Persisted

   • Issue: Replies or failures sent via Context may not reach the server if the transport connection drops.
   • Fix: Wrap Context operations in a try-catch and log failures:

     try {
         $context->reply($output);
     } catch (\Throwable $e) {
         \Monolog\Logger::getInstance()->error("Reply failed", ['error' => $e->getMessage()]);
     }
     

Debugging Tips

1. Log Job Payloads Add a debug processor to inspect incoming jobs:

   class DebugProcessor implements ProcessorInterface
   {
       public function process(?string $input, Context $context): void
       {
           \Monolog\Logger::getInstance()->debug("Job received", [
               'input' => $input,
               'context' => $context->toArray(),
           ]);
       }
   }
   

2. Test Locally with In-Memory Transport Use enqueue:amqp with localhost for testing:

   enqueue:
       client: "amqp://guest:guest@localhost:5672/%2f"
   

3. Check Worker Status Run the worker in foreground mode for real-time logs:

   php bin/console abc:job:worker --no-daemon
   

Extension Points

1. Custom Transport Adapter Extend Abc\JobWorkerBundle\Transport\TransportInterface to support unsupported transports (e.g., AWS SQS).

2. Middleware for Jobs Create a middleware stack for processors (e.g., logging, validation):

   namespace App\Job\Middleware;
   
   use Abc\JobWorkerBundle\Processor\Context;
   use Closure;
   
   class LoggingMiddleware
   {
       public function __invoke(Closure $next, ?string $input, Context $context)
       {
           \Monolog\Logger::getInstance()->info("Processing job", ['input' => $input]);
           return $next($input, $context);
       }
   }
   

   Register it in services.yaml:

   services:
       App\Job\Middleware\LoggingMiddleware:
           tags:
               - { name: 'abc.job.middleware' }
   

3. Batch Processing Use abc:job:worker --batch-size=10 to process jobs in batches (experimental; check bundle docs for updates).

4. Custom Job Serialization Override serialization in Abc\JobWorkerBundle\Serializer\SerializerInterface for non-JSON payloads (e.g., XML).