README

Stream-like file processing for PHP, inspired by Gulp.

Pulp moves File objects through a chain of handlers. Source handlers create files, transform handlers edit or replace them, branching handlers fan work out, and destination/result handlers write or collect the final output.

Features

Pipeline API: Compose file processing as Pulp::start()->pipe(...)->run().
Virtual files: Handlers pass OpenMapsight\pulp\File objects with fileName, srcFileName, and dynamic metadata.
Lazy file content: Files from disk are not read until $file->content is accessed.
Stream access: Large-file handlers can call $file->stream() to read without loading full content.
Branching: Use split, merge, shadow, and fileSwitch for common flow-control patterns.
Extensible: Implement handlers directly or build package-level helpers around them.

Quick Start

use OpenMapsight\Pulp;
use OpenMapsight\pulp\File;

Pulp::start()
    ->pipe(Pulp::src('.*\.txt', __DIR__ . '/input'))
    ->pipe(Pulp::map(static function (File $file): File {
        $file->content = strtoupper($file->content);

        return $file;
    }))
    ->pipe(Pulp::dest(__DIR__ . '/output'))
    ->run();

Core Concepts

Pipelines

A pipeline is a chain of handlers:

Pulp::start()
    ->pipe(Pulp::src('.*\.json', __DIR__ . '/data'))
    ->pipe(/* handler */)
    ->pipe(Pulp::dest(__DIR__ . '/result'))
    ->run();

run() starts the chain. If no source handler is present, you can pass files directly:

$file = new File('example.txt');
$file->content = 'Hello';

$results = Pulp::start()
    ->pipe(Pulp::map(static fn(File $file): File => $file))
    ->run($file);

Files

OpenMapsight\pulp\File represents a virtual file.

$file->fileName: the pipeline-relative file name.
$file->srcFileName: the original source name/path.
$file->content: dynamic content property. For files sourced from disk, this lazy-loads the full file on first access.
$file->stream(): returns a readable stream. For path-backed files this opens the source file directly; for generated string content it creates a temporary stream.
Additional properties can be attached dynamically, for example $file->stats or $file->isLogicallyEmpty.

Use $file->content for normal transforms. Use $file->stream() in handlers that need to process large files without loading them completely.

API

Pulp::start()

Creates an empty pipeline.

$pulp = Pulp::start();

Pulp::src($patterns, $directory = '.')

Creates files from matching paths.

$patterns: a string, array of strings, file path, or File.
$directory: base directory for recursive matching.

Patterns are regular expressions matched against relative file names.

Pulp::start()
    ->pipe(Pulp::src('.*\.csv', __DIR__ . '/data'))
    ->run();

Pulp::srcFile($fileName, $aliasFileName = null, array $options = [])

Creates one file from a path. If $aliasFileName is provided, it becomes the virtual fileName.

Pulp::srcHttp($method, $uri, array $guzzleOptions, $aliasFileName, array $options = [])

Creates one file from an HTTP response body.

Pulp::start()
    ->pipe(Pulp::srcHttp(
        method: 'GET',
        uri: 'https://example.com/large.json',
        guzzleOptions: ['timeout' => 180],
        aliasFileName: 'large.json',
        options: ['sink' => true, 'successStatuses' => [200, 304, 204]],
    ))
    ->run();

sink => true writes the body to a temp file and emits a path-backed File, so a ~100 MB response does not have to live as a PHP string. The temp file is removed when the last File that shares it is released, including clones from split, shadow, and results. Pass a path string to keep the body at a path you manage.
client injects a Guzzle Client (tests use a mock handler).
successStatuses throws unless the status is in the list. Pair with Guzzle http_errors => false when 304/204 are first-class.
The file also carries httpStatus, httpLastModified, httpEtag, and httpType.

Pulp::ensureDirectory($directory, $permissions = 0775)

Create $directory and its parents if needed, then return the path. Jobs use this for pulp/cache before writing sidecar files; dest uses the same helper internally.

