diff --git a/core/process/.gitignore b/core/process/.gitignore new file mode 100644 index 00000000..3cceda55 --- /dev/null +++ b/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 diff --git a/core/process/CHANGELOG.md b/core/process/CHANGELOG.md new file mode 100644 index 00000000..effe43c8 --- /dev/null +++ b/core/process/CHANGELOG.md @@ -0,0 +1,3 @@ +## 1.0.0 + +- Initial version. diff --git a/core/process/LICENSE.md b/core/process/LICENSE.md new file mode 100644 index 00000000..0fd0d03b --- /dev/null +++ b/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. \ No newline at end of file diff --git a/core/process/README.md b/core/process/README.md new file mode 100644 index 00000000..757f4c9f --- /dev/null +++ b/core/process/README.md @@ -0,0 +1 @@ +

\ No newline at end of file diff --git a/core/process/analysis_options.yaml b/core/process/analysis_options.yaml new file mode 100644 index 00000000..dee8927a --- /dev/null +++ b/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 diff --git a/core/process/example/basic_process/main.dart b/core/process/example/basic_process/main.dart new file mode 100644 index 00000000..3d425895 --- /dev/null +++ b/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(); + + 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'); +} diff --git a/core/process/example/process_pipeline/main.dart b/core/process/example/process_pipeline/main.dart new file mode 100644 index 00000000..29760508 --- /dev/null +++ b/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(); + + 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'); +} diff --git a/core/process/example/process_pool/main.dart b/core/process/example/process_pool/main.dart new file mode 100644 index 00000000..e88f3a31 --- /dev/null +++ b/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(); + + 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'); +} diff --git a/core/process/example/web_server_with_processes/main.dart b/core/process/example/web_server_with_processes/main.dart new file mode 100644 index 00000000..754e0284 --- /dev/null +++ b/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(); + 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(); + + 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'); +} diff --git a/core/process/example/web_server_with_processes/views/index.mustache b/core/process/example/web_server_with_processes/views/index.mustache new file mode 100644 index 00000000..cb292441 --- /dev/null +++ b/core/process/example/web_server_with_processes/views/index.mustache @@ -0,0 +1,39 @@ + + + + + + + Angel3 Process Example + + +

Run a Process

+
+ + +
+ + +
+ +
+
+ + + + diff --git a/core/process/lib/angel3_process.dart b/core/process/lib/angel3_process.dart new file mode 100644 index 00000000..1965d4a2 --- /dev/null +++ b/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'; diff --git a/core/process/lib/src/process.dart b/core/process/lib/src/process.dart new file mode 100644 index 00000000..4049683f --- /dev/null +++ b/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 _arguments; + final String? _workingDirectory; + final Map? _environment; + final Duration? _timeout; + final bool _tty; + final bool _enableReadError; + final Logger _logger; + + late final StreamController> _outputController; + late final StreamController> _errorController; + late final Completer _outputCompleter; + late final Completer _errorCompleter; + final Completer _errorOutputCompleter = Completer(); + bool _isOutputComplete = false; + bool _isErrorComplete = false; + + Process? _process; + DateTime? _startTime; + DateTime? _endTime; + bool _isDisposed = false; + + Angel3Process( + this._command, + this._arguments, { + String? workingDirectory, + Map? 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>.broadcast(), + _errorController = StreamController>.broadcast(), + _outputCompleter = Completer(), + _errorCompleter = Completer(); + + // Add this public getter + String get command => _command; + int? get pid => _process?.pid; + DateTime? get startTime => _startTime; + DateTime? get endTime => _endTime; + + Stream> get output => _outputController.stream; + Stream> get errorOutput => _errorController.stream; + + // Future get outputAsString => _outputCompleter.future; + // Future get errorOutputAsString => _errorCompleter.future; + + Future get exitCode => _process?.exitCode ?? Future.value(-1); + bool get isRunning => _process != null && !_process!.kill(); + + Future 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 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 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 write(String input) async { + if (_process != null) { + _process!.stdin.write(input); + await _process!.stdin.flush(); + } else { + throw StateError('Process has not been started'); + } + } + + Future writeLines(List lines) async { + for (final line in lines) { + await write('$line\n'); + } + } + + Future 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 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 get outputAsString async { + var buffer = await output.transform(utf8.decoder).join(); + return buffer; + } + + Future 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)'; + } +} diff --git a/core/process/lib/src/process_helper.dart b/core/process/lib/src/process_helper.dart new file mode 100644 index 00000000..5166698e --- /dev/null +++ b/core/process/lib/src/process_helper.dart @@ -0,0 +1,21 @@ +import 'process.dart'; + +Angel3Process angel3Process( + String command, + List arguments, { + String? workingDirectory, + Map? environment, + Duration? timeout, + bool tty = false, + bool enableReadError = true, +}) { + return Angel3Process( + command, + arguments, + workingDirectory: workingDirectory, + environment: environment, + timeout: timeout, + tty: tty, + enableReadError: enableReadError, + ); +} diff --git a/core/process/lib/src/process_manager.dart b/core/process/lib/src/process_manager.dart new file mode 100644 index 00000000..f2d48b6e --- /dev/null +++ b/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 _processes = {}; + final EventBus _eventBus = EventBus(); + final List _subscriptions = []; + final Logger _logger = Logger('ProcessManager'); + + Future start( + String id, + String command, + List arguments, { + String? workingDirectory, + Map? 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 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 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 get events => _eventBus.on(); + + Future> pool(List processes, + {int concurrency = 5}) async { + _logger.info('Running process pool with concurrency: $concurrency'); + final pool = ProcessPool(concurrency: concurrency); + return await pool.run(processes); + } + + Future pipeline(List 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 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 get props => throw UnimplementedError(); +} diff --git a/core/process/lib/src/process_pipeline.dart b/core/process/lib/src/process_pipeline.dart new file mode 100644 index 00000000..449c7087 --- /dev/null +++ b/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 _processes; + final Logger _logger = Logger('ProcessPipeline'); + + ProcessPipeline(this._processes); + + Future 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, + '', + ); + } +} diff --git a/core/process/lib/src/process_pool.dart b/core/process/lib/src/process_pool.dart new file mode 100644 index 00000000..67b323ff --- /dev/null +++ b/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 _queue = []; + int _running = 0; + final Logger _logger = Logger('ProcessPool'); + + ProcessPool({this.concurrency = 5}); + + Future> run(List processes) async { + final results = []; + final completer = Completer>(); + + _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 _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, + ); + } +} diff --git a/core/process/lib/src/process_service_provider.dart b/core/process/lib/src/process_service_provider.dart new file mode 100644 index 00000000..c762c94e --- /dev/null +++ b/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()); + _logger.info('Registered ProcessManager'); + } + + @override + void boots(Angel app) { + app.shutdownHooks.add((_) async { + _logger.info('Shutting down ProcessManager'); + final processManager = app.container.make(); + await processManager.killAll(); + processManager.dispose(); + }); + _logger.info('Added ProcessManager shutdown hook'); + } +} */ diff --git a/core/process/pubspec.yaml b/core/process/pubspec.yaml new file mode 100644 index 00000000..cd526f58 --- /dev/null +++ b/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 diff --git a/core/process/test/process_test.dart b/core/process/test/process_test.dart new file mode 100644 index 00000000..f737d084 --- /dev/null +++ b/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())); + }); + + 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())); + }, timeout: Timeout(Duration(seconds: 5))); +} diff --git a/core/process/test/process_test_extended.dart b/core/process/test/process_test_extended.dart new file mode 100644 index 00000000..1dffc1c0 --- /dev/null +++ b/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())); + }); + + 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())); + }); +}