Add: adding pipeline package not production ready

core/eventbus/lib/event_bus.dart

@@ -1,3 +1,3 @@
-library event_bus_plus;
+library event_bus;
 
 export 'res/res.dart';

core/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

core/pipeline/CHANGELOG.md

@@ -0,0 +1,3 @@
+## 1.0.0
+
+- Initial version.

core/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.

core/pipeline/README.md

@@ -0,0 +1 @@
+<p align="center"><a href="https://protevus.com" target="_blank"><img src="https://git.protevus.com/protevus/branding/raw/branch/main/protevus-logo-bg.png"></a></p>

core/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

core/pipeline/lib/pipeline.dart

@@ -0,0 +1,3 @@
+library;
+
+export 'src/pipeline.dart';

core/pipeline/lib/src/pipeline.dart

@@ -0,0 +1,223 @@
+import 'dart:async';
+import 'package:angel3_container/angel3_container.dart';
+import 'package:logging/logging.dart';
+
+/// 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<dynamic> then(dynamic Function(dynamic) destination);
+  Future<dynamic> thenReturn();
+}
+
+/// Provides conditional execution methods for the pipeline.
+mixin Conditionable<T> {
+  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;
+  }
+}
+
+/// The primary class for building and executing pipelines.
+class Pipeline with Conditionable<Pipeline> implements PipelineContract {
+  /// The container implementation.
+  Container? _container;
+
+  /// The object being passed through the pipeline.
+  dynamic _passable;
+
+  /// The array of class pipes.
+  final List<dynamic> _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]);
+
+  /// 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) {
+    _pipes.clear();
+    _pipes.addAll(pipes is Iterable ? pipes.toList() : [pipes]);
+    return this;
+  }
+
+  /// Push additional pipes onto the pipeline.
+  @override
+  Pipeline pipe(dynamic pipes) {
+    _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<dynamic> then(dynamic Function(dynamic) destination) async {
+    var pipeline = pipes().fold<Function>(
+      (passable) => prepareDestination(destination),
+      (Function next, pipe) => (passable) => carry(pipe, passable, next),
+    );
+
+    return pipeline(_passable);
+  }
+
+  /// Run the pipeline and return the result.
+  @override
+  Future<dynamic> thenReturn() async {
+    return then((passable) => passable);
+  }
+
+  /// Get the final piece of the Closure onion.
+  Function prepareDestination(Function destination) {
+    return (passable) async {
+      try {
+        var result = destination(passable);
+        return result is Future ? await result : result;
+      } catch (e) {
+        return handleException(passable, e);
+      }
+    };
+  }
+
+  /// Get a Closure that represents a slice of the application onion.
+  Future<dynamic> carry(dynamic pipe, dynamic passable, Function next) async {
+    try {
+      if (pipe is Function) {
+        var result = pipe(passable, next);
+        return result is Future ? await result : result;
+      }
+
+      List<String> parameters = [];
+      if (pipe is String) {
+        var parts = parsePipeString(pipe);
+        pipe = parts[0];
+        parameters = parts.sublist(1);
+      }
+
+      var instance = pipe is String ? getContainer().make(pipe as Type) : pipe;
+
+      if (instance == null) {
+        throw Exception('Unable to resolve pipe: $pipe');
+      }
+
+      var method = instance.call(_method);
+      if (method == null) {
+        throw Exception('Method $_method not found on instance: $instance');
+      }
+
+      var result = Function.apply(
+        method,
+        [passable, next, ...parameters],
+      );
+
+      result = result is Future ? await result : result;
+      return handleCarry(result);
+    } catch (e) {
+      return handleException(passable, e);
+    }
+  }
+
+  /// Parse full pipe string to get name and parameters.
+  List<String> parsePipeString(String pipe) {
+    var parts = pipe.split(':');
+    return [parts[0], if (parts.length > 1) ...parts[1].split(',')];
+  }
+
+  /// Get the array of configured pipes.
+  List<dynamic> 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;
+  }
+
+  /// Handle the given exception.
+  dynamic handleException(dynamic passable, Object e) {
+    _logger.severe('Exception occurred in pipeline', e);
+    if (e is Exception) {
+      // You can add custom exception handling logic here
+      // For example, you might want to return a default value or transform the exception
+    }
+    throw e;
+  }
+}
+
+/// Extension methods for the Pipeline class.
+extension PipelineExtensions on Pipeline {
+  /// Add a logging pipe to the pipeline.
+  Pipeline addLoggingPipe() {
+    return pipe((passable, next) {
+      _logger.info('Pipe input: $passable');
+      var result = next(passable);
+      _logger.info('Pipe output: $result');
+      return result;
+    });
+  }
+
+  /// Add an asynchronous pipe to the pipeline.
+  Pipeline addAsyncPipe(Future<dynamic> Function(dynamic) asyncOperation) {
+    return pipe((passable, next) async {
+      var result = await asyncOperation(passable);
+      return next(result);
+    });
+  }
+
+  /// Add a validation pipe to the pipeline.
+  Pipeline addValidationPipe(bool Function(dynamic) validator,
+      {String? errorMessage}) {
+    return pipe((passable, next) {
+      if (!validator(passable)) {
+        throw Exception(errorMessage ?? 'Validation failed');
+      }
+      return next(passable);
+    });
+  }
+}

