Pulp

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. Pass a path string to choose the file.
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.

->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::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();

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);
 }
}

Name		Name	Last commit message	Last commit date
src		src
test		test
CHANGELOG.md		CHANGELOG.md
Pulp.php		Pulp.php
README.md		README.md
composer.json		composer.json
phpunit.xml		phpunit.xml

Uh oh!

Latest commit

History

Folders and files

Repository files navigation