Fs Laravel Package

Getting Started

Minimal Setup

1. Installation

   composer require enqueue/fs
   

   Add to config/queue.php under connections:

   'fs' => [
       'driver' => 'fs',
       'path' => storage_path('queue/fs'),
       'serializer' => 'json', // or 'igbinary', 'php'
       'connection' => 'default', // optional connection name
   ],
   

2. First Use Case Dispatch a job:

   use Enqueue\Fs\FsConnectionFactory;
   use Interop\Queue\PsrMessage;
   
   $connection = (new FsConnectionFactory())->createConnection(['path' => storage_path('queue/fs')]);
   $queue = $connection->createQueue('default');
   
   $message = new PsrMessage(['job' => 'send_email']);
   $queue->send($message);
   

3. Consuming Messages

   $consumer = $connection->createConsumer(['queue_name' => 'default']);
   $consumer->consume(function (PsrMessage $message) {
       // Process message
       return PsrMessage::ACK; // or NACK
   });
   

Implementation Patterns

Core Workflows

1. Job Dispatching

   • Use FsConnectionFactory to create a connection.
   • Dispatch jobs via QueueInterface::send() with PsrMessage objects.
   • Leverage Laravel’s dispatch() helper with a custom queue connection:

     Queue::connection('fs')->dispatch(new SendEmailJob());
     

2. Consuming Jobs

   • Create a consumer with a callback:

     $consumer = $connection->createConsumer(['queue_name' => 'default']);
     $consumer->consume(function (PsrMessage $message) {
         $payload = json_decode($message->getBody(), true);
         // Process payload
         return PsrMessage::ACK;
     });
     

   • Run consumers via Artisan:

     php artisan queue:work --queue=default --connection=fs
     

3. Queue Management

   • List queues:

     $queues = $connection->getQueues();
     

   • Purge a queue:

     $queue = $connection->createQueue('default');
     $queue->purge();
     

Integration Tips

• Laravel Integration Extend Laravel’s Illuminate\Queue\QueueManager to support the fs driver:

  // app/Providers/QueueServiceProvider.php
  public function register()
  {
      Queue::extend('fs', function ($app) {
          return new FsQueueService($app);
      });
  }
  

• Message Serialization Configure serialization in config/queue.php:

  'fs' => [
      'serializer' => 'igbinary', // Faster than JSON for large payloads
  ],
  

• Error Handling Wrap consumer logic in try-catch blocks to handle failures gracefully:

  $consumer->consume(function (PsrMessage $message) {
      try {
          // Process message
          return PsrMessage::ACK;
      } catch (\Exception $e) {
          return PsrMessage::REJECT; // or PsrMessage::REQUEUE
      }
  });
  

Gotchas and Tips

Pitfalls

1. File Permissions Ensure the filesystem path (storage_path('queue/fs')) is writable by the web server:

   chmod -R 775 storage_path('queue/fs')
   

   • Debugging: Check for Permission denied errors in logs.

2. Message Expiry Messages persist indefinitely unless manually deleted. Use purge() or ack()/reject() to manage them.

3. Connection Leaks Always close connections explicitly to avoid resource leaks:

   $connection->close();
   

4. Thread Safety The filesystem transport is not thread-safe. Avoid concurrent writes to the same queue from multiple processes without synchronization.

Debugging

• Log Messages Enable debug logging in config/logging.php to trace message flow:

  'channels' => [
      'single' => [
          'driver' => 'single',
          'path' => storage_path('logs/enqueue.log'),
          'level' => 'debug',
      ],
  ],
  

• Inspect Queue Files Manually check the filesystem directory for stuck messages:

  ls -la storage_path('queue/fs/default')
  

Extension Points

1. Custom Serializers Implement Enqueue\Transport\Serializer\SerializerInterface for custom serialization:

   class CustomSerializer implements SerializerInterface {
       public function serialize($data) { ... }
       public function unserialize($data) { ... }
   }
   

   Register it in the connection config:

   'fs' => [
       'serializer' => CustomSerializer::class,
   ],
   

2. Queue Naming Strategy Override default queue naming by extending FsConnectionFactory:

   class CustomFsConnectionFactory extends FsConnectionFactory {
       protected function getQueueName($queueName) {
           return 'custom_prefix_' . $queueName;
       }
   }
   

3. Message TTL (Time-to-Live) Implement a cleanup cron job to delete expired messages:

   * * * * * php artisan enqueue:fs:cleanup --queue=default --ttl=86400
   

   (Requires custom Artisan command.)

Performance Tips

• Use igbinary Serializer Faster than JSON for large payloads (requires ext-igbinary).
• Batch Processing Process messages in batches to reduce filesystem I/O:

  $consumer->consume(function (PsrMessage $message) {
      // Batch logic here
  }, ['batch_size' => 10]);