core/pipeline/pubspec.yaml

@@ -0,0 +1,18 @@
+name: angel3_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/platformo
+
+environment:
+  sdk: ^3.4.2
+
+# Add regular dependencies here.
+dependencies:
+  angel3_container: ^8.0.0
+  logging: ^1.1.0
+
+dev_dependencies:
+  lints: ^3.0.0
+  test: ^1.24.0

core/process/.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

core/process/CHANGELOG.md

@@ -0,0 +1,3 @@
+## 1.0.0
+
+- Initial version.

core/process/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.

core/process/README.md

@@ -0,0 +1 @@
+<p align="center"><a href="https://protevus.com" target="_blank"><img src="https://git.protevus.com/protevus/branding/raw/branch/main/protevus-logo-bg.png"></a></p>

core/process/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

core/process/example/basic_process/main.dart

@@ -0,0 +1,36 @@
+// examples/basic_process/main.dart
+import 'package:angel3_framework/angel3_framework.dart';
+import 'package:angel3_framework/http.dart';
+import 'package:angel3_process/angel3_process.dart';
+import 'package:logging/logging.dart';
+import 'package:angel3_container/mirrors.dart';
+
+void main() async {
+  Logger.root.level = Level.ALL;
+  Logger.root.onRecord.listen((record) {
+    print('${record.level.name}: ${record.time}: ${record.message}');
+  });
+
+  // Create an Angel application with MirrorsReflector
+  var app = Angel(reflector: MirrorsReflector());
+  var http = AngelHttp(app);
+
+  // Use dependency injection for ProcessManager
+  app.container.registerSingleton(ProcessManager());
+
+  app.get('/', (req, res) async {
+    // Use the ioc function to get the ProcessManager instance
+    var processManager = await req.container?.make<ProcessManager>();
+
+    var process = await processManager?.start(
+      'example_process',
+      'echo',
+      ['Hello, Angel3 Process!'],
+    );
+    var result = await process?.run();
+    res.writeln('Process output: ${result?.output.trim()}');
+  });
+
+  await http.startServer('localhost', 3000);
+  print('Server listening at http://localhost:3000');
+}

core/process/example/process_pipeline/main.dart

@@ -0,0 +1,37 @@
+// examples/process_pipeline/main.dart
+import 'package:angel3_framework/angel3_framework.dart';
+import 'package:angel3_framework/http.dart';
+import 'package:angel3_process/angel3_process.dart';
+import 'package:logging/logging.dart';
+import 'package:angel3_container/mirrors.dart';
+
+void main() async {
+  Logger.root.level = Level.ALL;
+  Logger.root.onRecord.listen((record) {
+    print('${record.level.name}: ${record.time}: ${record.message}');
+  });
+
+  // Create an Angel application with MirrorsReflector
+  var app = Angel(reflector: MirrorsReflector());
+  var http = AngelHttp(app);
+
+  // Register ProcessManager as a singleton in the container
+  app.container.registerSingleton(ProcessManager());
+
+  app.get('/', (req, res) async {
+    // Use dependency injection to get the ProcessManager instance
+    var processManager = await req.container?.make<ProcessManager>();
+
+    var processes = [
+      angel3Process('echo', ['Hello']),
+      angel3Process('sed', ['s/Hello/Greetings/']),
+      angel3Process('tr', ['[:lower:]', '[:upper:]']),
+    ];
+
+    var result = await processManager?.pipeline(processes);
+    res.writeln('Pipeline output: ${result?.output.trim()}');
+  });
+
+  await http.startServer('localhost', 3000);
+  print('Server listening at http://localhost:3000');
+}

core/process/example/process_pool/main.dart

@@ -0,0 +1,37 @@
+// examples/process_pool/main.dart
+import 'package:angel3_framework/angel3_framework.dart';
+import 'package:angel3_framework/http.dart';
+import 'package:angel3_process/angel3_process.dart';
+import 'package:logging/logging.dart';
+import 'package:angel3_container/mirrors.dart';
+
+void main() async {
+  Logger.root.level = Level.ALL;
+  Logger.root.onRecord.listen((record) {
+    print('${record.level.name}: ${record.time}: ${record.message}');
+  });
+
+  // Create an Angel application with MirrorsReflector
+  var app = Angel(reflector: MirrorsReflector());
+  var http = AngelHttp(app);
+
+  // Register ProcessManager as a singleton in the container
+  app.container.registerSingleton(ProcessManager());
+
+  app.get('/', (req, res) async {
+    // Use dependency injection to get the ProcessManager instance
+    var processManager = await req.container?.make<ProcessManager>();
+
+    var processes =
+        List.generate(5, (index) => angel3Process('echo', ['Process $index']));
+    var results = await processManager?.pool(processes, concurrency: 3);
+    var output = results
+        ?.map((result) =>
+            '${result.process.command} output: ${result.output.trim()}')
+        .join('\n');
+    res.write(output);
+  });
+
+  await http.startServer('localhost', 3000);
+  print('Server listening at http://localhost:3000');
+}

