Stomp Laravel Package

Getting Started

Minimal Setup

1. Install the Package

   composer require enqueue/stomp
   

   Ensure stomp-php/stomp-php is installed (required dependency).

2. Configure the DSN Define a DSN (Data Source Name) in your Laravel .env:

   QUEUE_CONNECTION=stomp
   STOMP_DSN="stomp://user:pass@host:port/vhost"
   

   Example with TLS:

   STOMP_DSN="stomp+ssl://user:pass@host:61613/vhost"
   

3. First Use Case: Send a Job

   use Enqueue\Client\Producer;
   use Enqueue\Stomp\StompConnectionFactory;
   
   $connectionFactory = new StompConnectionFactory();
   $connection = $connectionFactory->createConnection(['dsn' => env('STOMP_DSN')]);
   $producer = new Producer($connection);
   
   $producer->send(new \QueueInterop\Message\Message('Hello STOMP!'), 'queue_name');
   

4. Consume Messages

   use Enqueue\Client\Consumer;
   
   $consumer = new Consumer($connection);
   $consumer->consume('queue_name', function ($message) {
       echo $message->getBody();
       $message->ack();
   });
   

Key Files to Review

• config/enqueue.php (Laravel config)
• vendor/enqueue/stomp/src (Core logic)
• Documentation (Advanced usage)

Implementation Patterns

Workflows

1. Producer-Consumer Pattern

   • Producers: Dispatch jobs to STOMP queues (e.g., Laravel jobs, cron tasks).

     $producer->send(new \QueueInterop\Message\Message(json_encode($data)), 'laravel.jobs');
     

   • Consumers: Process messages in a loop (e.g., Laravel queue workers).

     $consumer->consume('laravel.jobs', function ($message) {
         $data = json_decode($message->getBody(), true);
         // Process $data
         $message->ack(); // Acknowledge success
     });
     

2. Laravel Integration

   • Queue Configuration: Extend Laravel’s queue config to use STOMP:

     // config/queue.php
     'connections' => [
         'stomp' => [
             'driver' => 'stomp',
             'dsn' => env('STOMP_DSN'),
             'options' => [
                 'prefetch_count' => 10, // Optimize performance
                 'heartbeat' => 30000,  // 30s heartbeat
             ],
         ],
     ],
     

   • Dispatch Jobs:

     dispatch(new MyJob())->onConnection('stomp');
     

3. Error Handling

   • Use try-catch for message processing:

     try {
         $consumer->consume('queue_name', function ($message) {
             // Process
             $message->ack();
         });
     } catch (\Exception $e) {
         $message->reject(); // Requeue or dead-letter
     }
     

4. Batch Processing

   • Consume multiple messages at once:

     $consumer->consume('queue_name', function ($batch) {
         foreach ($batch as $message) {
             // Process each message
             $message->ack();
         }
     });
     

Integration Tips

• Laravel Horizon: Use STOMP as a backend for Horizon’s monitoring.
• Retry Logic: Implement dead-letter queues (DLQ) for failed jobs.
• Monitoring: Leverage andrewmy/rabbitmq-management-api (included) for STOMP metrics.
• Connection Pooling: Reuse connections for high-throughput scenarios.

Gotchas and Tips

Pitfalls

1. Connection Timeouts

   • STOMP connections may hang if the broker (e.g., ActiveMQ, RabbitMQ) is unreachable.
   • Fix: Implement exponential backoff in reconnection logic.

     $connectionFactory->setReconnectDelay(100); // ms
     

2. Message Ordering

   • STOMP does not guarantee FIFO by default. Use queue-specific features (e.g., RabbitMQ’s x-max-priority) if ordering is critical.

3. Memory Leaks

   • Unacked messages consume broker memory. Always call $message->ack() or $message->reject().
   • Tip: Use setAutoAck(false) in consumers to manually control acknowledgments.

4. SSL/TLS Issues

   • Self-signed certificates may cause StompException. Configure CA certs:

     $connectionFactory->setSslContext([
         'local_cert' => '/path/to/cert.pem',
         'local_pk'   => '/path/to/key.pem',
         'passphrase' => 'password',
     ]);
     

5. Queue Naming

   • STOMP queues are case-sensitive. Ensure consistency between producers/consumers.

Debugging

• Enable Logging:

  $connectionFactory->setLogger(new \Monolog\Logger('stomp'));
  

• Check Connection Status:

  if (!$connection->isConnected()) {
      $connection->reconnect();
  }
  

• Inspect Messages:

  $message->getHeaders(); // Inspect metadata
  $message->getBody();    // Inspect payload
  

Extension Points

1. Custom Middleware

   • Intercept messages with middleware:

     $producer->addMiddleware(new class implements \Enqueue\Client\Middleware {
         public function handle($message, callable $next) {
             // Pre-process
             $result = $next($message);
             // Post-process
             return $result;
         }
     });
     

2. Serialization

   • Override default JSON serialization:

     $message = new \QueueInterop\Message\Message(
         serialize($data),
         ['content-type' => 'application/x-php-serialized']
     );
     

3. Dynamic Queue Routing

   • Route messages to different queues based on headers:

     $queueName = $message->getHeader('queue_name') ?? 'default';
     $producer->send($message, $queueName);
     

4. Performance Tuning

   • Adjust prefetch count for high-throughput:

     $connectionFactory->setOptions(['prefetch_count' => 100]);
     

   • Use async consumers for non-blocking processing:

     $consumer->consumeAsync('queue_name', $callback);