$cacheDir = Pulp::ensureDirectory(__DIR__ . '/cache');

Pulp::dest($directory, array $options = [])

Writes incoming files into $directory using each file's fileName.

->pipe(Pulp::dest(__DIR__ . '/result'))

Common options:

flush: call fflush() after writing.
skipExceptions: log write errors and continue.
logSkipExceptions: stderr, stdout, or false.

Pulp::map($callback)

Transforms or drops files in plain PHP: one file in, at most one file out. For work that needs other handlers per file, use Pulp::each.

->pipe(Pulp::map(static function (File $file): ?File {
    if ($file->fileName === 'skip.txt') {
        return null;
    }

    $file->content .= "\n";

    return $file;
}))

Pulp::filter($callback)

Keeps only files where the callback returns truthy.

->pipe(Pulp::filter(static fn(File $file): bool => str_ends_with($file->fileName, '.json')))

Pulp::results($callback)

Collects all files that reach this handler and calls the callback at the end.

->pipe(Pulp::results(static function (array $files): void {
    // inspect results
}))

Pulp::split(...$pipelines)

Feeds the same incoming files into multiple branch pipelines and merges each branch's results back into the main pipeline.

Use this when one source should produce several outputs.

Pulp::start()
    ->pipe(Pulp::src('.*\.txt', __DIR__ . '/input'))
    ->pipe(Pulp::split(
        static fn(Pulp $p): Pulp => $p->pipe(Pulp::map(static function (File $file): File {
            $file->fileName = 'upper-' . $file->fileName;
            $file->content = strtoupper($file->content);

            return $file;
        })),
        static fn(Pulp $p): Pulp => $p->pipe(Pulp::map(static function (File $file): File {
            $file->fileName = 'lower-' . $file->fileName;
            $file->content = strtolower($file->content);

            return $file;
        })),
    ))
    ->pipe(Pulp::dest(__DIR__ . '/result'))
    ->run();

Branches may be Pulp instances or callbacks receiving a new Pulp instance.

Pulp::merge(...$pulps)

Runs independent pipelines and merges their output.

$a = Pulp::start()->pipe(Pulp::src('.*\.json', __DIR__ . '/a'));
$b = Pulp::start()->pipe(Pulp::src('.*\.json', __DIR__ . '/b'));

Pulp::start()
    ->pipe(Pulp::merge($a, $b))
    ->pipe(Pulp::dest(__DIR__ . '/result'))
    ->run();

Use merge for independent sources. Use split when you already have one incoming stream and want multiple outputs from it.

Pulp::shadow($callback)

Taps the stream into a side pipeline without changing the main stream.

->pipe(Pulp::shadow(static fn(Pulp $p): Pulp => $p
    ->pipe(Pulp::debug())
))

shadow is useful for diagnostics or side effects. Its branch results are not merged back. Use split if branch output should continue downstream.

Pulp::fileSwitch(array $patterns, ?callable $defaultCb = null)

Routes files into different sub-pipelines by file name.

->pipe(Pulp::fileSwitch([
    '.*\.json' => static fn(Pulp $p): Pulp => $p->pipe(/* JSON handlers */),
    '.*\.xml' => static fn(Pulp $p): Pulp => $p->pipe(/* XML handlers */),
], static fn(Pulp $p): Pulp => $p))

Pulp::each($callback, array $options = [])

Runs a pipeline for each incoming file and pushes its results instead of the file. The callback receives the file and returns the pipeline, which runs without input files. Use it when the work depends on the file, for example fetching a URL it names:

Pulp::start()
    ->pipe(Pulp::src('.*\.url', __DIR__ . '/feeds'))
    ->pipe(Pulp::each(
        static fn(File $feed): Pulp => Pulp::start()
            ->pipe(Pulp::srcHttp(
                method: 'GET',
                uri: trim($feed->content),
                guzzleOptions: ['timeout' => 30],
                aliasFileName: basename($feed->fileName, '.url') . '.json',
            )),
        ['skipExceptions' => true],
    ))
    ->pipe(Pulp::dest(__DIR__ . '/result'))
    ->run();