core/process/example/web_server_with_processes/main.dart

@@ -0,0 +1,67 @@
+// examples/web_server_with_processes/main.dart
+import 'package:angel3_framework/angel3_framework.dart';
+import 'package:angel3_framework/http.dart';
+import 'package:angel3_process/angel3_process.dart';
+import 'package:file/local.dart';
+import 'package:logging/logging.dart';
+import 'package:angel3_mustache/angel3_mustache.dart';
+import 'package:angel3_container/mirrors.dart';
+
+void main() async {
+  Logger.root.level = Level.ALL;
+  Logger.root.onRecord.listen((record) {
+    print('${record.level.name}: ${record.time}: ${record.message}');
+  });
+
+  // Create an Angel application with MirrorsReflector
+  var app = Angel(reflector: MirrorsReflector());
+  var http = AngelHttp(app);
+
+  // Register dependencies in the container
+  app.container.registerSingleton(const LocalFileSystem());
+  app.container.registerSingleton(ProcessManager());
+
+  // Set up the view renderer
+  var fs = await app.container.make<LocalFileSystem>();
+  var viewsDirectory = fs.directory('views');
+  await app.configure(mustache(viewsDirectory));
+
+  app.get('/', (req, res) async {
+    await res.render('index');
+  });
+
+  app.post('/run-process', (req, res) async {
+    var body = await req.bodyAsMap;
+    var command = body['command'] as String?;
+    var args = (body['args'] as String?)?.split(' ') ?? [];
+
+    if (command == null || command.isEmpty) {
+      throw AngelHttpException.badRequest(message: 'Command is required');
+    }
+
+    // Use dependency injection to get the ProcessManager instance
+    var processManager = await req.container?.make<ProcessManager>();
+
+    var process = await processManager?.start(
+      'user_process',
+      command,
+      args,
+    );
+    var result = await process?.run();
+
+    await res.json({
+      'output': result?.output.trim(),
+      'exitCode': result?.exitCode,
+    });
+  });
+
+  app.fallback((req, res) => throw AngelHttpException.notFound());
+
+  app.errorHandler = (e, req, res) {
+    res.writeln('Error: ${e.message}');
+    return false;
+  };
+
+  await http.startServer('localhost', 3000);
+  print('Server listening at http://localhost:3000');
+}

core/process/example/web_server_with_processes/views/index.mustache

@@ -0,0 +1,39 @@
+<!-- examples/web_server_with_processes/views/index.html -->
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Angel3 Process Example</title>
+</head>
+<body>
+    <h1>Run a Process</h1>
+    <form id="processForm">
+        <label for="command">Command:</label>
+        <input type="text" id="command" name="command" required>
+        <br>
+        <label for="args">Arguments:</label>
+        <input type="text" id="args" name="args">
+        <br>
+        <button type="submit">Run Process</button>
+    </form>
+    <div id="output"></div>
+
+    <script>
+        document.getElementById('processForm').addEventListener('submit', async (e) => {
+            e.preventDefault();
+            const formData = new FormData(e.target);
+            const response = await fetch('/run-process', {
+                method: 'POST',
+                body: formData
+            });
+            const result = await response.json();
+            document.getElementById('output').innerHTML = `
+                <h2>Output:</h2>
+                <pre>${result.output}</pre>
+                <p>Exit Code: ${result.exitCode}</p>
+            `;
+        });
+    </script>
+</body>
+</html>

core/process/lib/angel3_process.dart

@@ -0,0 +1,7 @@
+library;
+
+export 'src/process.dart';
+export 'src/process_helper.dart';
+export 'src/process_manager.dart';
+export 'src/process_pipeline.dart';
+export 'src/process_pool.dart';

core/process/lib/src/process.dart

