Dataflow Bundle Laravel Package

Getting Started

Read this first: code-rhapsodie/dataflow-bundle is a Symfony bundle, not a Laravel package. There is no service provider, no config/dataflow.php, no Artisan command, and no illuminate/* support. composer require code-rhapsodie/dataflow-bundle in a Laravel app gets you classes on disk and nothing wired up — no tagged dataflow types, no runner, no commands. So the day-to-day leverage takes one of two shapes:

1. Sidecar worker (recommended): a small Symfony 7.x app (current line is 6.x, e.g. v6.0.4; branch 5.x requires Symfony ^7.3, branch 4.x for Symfony ≤6) that owns the dataflow definitions and cron, sharing your Laravel database. Laravel stays the product; dataflow is an ops process.
2. Concept port: you keep the pipeline shape (Reader → N Steps → Writers, job + schedule tables, queue delegation) and implement it on top of Laravel's container, Eloquent and queues. Read DataflowType/DataflowBuilder, DataflowType/AbstractDataflowType, DataflowType/Writer/WriterInterface and MessengerMode/JobMessage — that's the whole architecture in four files.

Minimal sidecar setup:

composer require code-rhapsodie/dataflow-bundle:^6.0.4
# config/bundles.php
# CodeRhapsodie\DataflowBundle\CodeRhapsodieDataflowBundle::class => ['all' => true],
bin/console code-rhapsodie:dataflow:dump-schema        # paste into a Doctrine migration
bin/console debug:container --tag coderhapsodie.dataflow.type   # is my type registered?

First use case — CSV in, rows out, one writer:

class CsvImportType extends AbstractDataflowType
{
    public function __construct(private CsvReader $reader, private DbWriter $writer) {}

    protected function buildDataflow(DataflowBuilder $builder, array $options): void
    {
        $builder
            ->setReader($this->reader->read($options['from-file']))
            ->addStep(fn ($row) => [...$row, 'imported_at' => new \DateTimeImmutable()])
            ->addStep(fn ($row) => $row['email'] === '' ? false : $row)   // filter
            ->addWriter($this->writer);
    }

    protected function configureOptions(OptionsResolver $r): void
    {
        $r->setRequired('from-file');
    }

    public function getLabel(): string { return 'CSV import'; }
    public function getAliases(): iterable { return ['csv']; }
}

Then bin/console code-rhapsodie:dataflow:run-pending from cron every minute — that command is the heartbeat of the whole package, and everything else (schedule, enable/disable, list, last-job result) is configuration around it. Run bin/console list code-rhapsodie to enumerate the schedule-related commands, since the README lists them as features but not by name.

Implementation Patterns

• One DataflowType per import/export shape, options for the instance. Don't create ImportCustomersType, ImportProductsType, ImportOrdersType if the only difference is the source file and writer table. One type with from-file, connection, batch-size options, invoked multiple times with different option sets, keeps the CLI and schedule list readable.

• buildDataflow() is wiring, not logic. Inject readers/writers/steps as services and only compose them there. Anything you new inside buildDataflow() is untestable and invisible to the container, which defeats the point of the bundle.

• Readers must stream. Return a generator, never an array of the full file. setReader() accepts any iterable, and the runner consumes row by row:

  public function read(string $path): iterable {
      $fh = fopen($path, 'r');
      try { while (false !== ($line = fgets($fh))) { yield explode('|', trim($line)); } }
      finally { fclose($fh); }   // critical in long-running workers
  }
  

• Steps as the transformation contract: callable in, item or false out. Chain them small — normalize, validate, enrich, filter — so a failing step tells you where in the pipeline the row died.

• Writers implement prepare() / write($item) / finish(). Open handles in prepare(), buffer and flush for throughput in write(), close in finish(). To reuse a PortPHP writer, wrap it: $builder->addWriter(new PortWriterAdapter(new \Port\FileWriter())).

• Async steps carry a scale factor: addStep(function ($item): \Generator { yield new \Amp\Delayed(1000); ... }, 2). Use this for HTTP enrichment, not for ordering-sensitive work. Output order is only guaranteed when every step is scaled at 1.

• Scheduling: use enable/disable, not cron edits. The cr_dataflow_scheduled table plus the schedule/enable/disable commands mean ops can pause a nightly import without touching crontab. The cron only ever runs run-pending.

• Concurrency via Messenger mode (messenger_mode.enabled: true) when one run-pending tick has more jobs than it can finish before the next tick. Route JobMessage to an async transport and keep a worker running. In a Laravel port, this is exactly a queued job class fanned out to Horizon workers.

• Logging is your trace. AbstractDataflowType exposes $this->logger; inject it into readers/writers, and note that under run-pending the log is persisted onto the job row, so a Laravel dashboard can read cr_dataflow_job read-only via Eloquent without any coupling.

• Testing: unit-test reader generators, step closures and writers directly — they are plain PHP. Integration-test buildDataflow() with a stub DataflowBuilder that records the reader/steps/writers. You almost never need to boot the kernel to trust a type.

Gotchas and Tips

• No table creation on install. cr_dataflow_scheduled and cr_dataflow_job must exist before anything works. dump-schema emits create SQL; dump-schema --update emits alter SQL for an already-provisioned database. Forgetting this is the #1 first-run failure and it surfaces as a DBAL exception mid-job.
• false is the sentinel, not data. Readers may not yield false, and a step returning false silently drops the row. If your step legitimately maps to a falsy scalar, wrap it — ['value' => $x] — or you'll lose rows and never see an error.
• Exceptions are caught and logged, not propagated. A failed job can exit cleanly. Monitor the job table (last job result command / your own query), not the process exit code, or you'll discover failures weeks late.
• Required options explode at build time, before the first row is read. That's good design, but it reads as "the dataflow did nothing" in the job log — check the top of the log, not the row count.
• Messenger mode with no worker looks identical to a broken job. Jobs are dispatched, nothing consumes them. Confirm a worker is up before blaming the type.
• Amp API drift: the README's new \Amp\Delayed(...) is Amp v2. If you're on Amp v3 you need Amp\delay(). Check composer show amphp/amp before copying examples.
• Version matrix bites: 6.x is the current line and is API-identical to 5.x for everything described here — v6.0.4 is a patch containing only a monolog compatibility fix, a CI compatibility matrix and PHPStan cleanups, so there is nothing to re-learn and nothing to migrate. 5.x requires Symfony ^7.3; on Symfony 6.x you must pin the 4.x line. Don't let Composer float you into a major you can't run.
• Pin ^6.0.4 if you run monolog 3 in the sidecar. Earlier 6.0.x had a monolog incompatibility, and since the job log lives in the database and is your failure trace, a broken logger means silently blind imports rather than a loud crash.
• Auto-tagging only happens under Symfony autoconfigure. In a hand-wired service provider (i.e. any Laravel bridging attempt) you must apply the tag coderhapsodie.dataflow.type yourself, and you lose debug:container --tag as your registration sanity check.
• dbal_default_connection must be the connection that owns the cr_* tables. Multiple DBAL connections are supported, but each writer needs the right connection injected — the bundle's setting only chooses the schedule/job store.
• exceptions_mode: 'file' needs league/flysystem installed and a valid flysystem_service ID; a typo there turns a diagnostic feature into another exception.
• Watch log volume in run-pending mode. Every exception and custom message lands in the job row; a per-row log line in a 500k-row import will bloat that table.
• Bus factor is real: 18 stars, 0 dependents. The v6.0.4 release is a good sign (maintenance is alive, CI matrix added), but read the source before betting on it, pin exact versions, and keep your DataflowType classes thin so replacing the runner is a container-wiring change rather than a rewrite. The pipeline contract (iterable reader, callable steps, WriterInterface) is the part worth owning; the runner is the part worth forking.