diff --git a/CHANGELOG.md b/CHANGELOG.md
index a80b2ddf..2cb4ce3c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,7 @@ All Notable changes to `Csv` will be documented in this file
### Added
- Adding the `TabularDataReader::map` method.
+- Adding `CallbackStreamFilter` class
### Deprecated
diff --git a/docs/9.0/interoperability/callback-stream-filter.md b/docs/9.0/interoperability/callback-stream-filter.md
new file mode 100644
index 00000000..8ab4b242
--- /dev/null
+++ b/docs/9.0/interoperability/callback-stream-filter.md
@@ -0,0 +1,52 @@
+---
+layout: default
+title: Dynamic Stream Filter
+---
+
+# Callback Stream Filter
+
+Available since version 9.22.0
+
+Sometimes you may encounter a scenario where you need to create a specific stream filter
+to resolve a specific issue. Instead of having to put up with the hassle of creating a
+fully fledge stream filter, we are introducing a `CallbackStreamFilter`. This filter
+is a PHP stream filter which enables applying a callable onto the stream prior to it
+being actively consumed by the CSV process.
+
+## Usage with CSV objects
+
+Out of the box, the filter can not work, it requires a unique name and a callback to be usable.
+Once registered you can re-use the filter with CSV documents or with a resource.
+
+let's imagine we have a CSV document with the return carrier character as the end of line character.
+This type of document is parsable by the package but only if you enable the deprecated `auto_detect_line_endings`.
+
+If you no longer want to rely on that feature since it emits a deprecation warning you can use the new
+`CallbackStreamFilter` instead by swaping the offending character with a modern alternative.
+
+```php
+use League\Csv\CallbackStreamFilter;
+use League\Csv\Reader;
+
+$csv = "title1,title2,title3\rcontent11,content12,content13\rcontent21,content22,content23\r";
+
+$document = Reader::createFromString($csv);
+CallbackStreamFilter::addTo(
+ $document,
+ 'swap.carrier.return',
+ fn (string $bucket): string => str_replace("\r", "\n", $bucket)
+);
+$document->setHeaderOffset(0);
+return $document->first();
+// returns ['title1' => 'content11', 'title2' => 'content12', 'title3' => 'content13']
+```
+
+The `addTo` method register the filter with the unique `swap.carrier.return` name and then attach
+it to the CSV document object on read.
+
+On read, the CSV document content is never changed or replaced.
+Conversely, the changes are persisted during writing.
+
+Of course the `CallbackStreamFilter` can be use in other different scenario or with PHP stream resources.
+
+
diff --git a/docs/_data/menu.yml b/docs/_data/menu.yml
index 30454db3..56736732 100644
--- a/docs/_data/menu.yml
+++ b/docs/_data/menu.yml
@@ -27,6 +27,7 @@ version:
Force Enclosure : '/9.0/interoperability/enclose-field/'
Handling Delimiter : '/9.0/interoperability/swap-delimiter/'
Formula Injection : '/9.0/interoperability/escape-formula-injection/'
+ Callback Stream Filter : '/9.0/interoperability/callback-stream-filter/'
Converting Records:
Overview: '/9.0/converter/'
Charset Converter: '/9.0/converter/charset/'
diff --git a/src/CallbackStreamFilter.php b/src/CallbackStreamFilter.php
new file mode 100644
index 00000000..af8750d0
--- /dev/null
+++ b/src/CallbackStreamFilter.php
@@ -0,0 +1,140 @@
+
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace League\Csv;
+
+use Closure;
+use php_user_filter;
+use RuntimeException;
+use TypeError;
+
+use function array_key_exists;
+use function is_resource;
+
+final class CallbackStreamFilter extends php_user_filter
+{
+ private const FILTER_NAME = 'string.league.csv.stream.callback.filter';
+
+ public static function getFiltername(string $name): string
+ {
+ return self::FILTER_NAME.'.'.$name;
+ }
+
+ /**
+ * Static method to register the class as a stream filter.
+ */
+ public static function register(string $name): void
+ {
+ $filtername = self::getFiltername($name);
+ if (!in_array($filtername, stream_get_filters(), true)) {
+ stream_filter_register($filtername, self::class);
+ }
+ }
+
+ /**
+ * Static method to attach the stream filter to a CSV Reader or Writer instance.
+ */
+ public static function addTo(AbstractCsv $csv, string $name, callable $callback): void
+ {
+ self::register($name);
+
+ $csv->addStreamFilter(self::getFiltername($name), [
+ 'name' => $name,
+ 'callback' => $callback instanceof Closure ? $callback : $callback(...),
+ ]);
+ }
+
+ /**
+ * @param resource $stream
+ * @param callable(string): string $callback
+ *
+ * @throws TypeError
+ * @throws RuntimeException
+ *
+ * @return resource
+ */
+ public static function appendTo(mixed $stream, string $name, callable $callback): mixed
+ {
+ self::register($name);
+
+ is_resource($stream) || throw new TypeError('Argument passed must be a stream resource, '.gettype($stream).' given.');
+ 'stream' === ($type = get_resource_type($stream)) || throw new TypeError('Argument passed must be a stream resource, '.$type.' resource given');
+
+ set_error_handler(fn (int $errno, string $errstr, string $errfile, int $errline) => true);
+ $filter = stream_filter_append($stream, self::getFiltername($name), params: [
+ 'name' => $name,
+ 'callback' => $callback instanceof Closure ? $callback : $callback(...),
+ ]);
+ restore_error_handler();
+
+ if (!is_resource($filter)) {
+ throw new RuntimeException('Could not append the registered stream filter: '.self::getFiltername($name));
+ }
+
+ return $filter;
+ }
+
+ /**
+ * @param resource $stream
+ * @param callable(string): string $callback
+ *
+ * @throws TypeError
+ * @throws RuntimeException
+ *
+ * @return resource
+ */
+ public static function prependTo(mixed $stream, string $name, callable $callback): mixed
+ {
+ self::register($name);
+
+ is_resource($stream) || throw new TypeError('Argument passed must be a stream resource, '.gettype($stream).' given.');
+ 'stream' === ($type = get_resource_type($stream)) || throw new TypeError('Argument passed must be a stream resource, '.$type.' resource given');
+
+ $filtername = self::getFiltername($name);
+ set_error_handler(fn (int $errno, string $errstr, string $errfile, int $errline) => true);
+ $filter = stream_filter_append($stream, $filtername, params: [
+ 'name' => $name,
+ 'callback' => $callback instanceof Closure ? $callback : $callback(...),
+ ]);
+ restore_error_handler();
+
+ if (!is_resource($filter)) {
+ throw new RuntimeException('Could not append the registered stream filter: '.self::getFiltername($name));
+ }
+
+ return $filter;
+ }
+
+ public function onCreate(): bool
+ {
+ return is_array($this->params) &&
+ array_key_exists('name', $this->params) &&
+ self::getFiltername($this->params['name']) === $this->filtername &&
+ array_key_exists('callback', $this->params) &&
+ $this->params['callback'] instanceof Closure
+ ;
+ }
+
+ public function filter($in, $out, &$consumed, bool $closing): int
+ {
+ /** @var Closure(string): string $callback */
+ $callback = $this->params['callback']; /* @phpstan-ignore-line */
+ while (null !== ($bucket = stream_bucket_make_writeable($in))) {
+ $bucket->data = ($callback)($bucket->data);
+ $consumed += $bucket->datalen;
+ stream_bucket_append($out, $bucket);
+ }
+
+ return PSFS_PASS_ON;
+ }
+}
diff --git a/src/CallbackStreamFilterTest.php b/src/CallbackStreamFilterTest.php
new file mode 100644
index 00000000..c25b23f3
--- /dev/null
+++ b/src/CallbackStreamFilterTest.php
@@ -0,0 +1,65 @@
+
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace League\Csv;
+
+use PHPUnit\Framework\Attributes\Test;
+use PHPUnit\Framework\TestCase;
+
+use function str_replace;
+
+final class CallbackStreamFilterTest extends TestCase
+{
+ #[Test]
+ public function it_can_swap_the_delimiter_on_read(): void
+ {
+ $document = <<setDelimiter("\x02");
+ CallbackStreamFilter::addTo(
+ $reader,
+ 'swap.delemiter.in',
+ fn (string $bucket): string => str_replace('💩', "\x02", $bucket)
+ );
+ $reader->setHeaderOffset(0);
+
+ self::assertSame(
+ ['observedOn' => '2023-10-01', 'temperature' => '18', 'place' => 'Yamoussokro'],
+ $reader->first()
+ );
+ }
+
+ #[Test]
+ public function it_can_swap_the_delimiter_on_write(): void
+ {
+ $writer = Writer::createFromString();
+ $writer->setDelimiter("\x02");
+ CallbackStreamFilter::addTo(
+ $writer,
+ 'swap.delemiter.out',
+ fn (string $bucket): string => str_replace("\x02", '💩', $bucket)
+ );
+
+ $writer->insertOne(['observeedOn' => '2023-10-01', 'temperature' => '18', 'place' => 'Yamoussokro']);
+ self::assertSame('2023-10-01💩18💩Yamoussokro'."\n", $writer->toString());
+ }
+}