@@ -0,0 +1,250 @@
+import 'dart:async';
+import 'dart:io';
+import 'dart:convert';
+
+// import 'package:angel3_framework/angel3_framework.dart';
+// import 'package:angel3_mq/mq.dart';
+// import 'package:angel3_reactivex/angel3_reactivex.dart';
+// import 'package:angel3_event_bus/event_bus.dart';
+import 'package:logging/logging.dart';
+
+class Angel3Process {
+  final String _command;
+  final List<String> _arguments;
+  final String? _workingDirectory;
+  final Map<String, String>? _environment;
+  final Duration? _timeout;
+  final bool _tty;
+  final bool _enableReadError;
+  final Logger _logger;
+
+  late final StreamController<List<int>> _outputController;
+  late final StreamController<List<int>> _errorController;
+  late final Completer<String> _outputCompleter;
+  late final Completer<String> _errorCompleter;
+  final Completer<String> _errorOutputCompleter = Completer<String>();
+  bool _isOutputComplete = false;
+  bool _isErrorComplete = false;
+
+  Process? _process;
+  DateTime? _startTime;
+  DateTime? _endTime;
+  bool _isDisposed = false;
+
+  Angel3Process(
+    this._command,
+    this._arguments, {
+    String? workingDirectory,
+    Map<String, String>? environment,
+    Duration? timeout,
+    bool tty = false,
+    bool enableReadError = true,
+    Logger? logger,
+  })  : _workingDirectory = workingDirectory,
+        _environment = environment,
+        _timeout = timeout,
+        _tty = tty,
+        _enableReadError = enableReadError,
+        _logger = logger ?? Logger('Angel3Process'),
+        _outputController = StreamController<List<int>>.broadcast(),
+        _errorController = StreamController<List<int>>.broadcast(),
+        _outputCompleter = Completer<String>(),
+        _errorCompleter = Completer<String>();
+
+  // Add this public getter
+  String get command => _command;
+  int? get pid => _process?.pid;
+  DateTime? get startTime => _startTime;
+  DateTime? get endTime => _endTime;
+
+  Stream<List<int>> get output => _outputController.stream;
+  Stream<List<int>> get errorOutput => _errorController.stream;
+
+  // Future<String> get outputAsString => _outputCompleter.future;
+  // Future<String> get errorOutputAsString => _errorCompleter.future;
+
+  Future<int> get exitCode => _process?.exitCode ?? Future.value(-1);
+  bool get isRunning => _process != null && !_process!.kill();
+
+  Future<Angel3Process> start() async {
+    if (_isDisposed) {
+      throw StateError('This process has been disposed and cannot be reused.');
+    }
+    _startTime = DateTime.now();
+
+    try {
+      _process = await Process.start(
+        _command,
+        _arguments,
+        workingDirectory: _workingDirectory,
+        environment: _environment,
+        runInShell: _tty,
+      );
+
+      _process!.stdout.listen(
+        (data) {
+          _outputController.add(data);
+        },
+        onDone: () {
+          if (!_isOutputComplete) {
+            _isOutputComplete = true;
+            _outputController.close();
+          }
+        },
+        onError: (error) {
+          _logger.severe('Error in stdout stream', error);
+          _outputController.addError(error);
+          if (!_isOutputComplete) {
+            _isOutputComplete = true;
+            _outputController.close();
+          }
+        },
+      );
+
+      var errorBuffer = StringBuffer();
+      _process!.stderr.listen(
+        (data) {
+          _errorController.add(data);
+          errorBuffer.write(utf8.decode(data));
+        },
+        onDone: () {
+          if (!_isErrorComplete) {
+            _isErrorComplete = true;
+            _errorController.close();
+            _errorOutputCompleter.complete(errorBuffer.toString());
+          }
+        },
+        onError: (error) {
+          _logger.severe('Error in stderr stream', error);
+          _errorController.addError(error);
+          if (!_isErrorComplete) {
+            _isErrorComplete = true;
+            _errorController.close();
+            _errorOutputCompleter.completeError(error);
+          }
+        },
+      );
+
+      _logger.info('Process started: $_command ${_arguments.join(' ')}');
+    } catch (e) {
+      _logger.severe('Failed to start process', e);
+      rethrow;
+    }
+    return this;
+  }
+
+  Future<ProcessResult> run() async {
+    await start();
+    if (_timeout != null) {
+      return await runWithTimeout(_timeout!);
+    }
+    final exitCode = await this.exitCode;
+    final output = await outputAsString;
+    final errorOutput = await _errorOutputCompleter.future;
+    _endTime = DateTime.now();
+    return ProcessResult(pid!, exitCode, output, errorOutput);
+  }
+
+  Future<ProcessResult> runWithTimeout(Duration timeout) async {
+    final exitCodeFuture = this.exitCode.timeout(timeout, onTimeout: () {
+      kill();
+      throw TimeoutException('Process timed out', timeout);
+    });
+
+    try {
+      final exitCode = await exitCodeFuture;
+      final output = await outputAsString;
+      final errorOutput = await _errorOutputCompleter.future;
+      _endTime = DateTime.now();
+      return ProcessResult(pid!, exitCode, output, errorOutput);
+    } catch (e) {
+      if (e is TimeoutException) {
+        throw e;
+      }
+      rethrow;
+    }
+  }
+
+  Future<void> write(String input) async {
+    if (_process != null) {
+      _process!.stdin.write(input);
+      await _process!.stdin.flush();
+    } else {
+      throw StateError('Process has not been started');
+    }
+  }
+
+  Future<void> writeLines(List<String> lines) async {
+    for (final line in lines) {
+      await write('$line\n');
+    }
+  }
+
+  Future<void> kill({ProcessSignal signal = ProcessSignal.sigterm}) async {
+    if (_process != null) {
+      _logger.info('Killing process with signal: ${signal.name}');
+      final result = _process!.kill(signal);
+      if (!result) {
+        _logger.warning('Failed to kill process with signal: ${signal.name}');
+      }
+    }
+  }
+
+  bool sendSignal(ProcessSignal signal) {
+    return _process?.kill(signal) ?? false;
+  }
+
+  Future<void> dispose() async {
+    if (!_isDisposed) {
+      _isDisposed = true;
+      await _outputController.close();
+      await _errorController.close();
+      if (!_outputCompleter.isCompleted) {
+        _outputCompleter.complete('');
+      }
+      if (!_errorCompleter.isCompleted) {
+        _errorCompleter.complete('');
+      }
+      await kill();
+      _logger.info('Process disposed: $_command ${_arguments.join(' ')}');
+    }
+  }
+
+  Future<String> get outputAsString async {
+    var buffer = await output.transform(utf8.decoder).join();
+    return buffer;
+  }
+
+  Future<String> get errorOutputAsString => _errorOutputCompleter.future;
+}
+
+class ProcessResult {
+  final int pid;
+  final int exitCode;
+  final String output;
+  final String errorOutput;
+
+  ProcessResult(this.pid, this.exitCode, this.output, this.errorOutput);
+
+  @override
+  String toString() {
+    return 'ProcessResult(pid: $pid, exitCode: $exitCode, output: ${output.length} chars, errorOutput: ${errorOutput.length} chars)';
+  }
+}
+
+class InvokedProcess {
+  final Angel3Process process;
+  final DateTime startTime;
+  final DateTime endTime;
+  final int exitCode;
+  final String output;
+  final String errorOutput;
+
+  InvokedProcess(this.process, this.startTime, this.endTime, this.exitCode,
+      this.output, this.errorOutput);
+
+  @override
+  String toString() {
+    return 'InvokedProcess(command: ${process._command}, arguments: ${process._arguments}, startTime: $startTime, endTime: $endTime, exitCode: $exitCode)';
+  }
+}

