Guzzlestreams Laravel Package

Getting Started

Minimal Setup

1. Installation

   composer require ezimuel/guzzlestreams
   

   • No additional configuration is required beyond requiring the package.

2. First Use Case: Streaming Elasticsearch Responses

   use Ezimuel\GuzzleStreams\Stream;
   use GuzzleHttp\Client;
   
   $client = new Client();
   $response = $client->request('GET', 'https://your-elasticsearch-endpoint/_search');
   
   // Stream the response body
   $stream = Stream::fromResource($response->getBody());
   $decoder = new \Elasticsearch\Serializers\SmartSerializer();
   
   foreach ($stream as $chunk) {
       $data = $decoder->decode($chunk, 'json');
       // Process each chunk of data
   }
   

3. Where to Look First

   • Source Code: Focus on Stream.php for core functionality.
   • Guzzle Integration: Review how GuzzleHttp\Psr7\Stream is extended in the package.
   • Elasticsearch PHP Client: Check elasticsearch/elasticsearch for serialization/deserialization patterns.

Implementation Patterns

1. Streaming Large Responses

• Use Case: Fetch large Elasticsearch query results without loading everything into memory.
• Pattern:

  $response = $client->request('GET', 'https://your-elasticsearch-endpoint/_search', [
      'query' => 'your_query',
      'size' => 1000, // Batch size
  ]);
  
  $stream = Stream::fromResource($response->getBody());
  $decoder = new \Elasticsearch\Serializers\SmartSerializer();
  
  foreach ($stream as $chunk) {
      $batch = $decoder->decode($chunk, 'json');
      processBatch($batch); // Process in chunks
  }
  

2. Custom Stream Wrapping

• Use Case: Extend or modify stream behavior (e.g., filtering, transformation).
• Pattern:

  use Ezimuel\GuzzleStreams\Stream;
  
  $stream = Stream::fromResource($response->getBody())
      ->filter(function ($chunk) {
          // Example: Filter out specific fields
          $data = json_decode($chunk, true);
          unset($data['unwanted_field']);
          return json_encode($data);
      });
  

3. Integration with Elasticsearch PHP Client

• Use Case: Replace default response handling in elasticsearch/elasticsearch with streaming.
• Pattern:

  $client = Elastic\Elasticsearch\ClientBuilder::create()
      ->setHosts(['https://your-elasticsearch-endpoint'])
      ->setHandlerStack(GuzzleHttp\HandlerStack::create())
      ->setHandler(new \GuzzleHttp\Handler\CurlHandler())
      ->build();
  
  // Override default response handling
  $response = $client->search([
      'index' => 'your_index',
      'body' => ['query' => ['match_all' => new \stdClass()]],
  ]);
  
  $stream = Stream::fromResource($response->getBody());
  // Process stream as shown above
  

4. Error Handling and Retries

• Use Case: Handle stream interruptions (e.g., network issues) with retries.
• Pattern:

  $client = new Client(['timeout' => 30]);
  $retries = 3;
  
  while ($retries--) {
      try {
          $response = $client->request('GET', 'https://your-elasticsearch-endpoint/_search');
          $stream = Stream::fromResource($response->getBody());
          foreach ($stream as $chunk) {
              // Process chunk
          }
          break; // Success
      } catch (\Exception $e) {
          if ($retries === 0) throw $e;
          sleep(1); // Backoff
      }
  }
  

Gotchas and Tips

1. Resource Management

• Gotcha: Forgetting to close streams can lead to memory leaks or connection hangs.
• Tip: Use Stream::close() explicitly or rely on PHP's __destruct():

  $stream = Stream::fromResource($response->getBody());
  // ... processing ...
  $stream->close(); // Always close when done
  

• Alternative: Wrap streams in a try-finally block:

  try {
      $stream = Stream::fromResource($response->getBody());
      foreach ($stream as $chunk) { /* ... */ }
  } finally {
      $stream->close();
  }
  

2. Chunk Size and Performance

• Gotcha: Elasticsearch may split responses into chunks arbitrarily, leading to malformed JSON if not handled carefully.
• Tip: Use a JSON stream parser like spatie/fork or symfony/stream-writer to reconstruct JSON objects:

  use Spatie\Fork\Json\JsonStreamer;
  
  $streamer = new JsonStreamer();
  $streamer->on('data', function ($data) {
      // Process each JSON object
  });
  
  foreach ($stream as $chunk) {
      $streamer->feed($chunk);
  }
  

3. Guzzle Version Compatibility

• Gotcha: The package is a fork of guzzle/streams (abandoned) and may not support the latest Guzzle versions.
• Tip: Pin Guzzle to a compatible version in composer.json:

  "require": {
      "guzzlehttp/guzzle": "~6.0 || ~7.0" // Check package docs for exact version
  }
  

• Debugging: If streams fail, check Guzzle's Psr7 compatibility:

  if (!method_exists($response->getBody(), 'isSeekable')) {
      throw new \RuntimeException('Incompatible Guzzle version');
  }
  

4. Thread Safety

• Gotcha: Streams are not thread-safe. Concurrent access can corrupt data.
• Tip: Use a single stream per request and process chunks sequentially:

  $stream = Stream::fromResource($response->getBody());
  $chunks = iterator_to_array($stream); // Consume all chunks in one thread
  

5. Extension Points

• Custom Stream Classes: Extend Ezimuel\GuzzleStreams\Stream to add logic:

  class LoggingStream extends Stream {
      public function read($length) {
          $data = parent::read($length);
          \Log::debug('Read chunk:', ['length' => $length, 'data' => $data]);
          return $data;
      }
  }
  

• Event Listeners: Attach listeners to stream events (e.g., onRead, onClose) via traits or decorators.

6. Debugging Streams

• Tip: Dump raw chunks to debug issues:

  foreach ($stream as $chunk) {
      \Log::debug('Raw chunk:', ['chunk' => substr($chunk, 0, 100) . '...']);
      // Process chunk
  }
  

• Common Issues:
  • Truncated JSON: Ensure Elasticsearch's slices or search_after is used for pagination.
  • Connection Timeouts: Increase Guzzle's timeout or use connect_timeout.

7. Configuration Quirks

• Tip: If using with elasticsearch/elasticsearch, disable default response buffering:

  $client->getEngine()->getClient()->getConfig()->set('buffer_response', false);