Channel Laravel Package

Getting Started

Minimal Setup

1. Installation

   composer require php-standard-library/channel
   

   Add to composer.json if using a monorepo or custom package management.

2. Basic Channel Creation

   use PhpStandardLibrary\Channel;
   
   $channel = new Channel(); // Unbuffered by default
   

3. First Use Case: Simple Producer-Consumer

   // Producer (runs in a background process or queue job)
   $channel->send('Hello, Channel!');
   
   // Consumer (runs elsewhere)
   $message = $channel->receive();
   echo $message; // Outputs: "Hello, Channel!"
   

4. Where to Look First

   • Documentation: Check the README.md for basic examples.
   • Source Code: Focus on src/Channel.php for core logic.
   • Tests: Review tests/ for edge cases (e.g., ChannelTest.php).

Implementation Patterns

Core Workflows

1. Buffered vs. Unbuffered Channels

• Unbuffered (Default): Blocks sender until receiver is ready.

  $channel = new Channel(); // No buffer
  

• Buffered: Accepts N messages without blocking.

  $channel = new Channel(5); // Buffer size = 5
  

2. Async Communication with Queues

• Producer (Queue Job)

  use Illuminate\Bus\Queueable;
  use Illuminate\Contracts\Queue\ShouldQueue;
  
  class SendMessageJob implements ShouldQueue
  {
      use Queueable;
  
      public function handle()
      {
          $channel = app(Channel::class);
          $channel->send('Async message');
      }
  }
  

• Consumer (Queue Worker)

  $channel = app(Channel::class);
  $message = $channel->receive(); // Blocks until message arrives
  

3. Selective Receives (Non-Blocking)

$channel = new Channel();
$timeout = 2.0; // Seconds

if ($channel->select([$channel], $timeout)) {
    $message = $channel->receive();
    // Handle message
} else {
    // Timeout or other channel ready
}

4. Closing Channels Gracefully

$channel->close(); // Prevents further sends/receives
if ($channel->isClosed()) {
    // Cleanup logic
}

Integration Tips

Dependency Injection (Laravel)

// config/app.php
'bindings' => [
    Channel::class => function ($app) {
        return new Channel(10); // Singleton with buffer
    },
];

Event-Driven Architecture

• Use channels to decouple services:

  // Service A
  $channel = app('order.processed');
  $channel->send($orderId);
  
  // Service B (listens in a separate process)
  $channel = app('order.processed');
  $orderId = $channel->receive();
  // Process order...
  

Testing

• Mock channels in unit tests:

  $mockChannel = Mockery::mock(Channel::class);
  $mockChannel->shouldReceive('receive')->once()->andReturn('test');
  

Gotchas and Tips

Pitfalls

1. Deadlocks in Unbuffered Channels

• Issue: If sender and receiver aren’t synchronized, both may block indefinitely.
• Fix: Use buffered channels or select() for timeouts.

  $channel = new Channel(1); // Small buffer to prevent starvation
  

2. Memory Leaks with Buffered Channels

• Issue: Large buffers can consume memory if messages aren’t consumed.
• Fix: Monitor buffer size and implement backpressure (e.g., reject sends if buffer is full).

  if ($channel->isFull()) {
      throw new \RuntimeException('Channel buffer full');
  }
  

3. Race Conditions in Multi-Process Environments

• Issue: Channels aren’t thread-safe by default (PHP’s GIL limits this, but async workers may still cause issues).
• Fix: Use a single process per channel or implement external synchronization (e.g., Redis).

4. Serialization Quirks

• Issue: Custom objects may not serialize/deserialize correctly.
• Fix: Implement __serialize()/__unserialize() or use json_encode()/json_decode() explicitly.

  $channel->send(json_encode($complexObject));
  $data = $channel->receive();
  $object = json_decode($data, true);
  

Debugging Tips

1. Log Channel State

\Log::debug('Channel buffer size:', ['size' => $channel->bufferSize()]);

2. Timeout Handling

• Always use timeouts in production:

  $message = $channel->receive(5.0); // 5-second timeout
  if ($message === null) {
      \Log::warning('Channel receive timed out');
  }
  

3. Check for Closed Channels

if ($channel->isClosed()) {
    throw new \RuntimeException('Channel is closed');
}

Extension Points

1. Custom Channel Implementations

• Extend Channel for domain-specific logic:

  class OrderedChannel extends Channel
  {
      public function sendPrioritized($message, int $priority)
      {
          // Custom logic
      }
  }
  

2. Channel Middleware

• Intercept sends/receives:

  $channel = new Channel();
  $channel->onSend(function ($message) {
      \Log::info('Message sent:', ['message' => $message]);
  });
  

3. Persistent Channels (Using Files/Redis)

• Store channel state externally:

  $channel = new FileChannel('/tmp/channel.dat');
  

• Note: Requires custom implementation (not in core package).

Config Quirks

1. Default Buffer Size

• Unbuffered channels (new Channel()) have a buffer size of 0. Ensure this matches your use case.

2. Resource Limits

• Large buffers may hit PHP’s memory_limit. Monitor with:

  $bufferSize = $channel->bufferSize();
  if ($bufferSize > 1000) {
      \Log::warning('Large channel buffer detected');
  }