core/process/lib/src/process_helper.dart

@@ -0,0 +1,21 @@
+import 'process.dart';
+
+Angel3Process angel3Process(
+  String command,
+  List<String> arguments, {
+  String? workingDirectory,
+  Map<String, String>? environment,
+  Duration? timeout,
+  bool tty = false,
+  bool enableReadError = true,
+}) {
+  return Angel3Process(
+    command,
+    arguments,
+    workingDirectory: workingDirectory,
+    environment: environment,
+    timeout: timeout,
+    tty: tty,
+    enableReadError: enableReadError,
+  );
+}

core/process/lib/src/process_manager.dart

@@ -0,0 +1,148 @@
+import 'dart:async';
+import 'dart:io';
+
+// import 'package:angel3_framework/angel3_framework.dart';
+// import 'package:angel3_mq/mq.dart';
+// import 'package:angel3_reactivex/angel3_reactivex.dart';
+import 'package:angel3_event_bus/event_bus.dart';
+import 'package:logging/logging.dart';
+
+import 'process.dart';
+import 'process_pool.dart';
+import 'process_pipeline.dart';
+
+class ProcessManager {
+  final Map<String, Angel3Process> _processes = {};
+  final EventBus _eventBus = EventBus();
+  final List<StreamSubscription> _subscriptions = [];
+  final Logger _logger = Logger('ProcessManager');
+
+  Future<Angel3Process> start(
+    String id,
+    String command,
+    List<String> arguments, {
+    String? workingDirectory,
+    Map<String, String>? environment,
+    Duration? timeout,
+    bool tty = false,
+    bool enableReadError = true,
+  }) async {
+    if (_processes.containsKey(id)) {
+      throw Exception('Process with id $id already exists');
+    }
+
+    final process = Angel3Process(
+      command,
+      arguments,
+      workingDirectory: workingDirectory,
+      environment: environment,
+      timeout: timeout,
+      tty: tty,
+      enableReadError: enableReadError,
+      logger: Logger('Angel3Process:$id'),
+    );
+
+    try {
+      await process.start();
+      _processes[id] = process;
+
+      _eventBus.fire(ProcessStartedEvent(id, process) as AppEvent);
+
+      process.exitCode.then((exitCode) {
+        _eventBus.fire(ProcessExitedEvent(id, exitCode) as AppEvent);
+        _processes.remove(id);
+      });
+
+      _logger.info('Started process with id: $id');
+      return process;
+    } catch (e) {
+      _logger.severe('Failed to start process with id: $id', e);
+      rethrow;
+    }
+  }
+
+  Angel3Process? get(String id) => _processes[id];
+
+  Future<void> kill(String id,
+      {ProcessSignal signal = ProcessSignal.sigterm}) async {
+    final process = _processes[id];
+    if (process != null) {
+      await process.kill(signal: signal);
+      _processes.remove(id);
+      _logger.info('Killed process with id: $id');
+    } else {
+      _logger.warning('Attempted to kill non-existent process with id: $id');
+    }
+  }
+
+  Future<void> killAll({ProcessSignal signal = ProcessSignal.sigterm}) async {
+    _logger.info('Killing all processes');
+    await Future.wait(
+        _processes.values.map((process) => process.kill(signal: signal)));
+    _processes.clear();
+  }
+
+  Stream<ProcessEvent> get events => _eventBus.on<ProcessEvent>();
+
+  Future<List<InvokedProcess>> pool(List<Angel3Process> processes,
+      {int concurrency = 5}) async {
+    _logger.info('Running process pool with concurrency: $concurrency');
+    final pool = ProcessPool(concurrency: concurrency);
+    return await pool.run(processes);
+  }
+
+  Future<InvokedProcess> pipeline(List<Angel3Process> processes) async {
+    _logger.info('Running process pipeline');
+    final pipeline = ProcessPipeline(processes);
+    return await pipeline.run();
+  }
+
+  void dispose() {
+    _logger.info('Disposing ProcessManager');
+
+    // Cancel all event subscriptions
+    for (var subscription in _subscriptions) {
+      subscription.cancel();
+    }
+    _subscriptions.clear();
+
+    // Dispose all processes
+    for (var process in _processes.values) {
+      process.dispose();
+    }
+    _processes.clear();
+
+    _logger.info('ProcessManager disposed');
+  }
+}
+
+abstract class ProcessEvent extends AppEvent {}
+
+class ProcessStartedEvent extends ProcessEvent {
+  final String id;
+  final Angel3Process process;
+
+  ProcessStartedEvent(this.id, this.process);
+
+  @override
+  String toString() =>
+      'ProcessStartedEvent(id: $id, command: ${process.command})';
+
+  @override
+  // TODO: implement props
+  List<Object?> get props => throw UnimplementedError();
+}
+
+class ProcessExitedEvent extends ProcessEvent {
+  final String id;
+  final int exitCode;
+
+  ProcessExitedEvent(this.id, this.exitCode);
+
+  @override
+  String toString() => 'ProcessExitedEvent(id: $id, exitCode: $exitCode)';
+
+  @override
+  // TODO: implement props
+  List<Object?> get props => throw UnimplementedError();
+}

