Dbal Laravel Package

Getting Started

Minimal Setup

1. Installation Add the package via Composer:

   composer require enqueue/dbal
   

   Ensure doctrine/dbal is installed (required dependency).

2. Basic Configuration Register the transport in your Laravel service provider (e.g., AppServiceProvider):

   use Enqueue\Dbal\DbalConnectionFactory;
   use Enqueue\Client\Extension\Context;
   use Doctrine\DBAL\Connection;
   
   public function register()
   {
       $connection = new Connection([
           'url' => env('DATABASE_URL'),
           'driver' => 'pdo_mysql',
           'host' => env('DB_HOST'),
           'dbname' => env('DB_NAME'),
           'user' => env('DB_USER'),
           'password' => env('DB_PASSWORD'),
       ]);
   
       $this->app->singleton('enqueue.dbal.connection', function () use ($connection) {
           return DbalConnectionFactory::createWithContext(
               new Context(),
               $connection
           );
       });
   }
   

3. First Use Case: Publishing a Message Inject the connection into a service and publish a message:

   use Enqueue\Client\Producer;
   
   public function publishMessage(Producer $producer)
   {
       $producer->send(new Message('Hello, Queue!', ['priority' => 1]));
   }
   

Implementation Patterns

Workflow: Queue Consumer

1. Bootstrap Consumer Define a consumer in a Laravel command:

   use Enqueue\Client\Consumer;
   use Enqueue\Dbal\DbalConnectionFactory;
   
   class QueueWorker extends Command
   {
       protected $signature = 'queue:work';
       public function handle()
       {
           $connection = app('enqueue.dbal.connection');
           $consumer = new Consumer($connection);
   
           $consumer->setCallback(function (Message $message, Context $context) {
               // Process message logic
               $this->info('Processed: ' . $message->getBody());
               return new Response(200);
           });
   
           $consumer->run();
       }
   }
   

2. Handling Failures Use setErrorCallback to log or retry failed messages:

   $consumer->setErrorCallback(function (Message $message, Context $context, $e) {
       Log::error('Failed to process message: ' . $e->getMessage());
       return new Response(500);
   });
   

Integration Tips

• Laravel Queues: Use alongside Laravel’s queue system by extending Illuminate\Queue\QueueManager to include the DBAL transport.
• Batch Processing: Leverage DbalConnectionFactory’s createBatch() for bulk operations.
• Connection Pooling: Reuse the Connection instance across consumers to avoid overhead.

Gotchas and Tips

Pitfalls

1. Transaction Isolation DBAL transport relies on database transactions. Ensure your database supports REPEATABLE READ or higher isolation levels to avoid phantom reads during message processing.

2. Locking Mechanism Messages are locked using database rows. Long-running tasks may cause locks to expire (default: 30 seconds). Adjust lock_ttl in the connection factory if needed:

   DbalConnectionFactory::createWithContext(
       new Context(['lock_ttl' => 60]), // 60-second lock
       $connection
   );
   

3. Schema Migrations The package creates tables automatically, but manual schema changes (e.g., altering columns) may break functionality. Avoid altering enqueue_jobs or enqueue_lock tables.

Debugging

• Check Table Contents Inspect enqueue_jobs for stuck messages:

  SELECT * FROM enqueue_jobs WHERE status = 'waiting';
  

• Enable Logging Configure DBAL logging via Doctrine:

  $connection->getConfiguration()->setSQLLogger(new \Doctrine\DBAL\Logging\EchoSQLLogger());
  

Extension Points

1. Custom Message Serialization Override the default JSON serializer by extending Enqueue\Client\Message and injecting a custom serializer into the Context:

   $context = new Context(['serializer' => new CustomSerializer()]);
   

2. Queue Priorities Use the priority field in the message body to implement priority queues. The DBAL transport supports this natively.

3. Dead Letter Queue (DLQ) Implement a DLQ by extending the ErrorCallback to move failed messages to a separate table or queue. Example:

   $consumer->setErrorCallback(function (Message $message, Context $context, $e) {
       $dlqProducer = new Producer($connection);
       $dlqProducer->send(new Message($message->getBody(), [
           'original_priority' => $message->getProperty('priority'),
           'failed_at' => now(),
       ]));
       return new Response(500);
   });