diff --git a/packages/pipeline/.gitignore b/packages/pipeline/.gitignore
new file mode 100644
index 0000000..3cceda5
--- /dev/null
+++ b/packages/pipeline/.gitignore
@@ -0,0 +1,7 @@
+# https://dart.dev/guides/libraries/private-files
+# Created by `dart pub`
+.dart_tool/
+
+# Avoid committing pubspec.lock for library packages; see
+# https://dart.dev/guides/libraries/private-files#pubspeclock.
+pubspec.lock
diff --git a/packages/pipeline/CHANGELOG.md b/packages/pipeline/CHANGELOG.md
new file mode 100644
index 0000000..effe43c
--- /dev/null
+++ b/packages/pipeline/CHANGELOG.md
@@ -0,0 +1,3 @@
+## 1.0.0
+
+- Initial version.
diff --git a/packages/pipeline/LICENSE.md b/packages/pipeline/LICENSE.md
new file mode 100644
index 0000000..0fd0d03
--- /dev/null
+++ b/packages/pipeline/LICENSE.md
@@ -0,0 +1,10 @@
+The MIT License (MIT)
+
+The Laravel Framework is Copyright (c) Taylor Otwell
+The Fabric Framework is Copyright (c) Vieo, Inc.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
\ No newline at end of file
diff --git a/packages/pipeline/README.md b/packages/pipeline/README.md
new file mode 100644
index 0000000..586a156
--- /dev/null
+++ b/packages/pipeline/README.md
@@ -0,0 +1,380 @@
+
+
+# Platform Pipeline
+
+A Laravel-compatible pipeline implementation in Dart, providing a robust way to pass objects through a series of operations.
+
+[]()
+[]()
+
+## Table of Contents
+
+- [Overview](#overview)
+- [Features](#features)
+- [Requirements](#requirements)
+- [Installation](#installation)
+- [Usage](#usage)
+ - [Basic Usage](#basic-usage)
+ - [Class-Based Pipes](#class-based-pipes)
+ - [Invokable Classes](#invokable-classes)
+ - [Using Different Method Names](#using-different-method-names)
+ - [Passing Parameters to Pipes](#passing-parameters-to-pipes)
+ - [Early Pipeline Termination](#early-pipeline-termination)
+ - [Conditional Pipeline Execution](#conditional-pipeline-execution)
+- [Advanced Usage](#advanced-usage)
+ - [Working with Objects](#working-with-objects)
+ - [Async Operations](#async-operations)
+- [Laravel API Compatibility](#laravel-api-compatibility)
+- [Comparison with Laravel](#comparison-with-laravel)
+- [Troubleshooting](#troubleshooting)
+- [Testing](#testing)
+- [Contributing](#contributing)
+- [License](#license)
+
+## Overview
+
+Platform Pipeline is a 100% API-compatible port of Laravel's Pipeline to Dart. It allows you to pass an object through a series of operations (pipes) in a fluent, maintainable way. Each pipe can examine, modify, or replace the object before passing it to the next pipe in the sequence.
+
+## Features
+
+- ๐ฏ 100% Laravel Pipeline API compatibility
+- ๐ Support for class-based and callable pipes
+- ๐ฏ Dependency injection through container integration
+- โก Async operation support
+- ๐ Conditional pipeline execution
+- ๐ญ Method name customization via `via()`
+- ๐ Parameter passing to pipes
+- ๐ Early pipeline termination
+- ๐งช Comprehensive test coverage
+
+## Requirements
+
+- Dart SDK: >=2.17.0 <4.0.0
+- platform_container: ^1.0.0
+
+## Installation
+
+Add this to your package's `pubspec.yaml` file:
+
+```yaml
+dependencies:
+ platform_pipeline: ^1.0.0
+```
+
+## Usage
+
+### Basic Usage
+
+```dart
+import 'package:platform_pipeline/pipeline.dart';
+import 'package:platform_container/container.dart';
+
+void main() async {
+ // Create a container instance
+ var container = Container();
+
+ // Create a pipeline
+ var result = await Pipeline(container)
+ .send('Hello')
+ .through([
+ (String value, next) => next(value + ' World'),
+ (String value, next) => next(value + '!'),
+ ])
+ .then((value) => value);
+
+ print(result); // Outputs: Hello World!
+}
+```
+
+### Class-Based Pipes
+
+```dart
+class UppercasePipe {
+ Future handle(String value, Function next) async {
+ return next(value.toUpperCase());
+ }
+}
+
+class AddExclamationPipe {
+ Future handle(String value, Function next) async {
+ return next(value + '!');
+ }
+}
+
+void main() async {
+ var container = Container();
+
+ var result = await Pipeline(container)
+ .send('hello')
+ .through([
+ UppercasePipe(),
+ AddExclamationPipe(),
+ ])
+ .then((value) => value);
+
+ print(result); // Outputs: HELLO!
+}
+```
+
+### Invokable Classes
+
+```dart
+class TransformPipe {
+ Future call(String value, Function next) async {
+ return next(value.toUpperCase());
+ }
+}
+
+void main() async {
+ var container = Container();
+
+ var result = await Pipeline(container)
+ .send('hello')
+ .through([TransformPipe()])
+ .then((value) => value);
+
+ print(result); // Outputs: HELLO
+}
+```
+
+### Using Different Method Names
+
+```dart
+class CustomPipe {
+ Future transform(String value, Function next) async {
+ return next(value.toUpperCase());
+ }
+}
+
+void main() async {
+ var container = Container();
+
+ var result = await Pipeline(container)
+ .send('hello')
+ .through([CustomPipe()])
+ .via('transform')
+ .then((value) => value);
+
+ print(result); // Outputs: HELLO
+}
+```
+
+### Passing Parameters to Pipes
+
+```dart
+class PrefixPipe {
+ Future handle(
+ String value,
+ Function next, [
+ String prefix = '',
+ ]) async {
+ return next('$prefix$value');
+ }
+}
+
+void main() async {
+ var container = Container();
+ container.registerFactory((c) => PrefixPipe());
+
+ var pipeline = Pipeline(container);
+ pipeline.registerPipeType('PrefixPipe', PrefixPipe);
+
+ var result = await pipeline
+ .send('World')
+ .through('PrefixPipe:Hello ')
+ .then((value) => value);
+
+ print(result); // Outputs: Hello World
+}
+```
+
+### Early Pipeline Termination
+
+```dart
+void main() async {
+ var container = Container();
+
+ var result = await Pipeline(container)
+ .send('hello')
+ .through([
+ (value, next) => 'TERMINATED', // Pipeline stops here
+ (value, next) => next('NEVER REACHED'),
+ ])
+ .then((value) => value);
+
+ print(result); // Outputs: TERMINATED
+}
+```
+
+### Conditional Pipeline Execution
+
+```dart
+void main() async {
+ var container = Container();
+ var shouldTransform = true;
+
+ var result = await Pipeline(container)
+ .send('hello')
+ .when(() => shouldTransform, (Pipeline pipeline) {
+ pipeline.pipe([
+ (value, next) => next(value.toUpperCase()),
+ ]);
+ })
+ .then((value) => value);
+
+ print(result); // Outputs: HELLO
+}
+```
+
+## Advanced Usage
+
+### Working with Objects
+
+```dart
+class User {
+ String name;
+ int age;
+
+ User(this.name, this.age);
+}
+
+class AgeValidationPipe {
+ Future handle(User user, Function next) async {
+ if (user.age < 18) {
+ throw Exception('User must be 18 or older');
+ }
+ return next(user);
+ }
+}
+
+class NameFormattingPipe {
+ Future handle(User user, Function next) async {
+ user.name = user.name.trim().toLowerCase();
+ return next(user);
+ }
+}
+
+void main() async {
+ var container = Container();
+
+ var user = User('John Doe ', 20);
+
+ try {
+ user = await Pipeline(container)
+ .send(user)
+ .through([
+ AgeValidationPipe(),
+ NameFormattingPipe(),
+ ])
+ .then((value) => value);
+
+ print('${user.name} is ${user.age} years old');
+ // Outputs: john doe is 20 years old
+ } catch (e) {
+ print('Validation failed: $e');
+ }
+}
+```
+
+### Async Operations
+
+```dart
+class AsyncTransformPipe {
+ Future handle(String value, Function next) async {
+ // Simulate async operation
+ await Future.delayed(Duration(seconds: 1));
+ return next(value.toUpperCase());
+ }
+}
+
+void main() async {
+ var container = Container();
+
+ var result = await Pipeline(container)
+ .send('hello')
+ .through([AsyncTransformPipe()])
+ .then((value) => value);
+
+ print(result); // Outputs after 1 second: HELLO
+}
+```
+
+## Laravel API Compatibility
+
+This package maintains 100% API compatibility with Laravel's Pipeline implementation. All Laravel Pipeline features are supported:
+
+- `send()` - Set the object being passed through the pipeline
+- `through()` - Set the array of pipes
+- `pipe()` - Push additional pipes onto the pipeline
+- `via()` - Set the method to call on the pipes
+- `then()` - Run the pipeline with a final destination callback
+- `thenReturn()` - Run the pipeline and return the result
+
+## Comparison with Laravel
+
+| Feature | Laravel | Platform Pipeline |
+|---------|---------|------------------|
+| API Methods | โ | โ |
+| Container Integration | โ | โ |
+| Pipe Types | Class, Callable | Class, Callable |
+| Async Support | โ | โ |
+| Type Safety | โ | โ |
+| Parameter Passing | โ | โ |
+| Early Termination | โ | โ |
+| Method Customization | โ | โ |
+| Conditional Execution | โ | โ |
+
+## Troubleshooting
+
+### Common Issues
+
+1. Container Not Provided
+```dart
+// โ Wrong
+var pipeline = Pipeline(null);
+
+// โ Correct
+var container = Container();
+var pipeline = Pipeline(container);
+```
+
+2. Missing Type Registration
+```dart
+// โ Wrong
+pipeline.through('CustomPipe:param');
+
+// โ Correct
+pipeline.registerPipeType('CustomPipe', CustomPipe);
+pipeline.through('CustomPipe:param');
+```
+
+3. Incorrect Method Name
+```dart
+// โ Wrong
+class CustomPipe {
+ void process(value, next) {} // Wrong method name
+}
+
+// โ Correct
+class CustomPipe {
+ void handle(value, next) {} // Default method name
+}
+// Or specify the method name:
+pipeline.via('process').through([CustomPipe()]);
+```
+
+## Testing
+
+Run the tests with:
+
+```bash
+dart test
+```
+
+## Contributing
+
+Contributions are welcome! Please feel free to submit a Pull Request.
+
+## License
+
+This package is open-sourced software licensed under the MIT license.
diff --git a/packages/pipeline/analysis_options.yaml b/packages/pipeline/analysis_options.yaml
new file mode 100644
index 0000000..dee8927
--- /dev/null
+++ b/packages/pipeline/analysis_options.yaml
@@ -0,0 +1,30 @@
+# This file configures the static analysis results for your project (errors,
+# warnings, and lints).
+#
+# This enables the 'recommended' set of lints from `package:lints`.
+# This set helps identify many issues that may lead to problems when running
+# or consuming Dart code, and enforces writing Dart using a single, idiomatic
+# style and format.
+#
+# If you want a smaller set of lints you can change this to specify
+# 'package:lints/core.yaml'. These are just the most critical lints
+# (the recommended set includes the core lints).
+# The core lints are also what is used by pub.dev for scoring packages.
+
+include: package:lints/recommended.yaml
+
+# Uncomment the following section to specify additional rules.
+
+# linter:
+# rules:
+# - camel_case_types
+
+# analyzer:
+# exclude:
+# - path/to/excluded/files/**
+
+# For more information about the core and recommended set of lints, see
+# https://dart.dev/go/core-lints
+
+# For additional information about configuring this file, see
+# https://dart.dev/guides/language/analysis-options
diff --git a/packages/pipeline/doc/.gitkeep b/packages/pipeline/doc/.gitkeep
new file mode 100644
index 0000000..e69de29
diff --git a/packages/pipeline/examples/async_pipeline.dart b/packages/pipeline/examples/async_pipeline.dart
new file mode 100644
index 0000000..4291edc
--- /dev/null
+++ b/packages/pipeline/examples/async_pipeline.dart
@@ -0,0 +1,38 @@
+import 'package:platform_foundation/core.dart';
+import 'package:platform_foundation/http.dart';
+import 'package:platform_container/mirrors.dart';
+import 'package:platform_pipeline/pipeline.dart';
+
+class AsyncGreetingPipe {
+ Future handle(String input, Function next) async {
+ await Future.delayed(Duration(seconds: 1));
+ return next('Hello, $input');
+ }
+}
+
+class AsyncExclamationPipe {
+ Future handle(String input, Function next) async {
+ await Future.delayed(Duration(seconds: 1));
+ return next('$input!');
+ }
+}
+
+void main() async {
+ var app = Application(reflector: MirrorsReflector());
+ var http = PlatformHttp(app);
+
+ app.container.registerSingleton((c) => Pipeline(c));
+
+ app.get('/', (req, res) async {
+ var pipeline = app.container.make();
+ var result = await pipeline
+ .send('World')
+ .through(['AsyncGreetingPipe', 'AsyncExclamationPipe']).then(
+ (result) => result.toUpperCase());
+
+ res.write(result); // Outputs: "HELLO, WORLD!" (after 2 seconds)
+ });
+
+ await http.startServer('localhost', 3000);
+ print('Server started on http://localhost:3000');
+}
diff --git a/packages/pipeline/examples/basic_usage.dart b/packages/pipeline/examples/basic_usage.dart
new file mode 100644
index 0000000..a78c727
--- /dev/null
+++ b/packages/pipeline/examples/basic_usage.dart
@@ -0,0 +1,36 @@
+import 'package:platform_foundation/core.dart';
+import 'package:platform_foundation/http.dart';
+import 'package:platform_container/mirrors.dart';
+import 'package:platform_pipeline/pipeline.dart';
+
+class GreetingPipe {
+ dynamic handle(String input, Function next) {
+ return next('Hello, $input');
+ }
+}
+
+class ExclamationPipe {
+ dynamic handle(String input, Function next) {
+ return next('$input!');
+ }
+}
+
+void main() async {
+ var app = Application(reflector: MirrorsReflector());
+ var http = PlatformHttp(app);
+
+ app.container.registerSingleton((c) => Pipeline(c));
+
+ app.get('/', (req, res) async {
+ var pipeline = app.container.make();
+ var result = await pipeline
+ .send('World')
+ .through(['GreetingPipe', 'ExclamationPipe']).then(
+ (result) => result.toUpperCase());
+
+ res.write(result); // Outputs: "HELLO, WORLD!"
+ });
+
+ await http.startServer('localhost', 3000);
+ print('Server started on http://localhost:3000');
+}
diff --git a/packages/pipeline/examples/error_handling.dart b/packages/pipeline/examples/error_handling.dart
new file mode 100644
index 0000000..b141459
--- /dev/null
+++ b/packages/pipeline/examples/error_handling.dart
@@ -0,0 +1,34 @@
+import 'package:platform_foundation/core.dart';
+import 'package:platform_foundation/http.dart';
+import 'package:platform_container/mirrors.dart';
+import 'package:platform_pipeline/pipeline.dart';
+
+class ErrorPipe {
+ dynamic handle(String input, Function next) {
+ throw Exception('Simulated error');
+ }
+}
+
+void main() async {
+ var app = Application(reflector: MirrorsReflector());
+ var http = PlatformHttp(app);
+
+ app.container.registerSingleton((c) => Pipeline(c));
+
+ app.get('/', (req, res) async {
+ var pipeline = app.container.make();
+ try {
+ await pipeline
+ .send('World')
+ .through(['ErrorPipe']).then((result) => result.toUpperCase());
+ } catch (e) {
+ res.write('Error occurred: ${e.toString()}');
+ return;
+ }
+
+ res.write('This should not be reached');
+ });
+
+ await http.startServer('localhost', 3000);
+ print('Server started on http://localhost:3000');
+}
diff --git a/packages/pipeline/examples/mixed_pipes.dart b/packages/pipeline/examples/mixed_pipes.dart
new file mode 100644
index 0000000..e8aabf0
--- /dev/null
+++ b/packages/pipeline/examples/mixed_pipes.dart
@@ -0,0 +1,35 @@
+import 'package:platform_foundation/core.dart';
+import 'package:platform_foundation/http.dart';
+import 'package:platform_container/mirrors.dart';
+import 'package:platform_pipeline/pipeline.dart';
+
+class GreetingPipe {
+ dynamic handle(String input, Function next) {
+ return next('Hello, $input');
+ }
+}
+
+void main() async {
+ var app = Application(reflector: MirrorsReflector());
+ var http = PlatformHttp(app);
+
+ app.container.registerSingleton((c) => Pipeline(c));
+
+ app.get('/', (req, res) async {
+ var pipeline = app.container.make();
+ var result = await pipeline.send('World').through([
+ 'GreetingPipe',
+ (String input, Function next) => next('$input!'),
+ (String input, Function next) async {
+ await Future.delayed(Duration(seconds: 1));
+ return next(input.toUpperCase());
+ },
+ ]).then((result) => 'Final result: $result');
+
+ res.write(
+ result); // Outputs: "Final result: HELLO, WORLD!" (after 1 second)
+ });
+
+ await http.startServer('localhost', 3000);
+ print('Server started on http://localhost:3000');
+}
diff --git a/packages/pipeline/lib/pipeline.dart b/packages/pipeline/lib/pipeline.dart
new file mode 100644
index 0000000..9b73f9f
--- /dev/null
+++ b/packages/pipeline/lib/pipeline.dart
@@ -0,0 +1,5 @@
+library;
+
+export 'src/pipeline.dart';
+export 'src/conditionable.dart';
+export 'src/pipeline_contract.dart';
diff --git a/packages/pipeline/lib/src/conditionable.dart b/packages/pipeline/lib/src/conditionable.dart
new file mode 100644
index 0000000..ce796a0
--- /dev/null
+++ b/packages/pipeline/lib/src/conditionable.dart
@@ -0,0 +1,16 @@
+/// Provides conditional execution methods for the pipeline.
+mixin Conditionable {
+ T when(bool Function() callback, void Function(T) callback2) {
+ if (callback()) {
+ callback2(this as T);
+ }
+ return this as T;
+ }
+
+ T unless(bool Function() callback, void Function(T) callback2) {
+ if (!callback()) {
+ callback2(this as T);
+ }
+ return this as T;
+ }
+}
diff --git a/packages/pipeline/lib/src/pipeline.dart b/packages/pipeline/lib/src/pipeline.dart
new file mode 100644
index 0000000..37d9d68
--- /dev/null
+++ b/packages/pipeline/lib/src/pipeline.dart
@@ -0,0 +1,241 @@
+import 'dart:async';
+import 'dart:mirrors';
+import 'package:platform_container/container.dart';
+import 'package:logging/logging.dart';
+import 'pipeline_contract.dart';
+import 'conditionable.dart';
+
+/// Defines the signature for a pipe function.
+typedef PipeFunction = FutureOr Function(
+ dynamic passable, FutureOr Function(dynamic) next);
+
+/// The primary class for building and executing pipelines.
+class Pipeline with Conditionable implements PipelineContract {
+ /// The container implementation.
+ Container? _container;
+
+ final Map _typeMap = {};
+
+ /// The object being passed through the pipeline.
+ dynamic _passable;
+
+ /// The array of class pipes.
+ final List _pipes = [];
+
+ /// The method to call on each pipe.
+ String _method = 'handle';
+
+ /// Logger for the pipeline.
+ final Logger _logger = Logger('Pipeline');
+
+ /// Create a new class instance.
+ Pipeline(this._container);
+
+ void registerPipeType(String name, Type type) {
+ _typeMap[name] = type;
+ }
+
+ /// Set the object being sent through the pipeline.
+ @override
+ Pipeline send(dynamic passable) {
+ _passable = passable;
+ return this;
+ }
+
+ /// Set the array of pipes.
+ @override
+ Pipeline through(dynamic pipes) {
+ if (_container == null) {
+ throw Exception(
+ 'A container instance has not been passed to the Pipeline.');
+ }
+ _pipes.addAll(pipes is Iterable ? pipes.toList() : [pipes]);
+ return this;
+ }
+
+ /// Push additional pipes onto the pipeline.
+ @override
+ Pipeline pipe(dynamic pipes) {
+ if (_container == null) {
+ throw Exception(
+ 'A container instance has not been passed to the Pipeline.');
+ }
+ _pipes.addAll(pipes is Iterable ? pipes.toList() : [pipes]);
+ return this;
+ }
+
+ /// Set the method to call on the pipes.
+ @override
+ Pipeline via(String method) {
+ _method = method;
+ return this;
+ }
+
+ /// Run the pipeline with a final destination callback.
+ @override
+ Future then(FutureOr Function(dynamic) destination) async {
+ if (_container == null) {
+ throw Exception(
+ 'A container instance has not been passed to the Pipeline.');
+ }
+
+ var pipeline = (dynamic passable) async => await destination(passable);
+
+ for (var pipe in _pipes.reversed) {
+ var next = pipeline;
+ pipeline = (dynamic passable) async {
+ return await carry(pipe, passable, next);
+ };
+ }
+
+ return await pipeline(_passable);
+ }
+
+ /// Run the pipeline and return the result.
+ @override
+ Future thenReturn() async {
+ return then((passable) => passable);
+ }
+
+ /// Get a Closure that represents a slice of the application onion.
+ Future carry(dynamic pipe, dynamic passable, Function next) async {
+ try {
+ if (pipe is Function) {
+ return await pipe(passable, next);
+ }
+
+ if (pipe is String) {
+ if (_container == null) {
+ throw Exception('Container is null, cannot resolve pipe: $pipe');
+ }
+
+ final parts = parsePipeString(pipe);
+ final pipeClass = parts[0];
+ final parameters = parts.length > 1 ? parts.sublist(1) : [];
+
+ Type? pipeType;
+ if (_typeMap.containsKey(pipeClass)) {
+ pipeType = _typeMap[pipeClass];
+ } else {
+ // Try to resolve from mirrors
+ try {
+ for (var lib in currentMirrorSystem().libraries.values) {
+ for (var decl in lib.declarations.values) {
+ if (decl is ClassMirror &&
+ decl.simpleName == Symbol(pipeClass)) {
+ pipeType = decl.reflectedType;
+ break;
+ }
+ }
+ if (pipeType != null) break;
+ }
+ } catch (_) {}
+
+ if (pipeType == null) {
+ throw Exception('Type not registered for pipe: $pipe');
+ }
+ }
+
+ var instance = _container?.make(pipeType);
+ if (instance == null) {
+ throw Exception('Unable to resolve pipe: $pipe');
+ }
+
+ return await invokeMethod(
+ instance, _method, [passable, next, ...parameters]);
+ }
+
+ if (pipe is Type) {
+ if (_container == null) {
+ throw Exception('Container is null, cannot resolve pipe type');
+ }
+
+ var instance = _container?.make(pipe);
+ if (instance == null) {
+ throw Exception('Unable to resolve pipe type: $pipe');
+ }
+
+ return await invokeMethod(instance, _method, [passable, next]);
+ }
+
+ // Handle instance of a class
+ if (pipe is Object) {
+ return await invokeMethod(pipe, _method, [passable, next]);
+ }
+
+ throw Exception('Unsupported pipe type: ${pipe.runtimeType}');
+ } catch (e) {
+ return handleException(passable, e);
+ }
+ }
+
+ /// Parse full pipe string to get name and parameters.
+ List parsePipeString(String pipe) {
+ var parts = pipe.split(':');
+ return [parts[0], if (parts.length > 1) ...parts[1].split(',')];
+ }
+
+ /// Get the array of configured pipes.
+ List pipes() {
+ return List.unmodifiable(_pipes);
+ }
+
+ /// Get the container instance.
+ Container getContainer() {
+ if (_container == null) {
+ throw Exception(
+ 'A container instance has not been passed to the Pipeline.');
+ }
+ return _container!;
+ }
+
+ /// Set the container instance.
+ Pipeline setContainer(Container container) {
+ _container = container;
+ return this;
+ }
+
+ /// Handle the value returned from each pipe before passing it to the next.
+ dynamic handleCarry(dynamic carry) {
+ if (carry is Future) {
+ return carry.then((value) => value ?? _passable);
+ }
+ return carry ?? _passable;
+ }
+
+ Future invokeMethod(
+ dynamic instance, String methodName, List arguments) async {
+ // First try call() for invokable objects
+ if (instance is Function) {
+ return await instance(arguments[0], arguments[1]);
+ }
+
+ var instanceMirror = reflect(instance);
+
+ // Check for call method first (invokable objects)
+ var callSymbol = Symbol('call');
+ if (instanceMirror.type.declarations.containsKey(callSymbol)) {
+ var result = instanceMirror.invoke(callSymbol, arguments);
+ return await result.reflectee;
+ }
+
+ // Then try the specified method
+ var methodSymbol = Symbol(methodName);
+ if (!instanceMirror.type.declarations.containsKey(methodSymbol)) {
+ throw Exception('Method $methodName not found on instance: $instance');
+ }
+
+ var result = instanceMirror.invoke(methodSymbol, arguments);
+ return await result.reflectee;
+ }
+
+ /// Handle the given exception.
+ dynamic handleException(dynamic passable, Object e) {
+ if (e is Exception && e.toString().contains('Container is null')) {
+ throw Exception(
+ 'A container instance has not been passed to the Pipeline.');
+ }
+ _logger.severe('Exception occurred in pipeline', e);
+ throw e;
+ }
+}
diff --git a/packages/pipeline/lib/src/pipeline_contract.dart b/packages/pipeline/lib/src/pipeline_contract.dart
new file mode 100644
index 0000000..2b45e7f
--- /dev/null
+++ b/packages/pipeline/lib/src/pipeline_contract.dart
@@ -0,0 +1,9 @@
+/// Represents a series of "pipes" through which an object can be passed.
+abstract class PipelineContract {
+ PipelineContract send(dynamic passable);
+ PipelineContract through(dynamic pipes);
+ PipelineContract pipe(dynamic pipes);
+ PipelineContract via(String method);
+ Future then(dynamic Function(dynamic) destination);
+ Future thenReturn();
+}
diff --git a/packages/pipeline/pubspec.yaml b/packages/pipeline/pubspec.yaml
new file mode 100644
index 0000000..2a84b28
--- /dev/null
+++ b/packages/pipeline/pubspec.yaml
@@ -0,0 +1,19 @@
+name: platform_pipeline
+description: The Pipeline Package for the Protevus Platform
+version: 0.0.1
+homepage: https://protevus.com
+documentation: https://docs.protevus.com
+repository: https://github.com/protevus/platform
+
+environment:
+ sdk: ^3.4.2
+
+# Add regular dependencies here.
+dependencies:
+ platform_container: ^9.0.0
+ platform_foundation: ^9.0.0
+ logging: ^1.1.0
+
+dev_dependencies:
+ lints: ^3.0.0
+ test: ^1.24.0
diff --git a/packages/pipeline/test/laravel_pipeline_test.dart b/packages/pipeline/test/laravel_pipeline_test.dart
new file mode 100644
index 0000000..0fcdeaa
--- /dev/null
+++ b/packages/pipeline/test/laravel_pipeline_test.dart
@@ -0,0 +1,258 @@
+import 'package:platform_container/container.dart';
+import 'package:platform_pipeline/pipeline.dart';
+import 'package:test/test.dart';
+
+// Test pipe classes to match Laravel's test classes
+class PipelineTestPipeOne {
+ static String? testPipeOne;
+
+ Future handle(dynamic piped, Function next) async {
+ testPipeOne = piped.toString();
+ return next(piped);
+ }
+
+ Future differentMethod(dynamic piped, Function next) async {
+ return next(piped);
+ }
+}
+
+class PipelineTestPipeTwo {
+ static String? testPipeOne;
+
+ Future call(dynamic piped, Function next) async {
+ testPipeOne = piped.toString();
+ return next(piped);
+ }
+}
+
+class PipelineTestParameterPipe {
+ static List? testParameters;
+
+ Future handle(dynamic piped, Function next,
+ [String? parameter1, String? parameter2]) async {
+ testParameters = [
+ if (parameter1 != null) parameter1,
+ if (parameter2 != null) parameter2
+ ];
+ return next(piped);
+ }
+}
+
+void main() {
+ group('Laravel Pipeline Tests', () {
+ late Container container;
+ late Pipeline pipeline;
+
+ setUp(() {
+ container = Container(const EmptyReflector());
+ pipeline = Pipeline(container);
+
+ // Register test classes with container
+ container
+ .registerFactory((c) => PipelineTestPipeOne());
+ container
+ .registerFactory((c) => PipelineTestPipeTwo());
+ container.registerFactory(
+ (c) => PipelineTestParameterPipe());
+
+ // Register types with pipeline
+ pipeline.registerPipeType('PipelineTestPipeOne', PipelineTestPipeOne);
+ pipeline.registerPipeType('PipelineTestPipeTwo', PipelineTestPipeTwo);
+ pipeline.registerPipeType(
+ 'PipelineTestParameterPipe', PipelineTestParameterPipe);
+
+ // Reset static test variables
+ PipelineTestPipeOne.testPipeOne = null;
+ PipelineTestPipeTwo.testPipeOne = null;
+ PipelineTestParameterPipe.testParameters = null;
+ });
+
+ test('Pipeline basic usage', () async {
+ String? testPipeTwo;
+ final pipeTwo = (dynamic piped, Function next) {
+ testPipeTwo = piped.toString();
+ return next(piped);
+ };
+
+ final result = await Pipeline(container)
+ .send('foo')
+ .through([PipelineTestPipeOne(), pipeTwo]).then((piped) => piped);
+
+ expect(result, equals('foo'));
+ expect(PipelineTestPipeOne.testPipeOne, equals('foo'));
+ expect(testPipeTwo, equals('foo'));
+ });
+
+ test('Pipeline usage with objects', () async {
+ final result = await Pipeline(container)
+ .send('foo')
+ .through([PipelineTestPipeOne()]).then((piped) => piped);
+
+ expect(result, equals('foo'));
+ expect(PipelineTestPipeOne.testPipeOne, equals('foo'));
+ });
+
+ test('Pipeline usage with invokable objects', () async {
+ final result = await Pipeline(container)
+ .send('foo')
+ .through([PipelineTestPipeTwo()]).then((piped) => piped);
+
+ expect(result, equals('foo'));
+ expect(PipelineTestPipeTwo.testPipeOne, equals('foo'));
+ });
+
+ test('Pipeline usage with callable', () async {
+ String? testPipeOne;
+ final function = (dynamic piped, Function next) {
+ testPipeOne = 'foo';
+ return next(piped);
+ };
+
+ var result = await Pipeline(container)
+ .send('foo')
+ .through([function]).then((piped) => piped);
+
+ expect(result, equals('foo'));
+ expect(testPipeOne, equals('foo'));
+
+ testPipeOne = null;
+
+ result =
+ await Pipeline(container).send('bar').through(function).thenReturn();
+
+ expect(result, equals('bar'));
+ expect(testPipeOne, equals('foo'));
+ });
+
+ test('Pipeline usage with pipe', () async {
+ final object = {'value': 0};
+
+ final function = (dynamic obj, Function next) {
+ obj['value']++;
+ return next(obj);
+ };
+
+ final result = await Pipeline(container)
+ .send(object)
+ .through([function]).pipe([function]).then((piped) => piped);
+
+ expect(result, equals(object));
+ expect(object['value'], equals(2));
+ });
+
+ test('Pipeline usage with invokable class', () async {
+ final result = await Pipeline(container)
+ .send('foo')
+ .through([PipelineTestPipeTwo()]).then((piped) => piped);
+
+ expect(result, equals('foo'));
+ expect(PipelineTestPipeTwo.testPipeOne, equals('foo'));
+ });
+
+ test('Then method is not called if the pipe returns', () async {
+ String thenValue = '(*_*)';
+ String secondValue = '(*_*)';
+
+ final result = await Pipeline(container).send('foo').through([
+ (value, next) => 'm(-_-)m',
+ (value, next) {
+ secondValue = 'm(-_-)m';
+ return next(value);
+ },
+ ]).then((piped) {
+ thenValue = '(0_0)';
+ return piped;
+ });
+
+ expect(result, equals('m(-_-)m'));
+ // The then callback is not called
+ expect(thenValue, equals('(*_*)'));
+ // The second pipe is not called
+ expect(secondValue, equals('(*_*)'));
+ });
+
+ test('Then method input value', () async {
+ String? pipeReturn;
+ String? thenArg;
+
+ final result = await Pipeline(container).send('foo').through([
+ (value, next) async {
+ final nextValue = await next('::not_foo::');
+ pipeReturn = nextValue;
+ return 'pipe::$nextValue';
+ }
+ ]).then((piped) {
+ thenArg = piped;
+ return 'then$piped';
+ });
+
+ expect(result, equals('pipe::then::not_foo::'));
+ expect(thenArg, equals('::not_foo::'));
+ });
+
+ test('Pipeline usage with parameters', () async {
+ final parameters = ['one', 'two'];
+
+ final result = await Pipeline(container)
+ .send('foo')
+ .through('PipelineTestParameterPipe:${parameters.join(',')}')
+ .then((piped) => piped);
+
+ expect(result, equals('foo'));
+ expect(PipelineTestParameterPipe.testParameters, equals(parameters));
+ });
+
+ test('Pipeline via changes the method being called on the pipes', () async {
+ final result = await Pipeline(container)
+ .send('data')
+ .through(PipelineTestPipeOne())
+ .via('differentMethod')
+ .then((piped) => piped);
+
+ expect(result, equals('data'));
+ });
+
+ test('Pipeline throws exception on resolve without container', () async {
+ expect(
+ () => Pipeline(null)
+ .send('data')
+ .through(PipelineTestPipeOne())
+ .then((piped) => piped),
+ throwsA(isA().having(
+ (e) => e.toString(),
+ 'message',
+ contains(
+ 'A container instance has not been passed to the Pipeline'))));
+ });
+
+ test('Pipeline thenReturn method runs pipeline then returns passable',
+ () async {
+ final result = await Pipeline(container)
+ .send('foo')
+ .through([PipelineTestPipeOne()]).thenReturn();
+
+ expect(result, equals('foo'));
+ expect(PipelineTestPipeOne.testPipeOne, equals('foo'));
+ });
+
+ test('Pipeline conditionable', () async {
+ var result = await Pipeline(container).send('foo').when(() => true,
+ (Pipeline pipeline) {
+ pipeline.pipe([PipelineTestPipeOne()]);
+ }).then((piped) => piped);
+
+ expect(result, equals('foo'));
+ expect(PipelineTestPipeOne.testPipeOne, equals('foo'));
+
+ PipelineTestPipeOne.testPipeOne = null;
+
+ result = await Pipeline(container).send('foo').when(() => false,
+ (Pipeline pipeline) {
+ pipeline.pipe([PipelineTestPipeOne()]);
+ }).then((piped) => piped);
+
+ expect(result, equals('foo'));
+ expect(PipelineTestPipeOne.testPipeOne, isNull);
+ });
+ });
+}
diff --git a/packages/pipeline/test/pipeline_test.dart b/packages/pipeline/test/pipeline_test.dart
new file mode 100644
index 0000000..7f2fdb7
--- /dev/null
+++ b/packages/pipeline/test/pipeline_test.dart
@@ -0,0 +1,106 @@
+import 'package:test/test.dart';
+import 'package:platform_foundation/core.dart';
+import 'package:platform_container/container.dart';
+import 'package:platform_container/mirrors.dart';
+import 'package:platform_pipeline/pipeline.dart';
+
+class AddExclamationPipe {
+ Future handle(String input, Function next) async {
+ return await next('$input!');
+ }
+}
+
+class UppercasePipe {
+ Future handle(String input, Function next) async {
+ return await next(input.toUpperCase());
+ }
+}
+
+void main() {
+ late Application app;
+ late Container container;
+ late Pipeline pipeline;
+
+ setUp(() {
+ app = Application(reflector: MirrorsReflector());
+ container = app.container;
+ container.registerSingleton(AddExclamationPipe());
+ container.registerSingleton(UppercasePipe());
+ pipeline = Pipeline(container);
+ pipeline.registerPipeType('AddExclamationPipe', AddExclamationPipe);
+ pipeline.registerPipeType('UppercasePipe', UppercasePipe);
+ });
+
+ test('Pipeline should process simple string pipes', () async {
+ var result = await pipeline.send('hello').through(
+ ['AddExclamationPipe', 'UppercasePipe']).then((res) async => res);
+ expect(result, equals('HELLO!'));
+ });
+
+ test('Pipeline should process function pipes', () async {
+ var result = await pipeline.send('hello').through([
+ (String input, Function next) async {
+ var result = await next('$input, WORLD');
+ return result;
+ },
+ (String input, Function next) async {
+ var result = await next(input.toUpperCase());
+ return result;
+ },
+ ]).then((res) async => res as String);
+
+ expect(result, equals('HELLO, WORLD'));
+ });
+
+ test('Pipeline should handle mixed pipe types', () async {
+ var result = await pipeline.send('hello').through([
+ 'AddExclamationPipe',
+ (String input, Function next) async {
+ var result = await next(input.toUpperCase());
+ return result;
+ },
+ ]).then((res) async => res as String);
+ expect(result, equals('HELLO!'));
+ });
+
+ test('Pipeline should handle async pipes', () async {
+ var result = await pipeline.send('hello').through([
+ 'UppercasePipe',
+ (String input, Function next) async {
+ await Future.delayed(Duration(milliseconds: 100));
+ return next('$input, WORLD');
+ },
+ ]).then((res) async => res as String);
+ expect(result, equals('HELLO, WORLD'));
+ });
+
+ test('Pipeline should throw exception for unresolvable pipe', () {
+ expect(
+ () => pipeline
+ .send('hello')
+ .through(['NonExistentPipe']).then((res) => res),
+ throwsA(isA()),
+ );
+ });
+
+ test('Pipeline should allow chaining of pipes', () async {
+ var result = await pipeline
+ .send('hello')
+ .pipe('AddExclamationPipe')
+ .pipe('UppercasePipe')
+ .then((res) async => res as String);
+ expect(result, equals('HELLO!'));
+ });
+
+ test('Pipeline should respect the order of pipes', () async {
+ var result1 = await pipeline
+ .send('hello')
+ .through(['AddExclamationPipe', 'UppercasePipe']).then((res) => res);
+ var result2 = await pipeline
+ .send('hello')
+ .through(['UppercasePipe', 'AddExclamationPipe']).then((res) => res);
+ expect(result1, equals('HELLO!'));
+ expect(result2, equals('HELLO!!'));
+ expect(result1, isNot(equals(result2)));
+ });
+}