Async Laravel Package

Getting Started

Minimal Setup

1. Install the Package

   composer require aescarcha/async
   

2. Configure RabbitMQ Update config/packages/old_sound_rabbit_mq.yaml (or config.yml in older Symfony versions) with your RMQ connection details:

   old_sound_rabbit_mq:
       connections:
           default:
               host:     'localhost'
               port:     5672
               user:     'guest'
               password: 'guest'
               vhost:    '/'
       producers:
           async_refresher:
               connection: default
               exchange_options: {name: 'async-refresher', type: direct}
       consumers:
           async_refresher:
               connection: default
               exchange_options: {name: 'async-refresher', type: direct}
               queue_options: {name: 'async-refresher'}
               callback: aescarcha.persist_listener
   

3. Register the Listener Add the service definition to config/services.yaml:

   services:
       aescarcha.persist_listener:
           class: Aescarcha\AsyncBundle\Listener\PersistListener
           arguments: ['@serializer', '@doctrine.orm.entity_manager']
           tags:
               - { name: kernel.event_listener, event: 'prePersist', method: 'onPrePersist' }
               - { name: kernel.event_listener, event: 'preUpdate', method: 'onPreUpdate' }
               - { name: kernel.event_listener, event: 'preRemove', method: 'onPreRemove' }
   

4. First Use Case Trigger async operations on entity events (e.g., prePersist). Example:

   // In your entity or controller
   $entity = new YourEntity();
   $entityManager->persist($entity);
   $entityManager->flush(); // Async job is queued here
   

Implementation Patterns

Workflow: Async Entity Operations

1. Event-Driven Queueing

   • Attach listeners to Doctrine lifecycle events (prePersist, preUpdate, preRemove).
   • Example listener logic:

     public function onPrePersist(EntityManagerInterface $em, Entity $entity)
     {
         $job = new AsyncJob($entity, 'persist');
         $this->asyncProducer->publish($job);
     }
     

2. Consumer Processing

   • The async_refresher consumer processes queued jobs:

     // In AsyncBundle's PersistListener
     public function onMessage(AsyncJob $job)
     {
         switch ($job->getAction()) {
             case 'persist':
                 $this->entityManager->persist($job->getEntity());
                 break;
             case 'remove':
                 $this->entityManager->remove($job->getEntity());
                 break;
         }
         $this->entityManager->flush();
     }
     

3. Serialization Layer

   • Leverage Symfony’s Serializer to handle entity serialization/deserialization:

     # services.yaml
     Aescarcha\AsyncBundle\Serializer\EntitySerializer:
         arguments: ['@serializer']
     

4. Integration with Doctrine

   • Use Doctrine\ORM\Event\LifecycleEventArgs to access the entity manager:

     public function onPrePersist(LifecycleEventArgs $args)
     {
         $entity = $args->getEntity();
         $em = $args->getEntityManager();
         // Queue async job
     }
     

Common Patterns

• Bulk Operations: Queue multiple entities in a single batch to reduce RMQ overhead.
• Error Handling: Implement retry logic in the consumer for failed jobs.
• Custom Actions: Extend AsyncJob to support additional actions (e.g., reindex, notify).

Gotchas and Tips

Pitfalls

1. RabbitMQ Connection Issues

   • Symptom: Jobs fail silently or time out.
   • Fix: Verify old_sound_rabbit_mq configuration (host, credentials, vhost). Enable lazy: true for connection pooling:

     connections:
         default:
             lazy: true
     

2. Entity Serialization Failures

   • Symptom: Jobs fail with SerializationException.
   • Fix: Ensure entities implement Serializable or configure the Serializer to handle them:

     # config/packages/serializer.yaml
     framework:
         serializer:
             mapping:
                 paths: ['%kernel.project_dir%/config/serializer']
     

     Create a mapping file (e.g., config/serializer/YourEntity.yaml):

     Your\App\Entity\YourEntity:
         attributes:
             id: ~
             name: ~
     

3. Consumer Not Processing Messages

   • Symptom: Jobs are queued but never executed.
   • Fix: Check if the consumer is running:

     php bin/console rabbitmq:consumer async_refresher
     

     Ensure the callback in consumers matches the service ID (aescarcha.persist_listener).

4. Transaction Management

   • Gotcha: Async jobs run in separate transactions. Use flush() explicitly in the consumer to persist changes.

Debugging Tips

• Enable RabbitMQ Logging:

  old_sound_rabbit_mq:
      connections:
          default:
              log_level: debug
  

• Inspect Queues: Use a tool like RabbitMQ Management Plugin to monitor queues and messages.
• Test Locally: Use Docker to spin up RabbitMQ for development:

  # docker-compose.yml
  services:
      rabbitmq:
          image: rabbitmq:3-management
          ports:
              - "5672:5672"
              - "15672:15672"
  

Extension Points

1. Custom Job Types Extend AsyncJob to support domain-specific actions:

   class ReindexJob extends AsyncJob
   {
       public function __construct(Entity $entity, string $indexName)
       {
           parent::__construct($entity, 'reindex');
           $this->indexName = $indexName;
       }
   }
   

2. Dynamic Queue Routing Use exchange routing keys to direct jobs to different consumers:

   producers:
       async_refresher:
           exchange_options: {name: 'async-refresher', type: direct}
           routing_key: 'entity.{entity_class}'
   

3. Priority Queues Configure multiple queues with priorities:

   consumers:
       async_high_priority:
           queue_options: {name: 'async-high-priority', priority: 1}