core/process/lib/src/process_pipeline.dart

@@ -0,0 +1,50 @@
+import 'dart:async';
+import 'package:logging/logging.dart';
+import 'process.dart';
+
+class ProcessPipeline {
+  final List<Angel3Process> _processes;
+  final Logger _logger = Logger('ProcessPipeline');
+
+  ProcessPipeline(this._processes);
+
+  Future<InvokedProcess> run() async {
+    String input = '';
+    DateTime startTime = DateTime.now();
+    DateTime endTime;
+    int lastExitCode = 0;
+
+    _logger
+        .info('Starting process pipeline with ${_processes.length} processes');
+
+    for (final process in _processes) {
+      _logger.info('Running process: ${process.command}');
+      if (input.isNotEmpty) {
+        await process.write(input);
+      }
+      final result = await process.run();
+      input = result.output;
+      lastExitCode = result.exitCode;
+      _logger.info(
+          'Process completed: ${process.command} with exit code $lastExitCode');
+      if (lastExitCode != 0) {
+        _logger.warning(
+            'Pipeline stopped due to non-zero exit code: $lastExitCode');
+        break;
+      }
+    }
+
+    endTime = DateTime.now();
+    _logger.info(
+        'Pipeline completed. Total duration: ${endTime.difference(startTime)}');
+
+    return InvokedProcess(
+      _processes.last,
+      startTime,
+      endTime,
+      lastExitCode,
+      input,
+      '',
+    );
+  }
+}

core/process/lib/src/process_pool.dart

