Messenger Tarantool Bundle Laravel Package

Getting Started

Minimal Setup

1. Installation

   composer require alexlcdee/messenger-tarantool-bundle
   

   Ensure your config/bundles.php includes:

   Alexlcdee\MessengerTarantoolBundle\AlexlcdeeMessengerTarantoolBundle::class => ['all' => true],
   

2. Configuration Add to config/packages/alexlcdee_messenger_tarantool.yaml:

   messenger_tarantool:
       dsn: 'tarantool://user:password@localhost:3301/'
       space: 'messages'
       queue: 'default'
   

3. First Use Case Create a message class:

   namespace App\Message;
   
   class SendEmail {
       public string $email;
       public string $subject;
   }
   

   Dispatch it via Messenger:

   $bus->dispatch(new SendEmail('user@example.com', 'Welcome!'));
   

4. Worker Setup Run the worker:

   php bin/console messenger:consume messenger_tarantool -vv
   

Implementation Patterns

Core Workflow

1. Message Dispatching Use Symfony Messenger’s BusInterface to send messages to Tarantool:

   $bus->dispatch(new ProcessOrder($orderId));
   

   The bundle serializes and stores messages in Tarantool’s specified space.

2. Handling Messages Create a handler:

   namespace App\MessageHandler;
   
   use App\Message\ProcessOrder;
   use Symfony\Component\Messenger\Attribute\AsMessageHandler;
   
   #[AsMessageHandler]
   class ProcessOrderHandler {
       public function __invoke(ProcessOrder $message) {
           // Process logic
       }
   }
   

3. Queue Management

   • Multiple Queues: Configure multiple queues in config/packages/alexlcdee_messenger_tarantool.yaml:

     messenger_tarantool:
         queues:
             high_priority: { space: 'messages', queue: 'high' }
             default: { space: 'messages', queue: 'default' }
     

   • Dispatch to a specific queue:

     $bus->dispatch(new UrgentTask(), ['queue' => 'high_priority']);
     

4. Retry Logic Leverage Symfony Messenger’s retry middleware (enabled by default). Customize in config/packages/messenger.yaml:

   messenger:
       transports:
           messenger_tarantool:
               dsn: '%env(MESSENGER_TARANTOOL_DSN)%'
               retry_strategy:
                   max_retries: 3
                   delay: 1000
                   multiplier: 2
   

5. Batching Use Tarantool’s batch operations for bulk processing:

   $bus->dispatch(new BatchProcess([$task1, $task2, $task3]), ['queue' => 'batch']);
   

Integration Tips

1. Event-Driven Architecture Combine with Symfony Events for pre/post-processing:

   $dispatcher->addListener(MessageBusInterface::MESSAGE, function ($message) {
       // Log or transform messages
   });
   

2. Monitoring Use Tarantool’s admin interface to monitor queue sizes:

   tarantoolctl --host localhost --port 3301
   > box.space.messages:select()
   

3. Testing Mock Tarantool in PHPUnit:

   $this->getContainer()->get('messenger.transport.messenger_tarantool')->setClient($mockTarantool);
   

4. Async Validation Validate messages before dispatching:

   $validator = $this->getContainer()->get('validator');
   $errors = $validator->validate($message);
   if (count($errors) > 0) {
       throw new \RuntimeException('Validation failed');
   }
   

Gotchas and Tips

Pitfalls

1. Serialization Issues

   • Tarantool uses a binary protocol. Ensure your message classes are serializable:

     use Symfony\Component\Serializer\Annotation\SerializedName;
     
     class UserCreated {
         #[SerializedName('user_id')]
         public int $id;
     }
     

   • Avoid circular references or non-serializable objects (e.g., Doctrine entities without @SerializedName).

2. Connection Timeouts

   • Tarantool connections may hang. Configure timeouts in messenger_tarantool.yaml:

     messenger_tarantool:
         options:
             connect_timeout: 5.0
             recv_timeout: 10.0
     

3. Worker Stuck on Failures

   • If the worker crashes, messages may remain in the queue. Use messenger:failed:remove to clean up:

     php bin/console messenger:failed:remove --limit=100
     

4. Space/Queue Mismatch

   • Double-check space and queue names in config. Tarantool is case-sensitive:

     messenger_tarantool:
         space: 'Messages'  # Wrong if your space is named 'messages'
     

5. Missing Middleware

   • Ensure Symfony\Component\Messenger\Middleware\HandleMessage is included in your pipeline. The bundle adds it by default, but custom setups may omit it.

Debugging

1. Log Messages Enable debug mode in messenger_tarantool.yaml:

   messenger_tarantool:
       debug: true
   

   Logs will appear in var/log/dev.log.

2. Check Tarantool Logs Inspect Tarantool’s logs for connection errors:

   tail -f /var/log/tarantool/tarantool.log
   

3. Verify Message Storage Manually inspect the Tarantool space:

   -- In Tarantool console
   box.space.messages:select()
   

4. Worker Verbosity Run the worker with -vvv for detailed output:

   php bin/console messenger:consume messenger_tarantool -vvv
   

Extension Points

1. Custom Serializer Override the default serializer in config/packages/messenger.yaml:

   messenger:
       transports:
           messenger_tarantool:
               dsn: '%env(MESSENGER_TARANTOOL_DSN)%'
               serializer: App\Serializer\TarantoolSerializer
   

2. Pre/Post Dispatch Hooks Use Symfony’s MessageBus decorator to add logic:

   $bus = new MessageBus([
       new PreDispatchMiddleware(),
       $bus->all(),
       new PostDispatchMiddleware(),
   ]);
   

3. Dynamic Queue Routing Implement Symfony\Component\Messenger\Transport\Serialization\SerializerInterface to route messages dynamically:

   class DynamicQueueSerializer implements SerializerInterface {
       public function decode(array $encoded): array {
           $queue = $encoded['headers']['queue'] ?? 'default';
           return ['queue' => $queue, 'body' => $encoded['body']];
       }
   }
   

4. Tarantool Schema Management Extend the bundle to create/update spaces/queues on install:

   use Doctrine\DBAL\Connection;
   use Symfony\Component\HttpKernel\Bundle\Bundle;
   
   class CustomBundle extends Bundle {
       public function boot() {
           $connection = $this->container->get('database_connection');
           $connection->executeStatement('
               box.schema.space.create("messages", {if_not_exists=true})
           ');
       }
   }