Mq React Laravel Package

Getting Started

• Install with Composer: composer require clue/stream-filter. There is no Laravel service provider, facade, config file, or migration.
• Start with the package README and the namespaced functions: Clue\StreamFilter\append(), Clue\StreamFilter\prepend(), and Clue\StreamFilter\remove().
• First practical Laravel use case: stream a large CSV/log file from Storage and transform or redact it chunk-by-chunk before download or re-storage.

use function Clue\StreamFilter\append;

$stream = Storage::disk('local')->readStream('exports/customers.csv');

$filter = append($stream, function (string $chunk): string {
    // Strip UTF-8 BOM only if present at start, normalize CRLF, etc.
    return str_replace("\r\n", "\n", $chunk);
});

return response()->streamDownload(function () use ($stream) {
    try {
        fpassthru($stream);
    } finally {
        fclose($stream);
    }
}, 'customers.csv');

• v1.7.0 is primarily a quality/static-analysis release: improved template annotations for typed callback arguments, documentation/examples, and PHP 8.3/8.4 test coverage. No Laravel breaking changes are indicated.

Implementation Patterns

• Use it around Laravel storage streams rather than loading whole files:
  • Storage::disk('s3')->readStream($path) → append filter → Storage::disk('local')->writeStream($target, $filteredStream).
  • UploadedFile::getRealPath() → fopen($path, 'rb') → append sanitizer/redactor → stream response.
• Common day-to-day filters:
  • Strip BOMs, normalize line endings, redact secrets/tokens, replace legacy terms, normalize encoding.
  • Compose multiple filters in order; order matters because each filter receives the previous filter’s output.
• For line-oriented processing, keep a small stateful buffer in the callback. Treat data as arbitrary chunks, not complete lines:

$buffer = '';

$filter = append($stream, function (string $chunk) use (&$buffer): string {
    $buffer .= $chunk;
    $lines = explode("\n", $buffer);
    $buffer = array_pop($lines);

    return implode("\n", array_map(
        fn (string $line) => preg_replace('/secret=\S+/', 'secret=[redacted]', $line),
        $lines
    )) . "\n";
});

• Wrap repeated filter logic in small application services/classes, e.g. CsvSanitizer, LogRedactor, StreamFilterPipeline, so jobs and controllers do not duplicate callbacks.
• In queued jobs, process streams incrementally and close handles in finally. Use Storage::fake() and php://temp for tests.
• Use Laravel’s normal streamed responses (response()->streamDownload()) or Flysystem stream APIs; the package is not a replacement for League\Csv or Flysystem, it is a low-level filter layer.

Gotchas and Tips

• Stream filters operate on chunks, not lines or whole files. Do not assume a callback receives a complete CSV row, a complete UTF-8 character, or a full JSON document.
• Attach filters with the correct read/write mode and at the right time. If you write to a temp stream first, rewind before attaching a read filter; if you want to filter writes, attach before writing.
• Filters are attached to the stream resource and live until removed or the stream is closed. In long-running queues and Laravel Octane, remove filters or close streams in finally to avoid leaks.
• stream_get_contents() or $file->getContent() defeats the purpose. Consume incrementally with fread(), fpassthru(), or storage stream writes.
• For binary or multibyte data, avoid naive strlen/substr/preg_* assumptions across chunk boundaries. Buffer or use PHP’s built-in conversion filters where appropriate.
• v1.7.0’s improved template types help PHPStan/Psalm/Larastan infer typed callbacks, but only if your callbacks have explicit parameter/return types. Runtime behavior is unchanged.
• Safe upgrade path: run composer update clue/stream-filter, then your static analysis and stream-related tests. No migration or config change is expected.
• Extension point: prefer custom callback filters plus a small service class over forking the package. For reusable behavior, combine callback filters with built-in PHP stream filters such as encoding/compression filters.