@@ -0,0 +1,62 @@
+import 'dart:async';
+import 'package:logging/logging.dart';
+import 'process.dart';
+
+class ProcessPool {
+  final int concurrency;
+  final List<Function> _queue = [];
+  int _running = 0;
+  final Logger _logger = Logger('ProcessPool');
+
+  ProcessPool({this.concurrency = 5});
+
+  Future<List<InvokedProcess>> run(List<Angel3Process> processes) async {
+    final results = <InvokedProcess>[];
+    final completer = Completer<List<InvokedProcess>>();
+
+    _logger.info('Starting process pool with ${processes.length} processes');
+
+    for (final process in processes) {
+      _queue.add(() async {
+        try {
+          final result = await _runProcess(process);
+          results.add(result);
+        } catch (e) {
+          _logger.severe('Error running process in pool', e);
+        } finally {
+          _running--;
+          _processQueue();
+          if (_running == 0 && _queue.isEmpty) {
+            completer.complete(results);
+          }
+        }
+      });
+    }
+
+    _processQueue();
+
+    return completer.future;
+  }
+
+  void _processQueue() {
+    while (_running < concurrency && _queue.isNotEmpty) {
+      _running++;
+      _queue.removeAt(0)();
+    }
+  }
+
+  Future<InvokedProcess> _runProcess(Angel3Process process) async {
+    _logger.info('Running process: ${process.command}');
+    final result = await process.run();
+    _logger.info(
+        'Process completed: ${process.command} with exit code ${result.exitCode}');
+    return InvokedProcess(
+      process,
+      process.startTime!,
+      process.endTime!,
+      result.exitCode,
+      result.output,
+      result.errorOutput,
+    );
+  }
+}

core/process/lib/src/process_service_provider.dart

@@ -0,0 +1,24 @@
+/* import 'package:angel3_framework/angel3_framework.dart';
+import 'package:logging/logging.dart';
+import 'process_manager.dart';
+
+class ProcessServiceProvider extends Provider {
+  final Logger _logger = Logger('ProcessServiceProvider');
+
+  @override
+  void registers() {
+    container.singleton<ProcessManager>((_) => ProcessManager());
+    _logger.info('Registered ProcessManager');
+  }
+
+  @override
+  void boots(Angel app) {
+    app.shutdownHooks.add((_) async {
+      _logger.info('Shutting down ProcessManager');
+      final processManager = app.container.make<ProcessManager>();
+      await processManager.killAll();
+      processManager.dispose();
+    });
+    _logger.info('Added ProcessManager shutdown hook');
+  }
+} */

core/process/pubspec.yaml

@@ -0,0 +1,25 @@
+name: angel3_process
+description: The Process Package for the Protevus Platform
+version: 0.0.1
+homepage: https://protevus.com
+documentation: https://docs.protevus.com
+repository: https://github.com/protevus/platformo
+
+environment:
+  sdk: ^3.4.2
+
+# Add regular dependencies here.
+dependencies:
+  angel3_container: ^8.0.0
+  angel3_framework: ^8.0.0
+  angel3_mq: ^8.0.0
+  angel3_mustache: ^8.0.0
+  angel3_event_bus: ^8.0.0
+  angel3_reactivex: ^8.0.0
+  file: ^7.0.0
+  logging: ^1.1.0
+  path: ^1.8.0
+
+dev_dependencies:
+  lints: ^3.0.0
+  test: ^1.24.0

core/process/test/process_test.dart

@@ -0,0 +1,80 @@
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+import 'package:angel3_process/angel3_process.dart';
+import 'package:test/test.dart';
+
+void main() {
+  late Angel3Process process;
+
+  setUp(() {
+    process = Angel3Process('echo', ['Hello, World!']);
+  });
+
+  tearDown(() async {
+    await process.dispose();
+  });
+
+  test('Angel3Process initialization', () {
+    expect(process.command, equals('echo'));
+    expect(process.startTime, isNull);
+    expect(process.endTime, isNull);
+  });
+
+  test('Start and run a simple process', () async {
+    var result = await process.run();
+    expect(process.startTime, isNotNull);
+    expect(result.exitCode, equals(0));
+    expect(result.output.trim(), equals('Hello, World!'));
+    expect(process.endTime, isNotNull);
+  });
+
+  test('Stream output', () async {
+    await process.start();
+    var outputStream = process.output.transform(utf8.decoder);
+    var streamOutput = await outputStream.join();
+    await process.exitCode; // Wait for the process to complete
+    expect(streamOutput.trim(), equals('Hello, World!'));
+  });
+
+  test('Error output for non-existent command', () {
+    var errorProcess = Angel3Process('non_existent_command', []);
+    expect(errorProcess.start(), throwsA(isA<ProcessException>()));
+  });
+
+  test('Process with error output', () async {
+    Angel3Process errorProcess;
+    if (Platform.isWindows) {
+      errorProcess = Angel3Process('cmd', ['/c', 'dir', '/invalid_argument']);
+    } else {
+      errorProcess = Angel3Process('ls', ['/non_existent_directory']);
+    }
+
+    print('Starting error process...');
+    var result = await errorProcess.run();
+    print('Error process completed.');
+    print('Exit code: ${result.exitCode}');
+    print('Standard output: "${result.output}"');
+    print('Error output: "${result.errorOutput}"');
+
+    expect(result.exitCode, isNot(0), reason: 'Expected non-zero exit code');
+    expect(result.errorOutput.trim(), isNotEmpty,
+        reason: 'Expected non-empty error output');
+
+    await errorProcess.dispose();
+  });
+
+  test('Kill running process', () async {
+    var longRunningProcess = Angel3Process('sleep', ['5']);
+    await longRunningProcess.start();
+    await longRunningProcess.kill();
+    var exitCode = await longRunningProcess.exitCode;
+    expect(exitCode, isNot(0));
+  });
+
+  test('Process timeout', () async {
+    var timeoutProcess =
+        Angel3Process('sleep', ['10'], timeout: Duration(seconds: 1));
+    expect(() => timeoutProcess.run(), throwsA(isA<TimeoutException>()));
+  }, timeout: Timeout(Duration(seconds: 5)));
+}

