Nats Php Laravel Package

Getting Started

Minimal Setup

1. Installation

   composer require elandlord/nats-php
   

   Requires PHP 8.1+ and NATS server (v2.9+ recommended).

2. First Connection

   use Elandlord\NatsPhp\Client;
   
   $client = new Client(['servers' => ['nats://localhost:4222']]);
   $client->connect();
   

3. Publish a Message

   $client->publish('user.created', json_encode(['id' => 123]));
   

4. Subscribe to Messages

   $subscription = $client->subscribe('user.created');
   $subscription->onMessage(function ($msg) {
       echo $msg->getData(); // Decode JSON as needed
   });
   

Where to Look First

• Documentation: GitHub README (if available) or tests/ directory for usage examples.
• Core Classes:
  • Client (connection management)
  • Subscription (message handling)
  • JetStreamClient (for JetStream features)
• First Use Case: Replace direct NATS client usage in a Laravel queue worker or event system.

Implementation Patterns

Workflow: Event-Driven Architecture

1. Publish Events

   // In a Laravel controller/service
   event(new UserCreated($user));
   // Or directly via NATS
   $client->publish('user.created', $user->toJson());
   

2. Consume Events

   // In a NATS consumer service
   $subscription = $client->subscribe('user.created');
   $subscription->onMessage(function ($msg) {
       $user = json_decode($msg->getData(), true);
       // Process $user
   });
   

3. Laravel Integration

   // Queue worker (app/Console/Kernel.php)
   protected function schedule(Schedule $schedule)
   {
       $schedule->command('nats:consume')->everyMinute();
   }
   

JetStream Patterns

1. Stream Creation

   $js = $client->jetStream();
   $js->createStream('user_events', [
       'subjects' => ['user.>'],
       'retention' => 'limits',
       'max_messages' => 1000000,
   ]);
   

2. Durable Subscriptions

   $subscription = $client->subscribe('user.created', 'user_service');
   $subscription->setDurable(true); // Persists across restarts
   

3. Message Processing

   $subscription->onMessage(function ($msg) {
       try {
           // Process message
           $msg->ack(); // Acknowledge success
       } catch (\Exception $e) {
           $msg->nack(); // Negative acknowledgment
       }
   });
   

Error Handling

• Reconnection Logic

  $client->onReconnect(function () {
      Log::info('NATS reconnected');
  });
  

• Dead Letter Queues (DLQ)

  $js->createStream('user_events', [
      'deliver_subject' => 'user_events.dlq',
      'ack_policy' => 'all',
  ]);
  

Gotchas and Tips

Pitfalls

1. Connection Management

   • Issue: Dropped connections in long-running processes (e.g., Laravel queues).
   • Fix: Use Client::connect() with reconnect option or implement a health check loop.

     $client->connect(['reconnect' => true]);
     

2. Message Ordering

   • Issue: NATS does not guarantee message order by default.
   • Fix: Use JetStream with stream_name and sequence for ordered processing.

3. Memory Leaks

   • Issue: Unclosed subscriptions or subscriptions without onMessage handlers.
   • Fix: Always call $subscription->unsubscribe() when done or use context managers.

     $subscription->unsubscribe();
     

4. JetStream Quotas

   • Issue: Hitting storage limits on JetStream streams.
   • Fix: Monitor stream usage via NATS CLI or implement cleanup jobs.

     nats stream info user_events
     

Debugging Tips

1. Enable NATS Logging

   $client = new Client([
       'servers' => ['nats://localhost:4222'],
       'debug' => true, // Enable debug logs
   ]);
   

2. Inspect Messages

   $subscription->onMessage(function ($msg) {
       Log::debug('Received:', [
           'subject' => $msg->getSubject(),
           'data' => $msg->getData(),
           'headers' => $msg->getHeaders(),
       ]);
   });
   

3. Use NATS CLI Tools

   nats sub user.created --verbose
   nats pub user.created '{"test":1}'
   

Extension Points

1. Custom Middleware

   $subscription->onMessage(function ($msg) {
       // Pre-processing (e.g., validation)
       $data = json_decode($msg->getData());
       if (!valid($data)) {
           $msg->nack();
           return;
       }
       // Hand off to handler
   });
   

2. Laravel Service Provider

   // app/Providers/NatsServiceProvider.php
   public function register()
   {
       $this->app->singleton(Client::class, function ($app) {
           return new Client(config('nats.servers'));
       });
   }
   

3. Retry Logic

   $subscription->onMessage(function ($msg) {
       try {
           $this->process($msg);
           $msg->ack();
       } catch (\Exception $e) {
           if ($msg->getRedelivered()) {
               $msg->nack(); // Give up after retries
           } else {
               $msg->nack(); // Retry
           }
       }
   });
   

4. Metrics Integration

   $subscription->onMessage(function ($msg) use ($metrics) {
       $metrics->increment('nats.messages.processed');
       // Process message
   });