Start the pipeline with Pulp::src($file) to work on the file itself. Wrap a source in PulpCache::remember() inside the callback to keep the last good copy per file.

each or map?

map changes a file with plain PHP and returns it, another file, or null. It is cheap and runs inline.
each builds and runs a whole pipeline per file. Use it when that work needs handlers (srcHttp, PulpCache::remember, package handlers), may emit several files, or should fail per file (skipExceptions). map has no skipExceptions; an exception stops the run.
The pipeline of each starts empty. Without a source handler or Pulp::src($file) it emits nothing.
Handlers that collect files (results, PulpGeoJSON::merge, …) inside each see one file's results only. Put them after each to collect across files.

Options:

skipExceptions: log a failing pipeline and continue with the next file.
logSkipExceptions: stderr, stdout, or false.

Pulp::debug($length = 30)

Prints a short preview for each file.

Pulp::delete()

Deletes each file's srcFileName from disk.

Pulp::utf8encode() / Pulp::utf8decode()

Converts string content between ISO-8859-1 and UTF-8.

Pulp::serveHttp()

Sends file content as an HTTP response.

Common Patterns

Read, Transform, Write

Pulp::start()
    ->pipe(Pulp::src('.*\.txt', __DIR__ . '/input'))
    ->pipe(Pulp::map(static function (File $file): File {
        $file->content = trim($file->content) . "\n";

        return $file;
    }))
    ->pipe(Pulp::dest(__DIR__ . '/output'))
    ->run();

Fan Out One Source Into Multiple Outputs

Pulp::start()
    ->pipe(Pulp::src('.*\.txt', __DIR__ . '/input'))
    ->pipe(Pulp::split(
        static fn(Pulp $p): Pulp => $p->pipe(/* branch A */),
        static fn(Pulp $p): Pulp => $p->pipe(/* branch B */),
    ))
    ->pipe(Pulp::dest(__DIR__ . '/output'))
    ->run();

Choose a Per-File Handler

Handler	The callback gets	It returns	Files out per file	Use it for
`map`	the file	the file, another file, or `null`	0 or 1	changing content, name or properties in plain PHP
`filter`	the file	`true` to keep it	0 or 1	dropping files
`each`	the file	a pipeline, run without input	any	work that needs handlers per file, such as fetching what a file names
`fileSwitch`	a pipeline with the file as input	the pipeline	any	different handlers by file name
`split`	a pipeline with every file as input	the pipeline	any, per branch	several outputs from the same files

Prefer map and filter when plain PHP does the job. Reach for each when the work depends on the file and needs other handlers.

Process Large Files With Streams

->pipe(Pulp::map(static function (File $file): File {
    $stream = $file->stream();

    try {
        while (($line = fgets($stream)) !== false) {
            // Process without loading the full file.
        }
    } finally {
        fclose($stream);
    }

    return $file;
}))

Writing Custom Handlers

Most handlers extend OpenMapsight\pulp\AbstractHandler.

use OpenMapsight\pulp\AbstractHandler;
use OpenMapsight\pulp\File;

class UppercaseHandler extends AbstractHandler
{
    public function onFile(File $file): void
    {
        $file->content = strtoupper($file->content);
        $this->pushFile($file);
    }
}

Handlers can override:

onStart(): called before the first file.
onFile(File $file): called for each file.
onEnd(): called when the stream ends.

Use $this->pushFile($file) to pass output to the next handler. A handler may emit zero, one, or many files.

Constructor arguments are declared with getConstructorParamDefs() and accessed via $this->cp.

class PrefixHandler extends AbstractHandler
{
    protected function getConstructorParamDefs(): array
    {
        return ['prefix'];
    }

    public function onFile(File $file): void
    {
        $file->content = $this->cp->prefix . $file->content;
        $this->pushFile($file);
    }
}

mapsight / pulp

Package info

Security

Statistics