core/process/test/process_test_extended.dart

@@ -0,0 +1,103 @@
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io' show Directory, Platform, ProcessSignal;
+import 'package:angel3_process/angel3_process.dart';
+import 'package:test/test.dart';
+import 'package:path/path.dart' as path;
+
+void main() {
+  late Angel3Process process;
+
+  setUp(() {
+    process = Angel3Process('echo', ['Hello, World!']);
+  });
+
+  tearDown(() async {
+    await process.dispose();
+  });
+
+  // ... (existing tests remain the same)
+
+  test('Process with custom environment variables', () async {
+    var command = Platform.isWindows ? 'cmd' : 'sh';
+    var args = Platform.isWindows
+        ? ['/c', 'echo %TEST_VAR%']
+        : ['-c', r'echo $TEST_VAR']; // Use a raw string for Unix-like systems
+
+    var envProcess =
+        Angel3Process(command, args, environment: {'TEST_VAR': 'custom_value'});
+
+    var result = await envProcess.run();
+    expect(result.output.trim(), equals('custom_value'));
+  });
+
+  test('Process with custom working directory', () async {
+    var tempDir = Directory.systemTemp.createTempSync();
+    try {
+      var workingDirProcess = Angel3Process(Platform.isWindows ? 'cmd' : 'pwd',
+          Platform.isWindows ? ['/c', 'cd'] : [],
+          workingDirectory: tempDir.path);
+      var result = await workingDirProcess.run();
+      expect(path.equals(result.output.trim(), tempDir.path), isTrue);
+    } finally {
+      tempDir.deleteSync();
+    }
+  });
+
+  test('Process with input', () async {
+    var catProcess = Angel3Process('cat', []);
+    await catProcess.start();
+    catProcess.write('Hello, stdin!');
+    await catProcess.kill(); // End the process
+    var output = await catProcess.outputAsString;
+    expect(output.trim(), equals('Hello, stdin!'));
+  });
+
+  test('Longer-running process', () async {
+    var sleepProcess = Angel3Process(Platform.isWindows ? 'timeout' : 'sleep',
+        Platform.isWindows ? ['/t', '2'] : ['2']);
+    var startTime = DateTime.now();
+    await sleepProcess.run();
+    var endTime = DateTime.now();
+    expect(endTime.difference(startTime).inSeconds, greaterThanOrEqualTo(2));
+  });
+
+  test('Multiple concurrent processes', () async {
+    var processes =
+        List.generate(5, (_) => Angel3Process('echo', ['concurrent']));
+    var results = await Future.wait(processes.map((p) => p.run()));
+    for (var result in results) {
+      expect(result.output.trim(), equals('concurrent'));
+    }
+  });
+
+  test('Process signaling', () async {
+    if (!Platform.isWindows) {
+      // SIGSTOP/SIGCONT are not available on Windows
+      var longProcess = Angel3Process('sleep', ['10']);
+      await longProcess.start();
+      await longProcess.sendSignal(ProcessSignal.sigstop);
+      // Process should be stopped, so it shouldn't complete immediately
+      expect(longProcess.exitCode, doesNotComplete);
+      await longProcess.sendSignal(ProcessSignal.sigcont);
+      await longProcess.kill();
+      expect(await longProcess.exitCode, isNot(0));
+    }
+  });
+
+  test('Edge case: empty command', () {
+    expect(() => Angel3Process('', []), throwsA(isA<ArgumentError>()));
+  });
+
+  test('Edge case: empty arguments list', () {
+    // This should not throw an error
+    expect(() => Angel3Process('echo', []), returnsNormally);
+  });
+
+  test('Edge case: invalid argument type', () {
+    // This should throw a compile-time error, but we can't test for that directly
+    // Instead, we can test for runtime type checking if implemented
+    expect(() => Angel3Process('echo', [1, 2, 3] as dynamic),
+        throwsA(isA<ArgumentError>()));
+  });
+}