mrpastewart/platform
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Remove: deleting files moved to wspace

Browse source
This commit is contained in:
Patrick Stewart 2024-10-11 00:29:38 -07:00
parent 2153827523
commit 8d2f32b3e8
99 changed files with 0 additions and 5455 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
core/broadcasting/.gitignore vendored
View file

@ -1,7 +0,0 @@
# 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

3
core/broadcasting/CHANGELOG.md
View file

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

10
core/broadcasting/LICENSE.md
View file

@ -1,10 +0,0 @@
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.

1
core/broadcasting/README.md
View file

@ -1 +0,0 @@
<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>

30
core/broadcasting/analysis_options.yaml
View file

@ -1,30 +0,0 @@
# 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

17
core/broadcasting/pubspec.yaml
View file

@ -1,17 +0,0 @@
name: angel3_broadcasting
description: The Broadcasting 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:
# path: ^1.8.0
dev_dependencies:
lints: ^3.0.0
test: ^1.24.0

0
core/broadcasting/test/.gitkeep
View file

7
core/bus/.gitignore vendored
View file

@ -1,7 +0,0 @@
# 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

3
core/bus/CHANGELOG.md
View file

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

10
core/bus/LICENSE.md
View file

@ -1,10 +0,0 @@
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.

1
core/bus/README.md
View file

@ -1 +0,0 @@
<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>

30
core/bus/analysis_options.yaml
View file

@ -1,30 +0,0 @@
# 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

9
core/bus/lib/angel3_bus.dart
View file

@ -1,9 +0,0 @@
library angel3_bus;
export 'src/dispatcher.dart';
export 'src/command.dart';
export 'src/handler.dart';
export 'src/queue.dart';
export 'src/batch.dart';
export 'src/chain.dart';
export 'src/bus_service_provider.dart';

19
core/bus/lib/src/batch.dart
View file

@ -1,19 +0,0 @@
import 'command.dart';
import 'dispatcher.dart';
class Batch {
// Implement Batch
}
class PendingBatch {
final Dispatcher _dispatcher;
final List<Command> _commands;
PendingBatch(this._dispatcher, this._commands);
Future<void> dispatch() async {
for (var command in _commands) {
await _dispatcher.dispatch(command);
}
}
}

60
core/bus/lib/src/bus_service_provider.dart
View file

@ -1,60 +0,0 @@
// // lib/src/bus_service_provider.dart
// import 'package:angel3_framework/angel3_framework.dart';
// import 'package:angel3_event_bus/angel3_event_bus.dart';
// import 'package:angel3_mq/angel3_mq.dart';
// import 'dispatcher.dart';
// class BusServiceProvider extends Provider {
// @override
// Future<void> boot(Angel app) async {
// // Register EventBus
// app.container.registerSingleton<EventBus>(EventBus());
// // Register Queue
// app.container.registerSingleton<Queue>(MemoryQueue());
// // Create and register the Dispatcher
// final dispatcher = Dispatcher(app.container);
// app.container.registerSingleton<Dispatcher>(dispatcher);
// // Register any global middleware or mappings
// dispatcher.pipeThrough([
// // Add any global middleware here
// ]);
// // Register command-to-handler mappings
// dispatcher.map({
// // Add your command-to-handler mappings here
// // Example: ExampleCommand: ExampleCommandHandler,
// });
// }
// }
// class MemoryQueue implements Queue {
// final List<Command> _queue = [];
// @override
// Future<void> push(Command command) async {
// _queue.add(command);
// }
// @override
// Future<void> later(Duration delay, Command command) async {
// await Future.delayed(delay);
// _queue.add(command);
// }
// @override
// Future<void> pushOn(String queue, Command command) async {
// // For simplicity, ignoring the queue parameter in this implementation
// _queue.add(command);
// }
// @override
// Future<void> laterOn(String queue, Duration delay, Command command) async {
// // For simplicity, ignoring the queue parameter in this implementation
// await Future.delayed(delay);
// _queue.add(command);
// }
// }

15
core/bus/lib/src/chain.dart
View file

@ -1,15 +0,0 @@
import 'command.dart';
import 'dispatcher.dart';
class PendingChain {
final Dispatcher _dispatcher;
final List<Command> _commands;
PendingChain(this._dispatcher, this._commands);
Future<void> dispatch() async {
for (var command in _commands) {
await _dispatcher.dispatch(command);
}
}
}

5
core/bus/lib/src/command.dart
View file

@ -1,5 +0,0 @@
// lib/src/command.dart
abstract class Command {}
abstract class ShouldQueue implements Command {}

251
core/bus/lib/src/dispatcher.dart
View file

@ -1,251 +0,0 @@
// lib/src/dispatcher.dart
import 'dart:async';
import 'package:angel3_container/angel3_container.dart';
import 'package:angel3_reactivex/angel3_reactivex.dart';
import 'package:angel3_event_bus/event_bus.dart';
import 'package:angel3_mq/mq.dart';
import 'command.dart';
import 'handler.dart';
import 'batch.dart';
import 'chain.dart';
/// A class that handles dispatching and processing of commands.
///
/// This dispatcher supports both synchronous and asynchronous command execution,
/// as well as queueing commands for later processing.
class Dispatcher implements QueueingDispatcher {
final Container container;
final EventBus _eventBus;
final Subject<Command> _commandSubject;
final MQClient _queue;
final Map<Type, Type> _handlers = {};
/// Creates a new [Dispatcher] instance.
///
/// [container] is used for dependency injection and to retrieve necessary services.
Dispatcher(this.container)
: _eventBus = container.make<EventBus>(),
_commandSubject = BehaviorSubject<Command>(),
_queue = container.make<MQClient>() {
_setupCommandProcessing();
}
/// Sets up the command processing pipeline.
///
/// This method initializes the stream that processes commands and emits events.
void _setupCommandProcessing() {
_commandSubject
.flatMap((command) => Stream.fromFuture(_processCommand(command))
.map((result) => CommandEvent(command, result: result))
.onErrorReturnWith(
(error, stackTrace) => CommandEvent(command, error: error)))
.listen((event) {
_eventBus.fire(event);
});
}
/// Dispatches a command for execution.
///
/// If the command implements [ShouldQueue], it will be dispatched to a queue.
/// Otherwise, it will be executed immediately.
///
/// [command] is the command to be dispatched.
@override
Future<dynamic> dispatch(Command command) {
if (command is ShouldQueue) {
return dispatchToQueue(command);
} else {
return dispatchNow(command);
}
}
/// Dispatches a command for immediate execution.
///
/// [command] is the command to be executed.
/// [handler] is an optional specific handler for the command.
@override
Future<dynamic> dispatchNow(Command command, [Handler? handler]) {
final completer = Completer<dynamic>();
_commandSubject.add(command);
_eventBus
.on<CommandEvent>()
.where((event) => event.command == command)
.take(1)
.listen((event) {
if (event.error != null) {
completer.completeError(event.error);
} else {
completer.complete(event.result);
}
});
return completer.future;
}
/// Processes a command by finding and executing its appropriate handler.
///
/// [command] is the command to be processed.
Future<dynamic> _processCommand(Command command) async {
final handlerType = _handlers[command.runtimeType];
if (handlerType != null) {
final handler = container.make(handlerType) as Handler;
return await handler.handle(command);
} else {
throw Exception('No handler found for command: ${command.runtimeType}');
}
}
/// Dispatches a command to a queue for later processing.
///
/// [command] is the command to be queued.
@override
Future<dynamic> dispatchToQueue(Command command) async {
final message = Message(
payload: command,
headers: {
'commandType': command.runtimeType.toString(),
},
);
_queue.sendMessage(
message: message,
// You might want to specify an exchange name and routing key if needed
// exchangeName: 'your_exchange_name',
// routingKey: 'your_routing_key',
);
return message.id;
}
/// Dispatches a command synchronously.
///
/// This is an alias for [dispatchNow].
///
/// [command] is the command to be executed.
/// [handler] is an optional specific handler for the command.
@override
Future<dynamic> dispatchSync(Command command, [Handler? handler]) {
return dispatchNow(command, handler);
}
/// Finds a batch by its ID.
///
/// [batchId] is the ID of the batch to find.
@override
Future<Batch?> findBatch(String batchId) async {
// Implement batch finding logic
throw UnimplementedError();
}
/// Creates a new pending batch of commands.
///
/// [commands] is the list of commands to be included in the batch.
@override
PendingBatch batch(List<Command> commands) {
return PendingBatch(this, commands);
}
/// Creates a new pending chain of commands.
///
/// [commands] is the list of commands to be included in the chain.
@override
PendingChain chain(List<Command> commands) {
return PendingChain(this, commands);
}
/// Applies a list of pipes to the command processing pipeline.
///
/// [pipes] is the list of pipes to be applied.
@override
Dispatcher pipeThrough(List<Pipe> pipes) {
_commandSubject.transform(
StreamTransformer.fromHandlers(
handleData: (data, sink) {
var result = data;
for (var pipe in pipes) {
result = pipe(result);
}
sink.add(result);
},
),
);
return this;
}
/// Maps command types to their respective handler types.
///
/// [handlers] is a map where keys are command types and values are handler types.
@override
Dispatcher map(Map<Type, Type> handlers) {
_handlers.addAll(handlers);
return this;
}
/// Dispatches a command to be executed after the current request-response cycle.
///
/// [command] is the command to be dispatched after the response.
@override
void dispatchAfterResponse(Command command) {
final message = Message(
payload: command,
headers: {
'commandType': command.runtimeType.toString(),
'dispatchAfterResponse': 'true',
},
);
_queue.sendMessage(
message: message,
// You might want to specify an exchange name if needed
// exchangeName: 'your_exchange_name',
// If you want to use a specific queue for after-response commands:
routingKey: 'after_response_queue',
);
}
}
abstract class QueueingDispatcher {
Future<dynamic> dispatch(Command command);
Future<dynamic> dispatchSync(Command command, [Handler? handler]);
Future<dynamic> dispatchNow(Command command, [Handler? handler]);
Future<dynamic> dispatchToQueue(Command command);
Future<Batch?> findBatch(String batchId);
PendingBatch batch(List<Command> commands);
PendingChain chain(List<Command> commands);
Dispatcher pipeThrough(List<Pipe> pipes);
Dispatcher map(Map<Type, Type> handlers);
void dispatchAfterResponse(Command command);
}
typedef Pipe = Command Function(Command);
class CommandCompletedEvent extends AppEvent {
final dynamic result;
CommandCompletedEvent(this.result);
@override
List<Object?> get props => [result];
}
class CommandErrorEvent extends AppEvent {
final dynamic error;
CommandErrorEvent(this.error);
@override
List<Object?> get props => [error];
}
class CommandEvent extends AppEvent {
final Command command;
final dynamic result;
final dynamic error;
CommandEvent(this.command, {this.result, this.error});
@override
List<Object?> get props => [command, result, error];
}

5
core/bus/lib/src/handler.dart
View file

@ -1,5 +0,0 @@
import 'command.dart';
abstract class Handler {
Future<dynamic> handle(Command command);
}

8
core/bus/lib/src/queue.dart
View file

@ -1,8 +0,0 @@
import 'command.dart';
abstract class Queue {
Future<void> push(Command command);
Future<void> later(Duration delay, Command command);
Future<void> pushOn(String queue, Command command);
Future<void> laterOn(String queue, Duration delay, Command command);
}

24
core/bus/pubspec.yaml
View file

@ -1,24 +0,0 @@
name: angel3_bus
description: The Queue 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: ^9.0.0
angel3_framework: ^9.0.0
angel3_reactivex: ^9.0.0
angel3_event_bus: ^9.0.0
angel3_mq: ^9.0.0
# path: ^1.8.0
dev_dependencies:
build_runner: ^2.1.0
lints: ^3.0.0
mockito: ^5.3.0
test: ^1.24.0

197
core/bus/test/dispatcher_test.dart
View file

@ -1,197 +0,0 @@
import 'dart:async';
import 'package:angel3_bus/angel3_bus.dart';
import 'package:angel3_container/angel3_container.dart';
import 'package:angel3_event_bus/event_bus.dart';
import 'package:angel3_mq/mq.dart';
import 'package:mockito/mockito.dart';
import 'package:test/test.dart';
class IsMessage extends Matcher {
@override
bool matches(item, Map matchState) => item is Message;
@override
Description describe(Description description) =>
description.add('is a Message');
}
class MockContainer extends Mock implements Container {
final Map<Type, dynamic> _instances = {};
@override
T make<T>([Type? type]) {
type ??= T;
return _instances[type] as T;
}
void registerInstance<T>(T instance) {
_instances[T] = instance;
}
}
class MockEventBus extends Mock implements EventBus {
@override
Stream<T> on<T extends AppEvent>() {
return super.noSuchMethod(
Invocation.method(#on, [], {#T: T}),
returnValue: Stream<T>.empty(),
) as Stream<T>;
}
}
class MockMQClient extends Mock implements MQClient {
Message? capturedMessage;
String? capturedExchangeName;
String? capturedRoutingKey;
@override
dynamic noSuchMethod(Invocation invocation,
{Object? returnValue, Object? returnValueForMissingStub}) {
if (invocation.memberName == #sendMessage) {
final namedArgs = invocation.namedArguments;
capturedMessage = namedArgs[#message] as Message?;
capturedExchangeName = namedArgs[#exchangeName] as String?;
capturedRoutingKey = namedArgs[#routingKey] as String?;
return null;
}
return super.noSuchMethod(invocation,
returnValue: returnValue,
returnValueForMissingStub: returnValueForMissingStub);
}
}
class TestCommand implements Command {
final String data;
TestCommand(this.data);
}
class TestHandler implements Handler {
@override
Future<dynamic> handle(Command command) async {
if (command is TestCommand) {
return 'Handled: ${command.data}';
}
throw UnimplementedError();
}
}
class TestQueuedCommand implements Command, ShouldQueue {
final String data;
TestQueuedCommand(this.data);
}
void main() {
late MockContainer container;
late MockEventBus eventBus;
late MockMQClient mqClient;
late Dispatcher dispatcher;
setUp(() {
container = MockContainer();
eventBus = MockEventBus();
mqClient = MockMQClient();
container.registerInstance<EventBus>(eventBus);
container.registerInstance<MQClient>(mqClient);
dispatcher = Dispatcher(container);
});
group('Dispatcher', () {
test('dispatchNow should handle command and return result', () async {
final command = TestCommand('test data');
final handler = TestHandler();
container.registerInstance<TestHandler>(handler);
dispatcher.map({TestCommand: TestHandler});
final commandEventController = StreamController<CommandEvent>();
when(eventBus.on<CommandEvent>())
.thenAnswer((_) => commandEventController.stream);
final future = dispatcher.dispatchNow(command);
// Simulate the event firing
commandEventController
.add(CommandEvent(command, result: 'Handled: test data'));
final result = await future;
expect(result, equals('Handled: test data'));
await commandEventController.close();
});
test('dispatch should handle regular commands immediately', () async {
final command = TestCommand('regular');
final handler = TestHandler();
container.registerInstance<TestHandler>(handler);
dispatcher.map({TestCommand: TestHandler});
final commandEventController = StreamController<CommandEvent>();
when(eventBus.on<CommandEvent>())
.thenAnswer((_) => commandEventController.stream);
final future = dispatcher.dispatch(command);
// Simulate the event firing
commandEventController
.add(CommandEvent(command, result: 'Handled: regular'));
final result = await future;
expect(result, equals('Handled: regular'));
await commandEventController.close();
});
test('dispatch should queue ShouldQueue commands', () async {
final command = TestQueuedCommand('queued data');
// Dispatch the command
await dispatcher.dispatch(command);
// Verify that sendMessage was called and check the message properties
expect(mqClient.capturedMessage, isNotNull);
expect(mqClient.capturedMessage!.payload, equals(command));
expect(mqClient.capturedMessage!.headers?['commandType'],
equals('TestQueuedCommand'));
// Optionally, verify exchange name and routing key if needed
expect(mqClient.capturedExchangeName, isNull);
expect(mqClient.capturedRoutingKey, isNull);
});
test(
'dispatchAfterResponse should send message to queue with specific header',
() {
final command = TestCommand('after response data');
// Call dispatchAfterResponse
dispatcher.dispatchAfterResponse(command);
// Verify that sendMessage was called and check the message properties
expect(mqClient.capturedMessage, isNotNull);
expect(mqClient.capturedMessage!.payload, equals(command));
expect(mqClient.capturedMessage!.headers?['commandType'],
equals('TestCommand'));
expect(mqClient.capturedMessage!.headers?['dispatchAfterResponse'],
equals('true'));
// Verify routing key
expect(mqClient.capturedRoutingKey, equals('after_response_queue'));
// Optionally, verify exchange name if needed
expect(mqClient.capturedExchangeName, isNull);
});
test('map should register command handlers', () {
dispatcher.map({TestCommand: TestHandler});
// Mock the event bus behavior for this test
when(eventBus.on<CommandEvent>()).thenAnswer((_) => Stream.empty());
// This test is a bit tricky to verify directly, but we can check if dispatch doesn't throw
expect(() => dispatcher.dispatch(TestCommand('test')), returnsNormally);
});
});
}

7
core/console/.gitignore vendored
View file

@ -1,7 +0,0 @@
# 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

3
core/console/CHANGELOG.md
View file

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

10
core/console/LICENSE.md
View file

@ -1,10 +0,0 @@
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.

1
core/console/README.md
View file

@ -1 +0,0 @@
<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>

30
core/console/analysis_options.yaml
View file

@ -1,30 +0,0 @@
# 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

0
core/console/lib/src/.gitkeep
View file

17
core/console/pubspec.yaml
View file

@ -1,17 +0,0 @@
name: Angel3_console
description: The Console 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:
# path: ^1.8.0
dev_dependencies:
lints: ^3.0.0
test: ^1.24.0

0
core/console/test/.gitkeep
View file

7
core/events/.gitignore vendored
View file

@ -1,7 +0,0 @@
# 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

3
core/events/CHANGELOG.md
View file

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

10
core/events/LICENSE.md
View file

@ -1,10 +0,0 @@
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.

1
core/events/README.md
View file

@ -1 +0,0 @@
<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>

30
core/events/analysis_options.yaml
View file

@ -1,30 +0,0 @@
# 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

3
core/events/lib/dispatcher.dart
View file

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

499
core/events/lib/src/dispatcher.dart
View file

@ -1,499 +0,0 @@
import 'dart:async';
import 'package:angel3_container/angel3_container.dart';
import 'package:angel3_reactivex/angel3_reactivex.dart';
import 'package:angel3_event_bus/event_bus.dart';
import 'package:angel3_mq/mq.dart';
// Simulating some of the Laravel interfaces/classes
abstract class ShouldBroadcast {}
abstract class ShouldQueue {}
abstract class ShouldBeEncrypted {}
abstract class ShouldDispatchAfterCommit {}
class Dispatcher implements DispatcherContract {
// Properties as specified in YAML
final Container container;
final Map<String, List<Function>> _listeners = {};
final Map<String, List<Function>> _wildcards = {};
final Map<String, List<Function>> _wildcardsCache = {};
late final Function _queueResolver;
late final Function _transactionManagerResolver;
final Map<String, List<Function>> _eventBusListeners = {};
final Map<String, Completer<dynamic>> _untilCompleters = {};
final Map<String, StreamSubscription> _eventBusSubscriptions = {};
final Set<String> _processedMessageIds = {};
// Properties for Angel3 packages
final EventBus _eventBus;
late final MQClient? _mqClient;
final Map<String, BehaviorSubject<dynamic>> _subjects = {};
// Queue and exchange names
static const String _eventsQueue = 'events_queue';
static const String _delayedEventsQueue = 'delayed_events_queue';
static const String _eventsExchange = 'events_exchange';
Dispatcher(this.container) : _eventBus = EventBus();
// Setter for _mqClient
set mqClient(MQClient client) {
_mqClient = client;
_setupQueuesAndExchanges();
_startProcessingQueuedEvents();
}
void _setupQueuesAndExchanges() {
_mqClient?.declareQueue(_eventsQueue);
_mqClient?.declareQueue(_delayedEventsQueue);
_mqClient?.declareExchange(
exchangeName: _eventsExchange,
exchangeType: ExchangeType.direct,
);
_mqClient?.bindQueue(
queueId: _eventsQueue,
exchangeName: _eventsExchange,
bindingKey: _eventsQueue,
);
_mqClient?.bindQueue(
queueId: _delayedEventsQueue,
exchangeName: _eventsExchange,
bindingKey: _delayedEventsQueue,
);
}
void _startProcessingQueuedEvents() {
_mqClient?.fetchQueue(_eventsQueue).listen((Message message) async {
if (message.payload is Map) {
final eventData = message.payload as Map<String, dynamic>;
if (eventData.containsKey('event') &&
eventData.containsKey('payload')) {
await dispatch(eventData['event'], eventData['payload']);
} else {
print('Invalid message format: ${message.payload}');
}
} else {
print('Unexpected payload type: ${message.payload.runtimeType}');
}
});
}
@override
void listen(dynamic events, dynamic listener) {
if (events is String) {
_addListener(events, listener);
} else if (events is List) {
for (var event in events) {
_addListener(event, listener);
}
}
if (events is String && events.contains('*')) {
_setupWildcardListen(events, listener);
}
}
void _addListener(String event, dynamic listener) {
_listeners.putIfAbsent(event, () => []).add(listener);
// Create a subject for this event if it doesn't exist
_subjects.putIfAbsent(event, () => BehaviorSubject<dynamic>());
// Add EventBus listener and store the subscription
final subscription = _eventBus.on().listen((AppEvent busEvent) {
if (busEvent is CustomAppEvent && busEvent.eventName == event) {
listener(event, busEvent.payload);
}
});
_eventBusSubscriptions[event] = subscription;
}
void _setupWildcardListen(String event, Function listener) {
_wildcards.putIfAbsent(event, () => []).add(listener);
_wildcardsCache.clear();
}
@override
bool hasListeners(String eventName) {
return _listeners.containsKey(eventName) ||
_wildcards.containsKey(eventName) ||
hasWildcardListeners(eventName);
}
bool hasWildcardListeners(String eventName) {
return _wildcards.keys
.any((pattern) => _isWildcardMatch(pattern, eventName));
}
@override
void push(String event, [dynamic payload]) {
final effectivePayload = payload ?? [];
_mqClient?.sendMessage(
exchangeName: _eventsExchange,
routingKey: _delayedEventsQueue,
message: Message(
headers: {'expiration': '5000'}, // 5 seconds delay
payload: {
'event': event,
'payload':
effectivePayload is List ? effectivePayload : [effectivePayload],
},
timestamp: DateTime.now().toIso8601String(),
id: 'msg_${DateTime.now().millisecondsSinceEpoch}', // Ensure unique ID
),
);
}
@override
Future<void> flush(String event) async {
final messageStream = _mqClient?.fetchQueue(_delayedEventsQueue);
if (messageStream == null) {
print('Warning: MQClient is not initialized');
return;
}
final messagesToProcess = <Message>[];
// Collect messages to process
await for (final message in messageStream) {
print('Examining message: ${message.id}');
if (message.payload is Map<String, dynamic> &&
!_processedMessageIds.contains(message.id)) {
final eventData = message.payload as Map<String, dynamic>;
if (eventData['event'] == event) {
print('Adding message to process: ${message.id}');
messagesToProcess.add(message);
}
}
}
print('Total messages to process: ${messagesToProcess.length}');
// Process collected messages
for (final message in messagesToProcess) {
final eventData = message.payload as Map<String, dynamic>;
print('Processing message: ${message.id}');
await dispatch(eventData['event'], eventData['payload']);
_mqClient?.deleteMessage(_delayedEventsQueue, message);
_processedMessageIds.add(message.id);
}
}
@override
void subscribe(dynamic subscriber) {
if (subscriber is EventBusSubscriber) {
subscriber.subscribe(_eventBus);
} else {
// Handle other types of subscribers
}
}
@override
Future<dynamic> until(dynamic event, [dynamic payload]) {
if (event is String) {
final completer = Completer<dynamic>();
_untilCompleters[event] = completer;
// Set up a one-time listener for this event
listen(event, (dynamic e, dynamic p) {
if (!completer.isCompleted) {
completer.complete(p);
_untilCompleters.remove(event);
}
});
// If payload is provided, dispatch the event immediately
if (payload != null) {
// Use dispatch instead of push to ensure immediate processing
dispatch(event, payload);
}
return completer.future;
}
throw ArgumentError('Event must be a String');
}
@override
Future<dynamic> dispatch(dynamic event, [dynamic payload, bool? halt]) async {
final eventName = event is String ? event : event.runtimeType.toString();
final eventPayload = payload ?? (event is AppEvent ? event : []);
if (event is ShouldBroadcast ||
(eventPayload is List &&
eventPayload.isNotEmpty &&
eventPayload[0] is ShouldBroadcast)) {
await _broadcastEvent(event);
}
if (event is ShouldQueue ||
(eventPayload is List &&
eventPayload.isNotEmpty &&
eventPayload[0] is ShouldQueue)) {
return _queueEvent(eventName, eventPayload);
}
final listeners = getListeners(eventName);
for (var listener in listeners) {
final response =
await Function.apply(listener, [eventName, eventPayload]);
if (halt == true && response != null) {
return response;
}
if (response == false) {
break;
}
}
return halt == true ? null : listeners;
}
// void _addToSubject(String eventName, dynamic payload) {
// if (_subjects.containsKey(eventName)) {
// _subjects[eventName]!.add(payload);
// }
// }
@override
List<Function> getListeners(String eventName) {
var listeners = <Function>[
...(_listeners[eventName] ?? []),
...(_wildcardsCache[eventName] ?? _getWildcardListeners(eventName)),
...(_eventBusListeners[eventName] ?? []),
];
return listeners;
}
List<Function> _getWildcardListeners(String eventName) {
final wildcardListeners = <Function>[];
for (var entry in _wildcards.entries) {
if (_isWildcardMatch(entry.key, eventName)) {
wildcardListeners.addAll(entry.value);
}
}
_wildcardsCache[eventName] = wildcardListeners;
return wildcardListeners;
}
@override
void forget(String event) {
// Remove from _listeners
_listeners.remove(event);
// Remove from _subjects
if (_subjects.containsKey(event)) {
_subjects[event]?.close();
_subjects.remove(event);
}
// Cancel and remove EventBus subscription
_eventBusSubscriptions[event]?.cancel();
_eventBusSubscriptions.remove(event);
// Remove from wildcards if applicable
if (event.contains('*')) {
_wildcards.remove(event);
_wildcardsCache.clear();
} else {
// If it's not a wildcard, we need to remove it from any matching wildcard listeners
_wildcards.forEach((pattern, listeners) {
if (_isWildcardMatch(pattern, event)) {
_wildcards[pattern] = listeners
.where((listener) => listener != _listeners[event])
.toList();
}
});
_wildcardsCache.clear();
}
// Remove any 'until' completers for this event
_untilCompleters.remove(event);
}
@override
void forgetPushed() {
_listeners.removeWhere((key, _) => key.endsWith('_pushed'));
_eventBusListeners.removeWhere((key, _) => key.endsWith('_pushed'));
// Note: We're not clearing all EventBus listeners here, as that might affect other parts of your application
}
@override
void setQueueResolver(Function resolver) {
_queueResolver = resolver;
}
@override
void setTransactionManagerResolver(Function resolver) {
_transactionManagerResolver = resolver;
}
// Add these methods for testing purposes
void triggerQueueResolver() {
_queueResolver();
}
void triggerTransactionManagerResolver() {
_transactionManagerResolver();
}
@override
Map<String, List<Function>> getRawListeners() {
return Map.unmodifiable(_listeners);
}
bool _shouldBroadcast(List payload) {
return payload.isNotEmpty && payload[0] is ShouldBroadcast;
}
Future<void> _broadcastEvent(dynamic event) async {
// Implement broadcasting logic here
// For now, we'll just print a message
print('Broadcasting event: ${event.runtimeType}');
}
bool _isWildcardMatch(String pattern, String eventName) {
final regExp = RegExp('^${pattern.replaceAll('*', '.*')}\$');
return regExp.hasMatch(eventName);
}
bool _shouldQueue(List payload) {
return payload.isNotEmpty && payload[0] is ShouldQueue;
}
Future<void> _queueEvent(String eventName, dynamic payload) async {
_mqClient?.sendMessage(
exchangeName: _eventsExchange,
routingKey: _eventsQueue,
message: Message(
payload: {'event': eventName, 'payload': payload},
timestamp: DateTime.now().toIso8601String(),
),
);
}
// Updated on<T> method
Stream<T> on<T>(String event) {
return (_subjects
.putIfAbsent(event, () => BehaviorSubject<dynamic>())
.stream as Stream<T>)
.where((event) => event is T)
.cast<T>();
}
// In your Dispatcher class
void setMQClient(MQClient client) {
_mqClient = client;
}
// Method to close the MQClient connection
Future<void> close() async {
_mqClient?.close();
}
// Don't forget to close the subjects when they're no longer needed
void dispose() {
for (var subject in _subjects.values) {
subject.close();
}
}
}
// ... rest of the code (DispatcherContract, EventBusSubscriber, etc.) remains the same
abstract class DispatcherContract {
void listen(dynamic events, dynamic listener);
bool hasListeners(String eventName);
void push(String event, [dynamic payload]);
Future<void> flush(String event);
void subscribe(dynamic subscriber);
Future<dynamic> until(dynamic event, [dynamic payload]);
Future<dynamic> dispatch(dynamic event, [dynamic payload, bool halt]);
List<Function> getListeners(String eventName);
void forget(String event);
void forgetPushed();
void setQueueResolver(Function resolver);
void setTransactionManagerResolver(Function resolver);
Map<String, List<Function>> getRawListeners();
}
// Helper class for EventBus subscribers
abstract class EventBusSubscriber {
void subscribe(EventBus eventBus);
}
// Mixin to simulate Macroable trait
mixin Macroable {
// Implementation of Macroable functionality
}
// Mixin to simulate ReflectsClosures trait
mixin ReflectsClosures {
// Implementation of ReflectsClosures functionality
}
// If not already defined, you might need to create an Event class
class Event {
final String name;
final dynamic data;
Event(this.name, this.data);
}
// Custom AppEvent subclasses for handling different event types
class StringBasedEvent extends AppEvent {
final String eventName;
final dynamic payload;
StringBasedEvent(this.eventName, this.payload);
@override
List<Object?> get props => [eventName, payload];
}
class CustomAppEvent extends AppEvent {
final String eventName;
final dynamic payload;
CustomAppEvent(this.eventName, this.payload);
@override
List<Object?> get props => [eventName, payload];
}
// This is a simple implementation of Reflector that does nothing
class EmptyReflector implements Reflector {
const EmptyReflector();
@override
ReflectedType reflectType(Type type) {
throw UnimplementedError();
}
@override
ReflectedInstance reflectInstance(Object object) {
throw UnimplementedError();
}
@override
ReflectedType reflectFutureOf(Type type) {
throw UnimplementedError();
}
@override
String? getName(Symbol symbol) {
// TODO: implement getName
throw UnimplementedError();
}
@override
ReflectedClass? reflectClass(Type clazz) {
// TODO: implement reflectClass
throw UnimplementedError();
}
@override
ReflectedFunction? reflectFunction(Function function) {
// TODO: implement reflectFunction
throw UnimplementedError();
}
}

21
core/events/pubspec.yaml
View file

@ -1,21 +0,0 @@
name: angel3_events
description: The Events 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: ^9.0.0
angel3_mq: ^9.0.0
angel3_event_bus: ^9.0.0
angel3_framework: ^9.0.0
angel3_reactivex: ^0.27.5
# path: ^1.8.0
dev_dependencies:
lints: ^3.0.0
test: ^1.24.0

430
core/events/test/event_test.dart
View file

@ -1,430 +0,0 @@
import 'package:angel3_event_bus/res/app_event.dart';
import 'package:test/test.dart';
import 'package:angel3_container/angel3_container.dart';
import 'package:angel3_mq/mq.dart';
import 'package:angel3_events/dispatcher.dart'; // Replace with the actual import path
void main() {
late Dispatcher dispatcher;
late MockMQClient mockMQClient;
setUp(() {
var container = Container(EmptyReflector());
dispatcher = Dispatcher(container);
mockMQClient = MockMQClient();
dispatcher.mqClient = mockMQClient; // Use the setter
// Clear the queue before each test
mockMQClient.queuedMessages.clear();
});
group('Dispatcher', () {
test('listen and dispatch', () async {
var callCount = 0;
dispatcher.listen('test_event', (dynamic event, dynamic payload) {
expect(event, equals('test_event'));
expect(payload, equals(['test_payload']));
callCount++;
});
await dispatcher.dispatch('test_event', ['test_payload']);
expect(callCount, equals(1));
});
test('wildcard listener', () async {
var callCount = 0;
dispatcher.listen('test.*', (dynamic event, dynamic payload) {
expect(event, matches(RegExp(r'^test\.')));
callCount++;
});
await dispatcher.dispatch('test.one', ['payload1']);
await dispatcher.dispatch('test.two', ['payload2']);
expect(callCount, equals(2));
});
test('hasListeners', () {
dispatcher.listen('test_event', (dynamic event, dynamic payload) {});
expect(dispatcher.hasListeners('test_event'), isTrue);
expect(dispatcher.hasListeners('non_existent_event'), isFalse);
});
test('until', () async {
// Test without pushing the event immediately
var futureResult = dispatcher.until('test_event');
// Use a small delay to ensure the until listener is set up
await Future.delayed(Duration(milliseconds: 10));
await dispatcher.dispatch('test_event', ['test_payload']);
var result = await futureResult;
expect(result, equals(['test_payload']));
// Test with pushing the event immediately
result =
await dispatcher.until('another_test_event', ['another_payload']);
expect(result, equals(['another_payload']));
}, timeout: Timeout(Duration(seconds: 5))); // Add a reasonable timeout
test('forget', () async {
var callCount = 0;
dispatcher.listen('test_event', (dynamic event, dynamic payload) {
callCount++;
});
await dispatcher.dispatch('test_event');
expect(callCount, equals(1));
dispatcher.forget('test_event');
await dispatcher.dispatch('test_event');
expect(callCount, equals(1)); // Should not increase
});
test('push and flush', () async {
print('Starting push and flush test');
// Push 4 messages
for (var i = 0; i < 4; i++) {
dispatcher.push('delayed_event', ['delayed_payload_$i']);
}
// Verify that 4 messages were queued
expect(mockMQClient.queuedMessages['delayed_events_queue']?.length,
equals(4),
reason: 'Should have queued exactly 4 messages');
print(
'Queued messages: ${mockMQClient.queuedMessages['delayed_events_queue']?.length}');
var callCount = 0;
var processedPayloads = <String>[];
// Remove any existing listeners
dispatcher.forget('delayed_event');
dispatcher.listen('delayed_event', (dynamic event, dynamic payload) {
print('Listener called with payload: $payload');
expect(event, equals('delayed_event'));
expect(payload[0], startsWith('delayed_payload_'));
processedPayloads.add(payload[0]);
callCount++;
});
await dispatcher.flush('delayed_event');
print('After flush - Call count: $callCount');
print('Processed payloads: $processedPayloads');
expect(callCount, equals(4), reason: 'Should process exactly 4 messages');
expect(processedPayloads.toSet().length, equals(4),
reason: 'All payloads should be unique');
// Verify that all messages were removed from the queue
expect(mockMQClient.queuedMessages['delayed_events_queue']?.length,
equals(0),
reason: 'Queue should be empty after flush');
// Flush again to ensure no more messages are processed
await dispatcher.flush('delayed_event');
expect(callCount, equals(4),
reason: 'Should still be 4 after second flush');
});
test('shouldBroadcast', () async {
var broadcastEvent = BroadcastTestEvent();
var callCount = 0;
dispatcher.listen('BroadcastTestEvent', (dynamic event, dynamic payload) {
callCount++;
});
await dispatcher.dispatch(broadcastEvent);
expect(callCount, equals(1));
});
test('shouldQueue', () async {
var queueEvent = QueueTestEvent();
await dispatcher.dispatch(queueEvent);
expect(mockMQClient.queuedMessages['events_queue'], isNotEmpty);
expect(mockMQClient.queuedMessages['events_queue']!.first.payload,
containsPair('event', 'QueueTestEvent'));
});
test('forgetPushed removes only pushed events', () {
dispatcher.listen('event_pushed', (_, __) {});
dispatcher.listen('normal_event', (_, __) {});
dispatcher.forgetPushed();
expect(dispatcher.hasListeners('event_pushed'), isFalse);
expect(dispatcher.hasListeners('normal_event'), isTrue);
});
test('setQueueResolver and setTransactionManagerResolver', () {
var queueResolverCalled = false;
var transactionManagerResolverCalled = false;
dispatcher.setQueueResolver(() {
queueResolverCalled = true;
});
dispatcher.setTransactionManagerResolver(() {
transactionManagerResolverCalled = true;
});
// Trigger the resolvers
dispatcher.triggerQueueResolver();
dispatcher.triggerTransactionManagerResolver();
expect(queueResolverCalled, isTrue);
expect(transactionManagerResolverCalled, isTrue);
});
test('getRawListeners returns unmodifiable map', () {
dispatcher.listen('test_event', (_, __) {});
var rawListeners = dispatcher.getRawListeners();
expect(rawListeners, isA<Map<String, List<Function>>>());
expect(() => rawListeners['new_event'] = [], throwsUnsupportedError);
});
test('multiple listeners for same event', () async {
var callCount1 = 0;
var callCount2 = 0;
dispatcher.listen('multi_event', (_, __) => callCount1++);
dispatcher.listen('multi_event', (_, __) => callCount2++);
await dispatcher.dispatch('multi_event');
expect(callCount1, equals(1));
expect(callCount2, equals(1));
});
});
}
abstract class MQClientWrapper {
Stream<Message> fetchQueue(String queueId);
void sendMessage({
required Message message,
String? exchangeName,
String? routingKey,
});
String declareQueue(String queueId);
void declareExchange({
required String exchangeName,
required ExchangeType exchangeType,
});
void bindQueue({
required String queueId,
required String exchangeName,
String? bindingKey,
});
void close();
}
class RealMQClientWrapper implements MQClientWrapper {
final MQClient _client;
RealMQClientWrapper(this._client);
@override
Stream<Message> fetchQueue(String queueId) => _client.fetchQueue(queueId);
@override
void sendMessage({
required Message message,
String? exchangeName,
String? routingKey,
}) =>
_client.sendMessage(
message: message,
exchangeName: exchangeName,
routingKey: routingKey,
);
@override
String declareQueue(String queueId) => _client.declareQueue(queueId);
@override
void declareExchange({
required String exchangeName,
required ExchangeType exchangeType,
}) =>
_client.declareExchange(
exchangeName: exchangeName,
exchangeType: exchangeType,
);
@override
void bindQueue({
required String queueId,
required String exchangeName,
String? bindingKey,
}) =>
_client.bindQueue(
queueId: queueId,
exchangeName: exchangeName,
bindingKey: bindingKey,
);
@override
void close() => _client.close();
}
class MockMQClient implements MQClient {
Map<String, List<Message>> queuedMessages = {};
int _messageIdCounter = 0;
void queueMessage(String queueName, Message message) {
queuedMessages.putIfAbsent(queueName, () => []).add(message);
print(
'Queued message. Queue $queueName now has ${queuedMessages[queueName]?.length} messages');
}
@override
String declareQueue(String queueId) {
queuedMessages[queueId] = [];
return queueId;
}
@override
void deleteQueue(String queueId) {
queuedMessages.remove(queueId);
}
@override
Stream<Message> fetchQueue(String queueId) {
print('Fetching queue: $queueId');
return Stream.fromIterable(queuedMessages[queueId] ?? []);
}
@override
void sendMessage({
required Message message,
String? exchangeName,
String? routingKey,
}) {
print('Sending message to queue: $routingKey');
final newMessage = Message(
payload: message.payload,
headers: message.headers,
timestamp: message.timestamp,
id: 'msg_${_messageIdCounter++}',
);
queueMessage(routingKey ?? '', newMessage);
}
@override
Message? getLatestMessage(String queueId) {
final messages = queuedMessages[queueId];
return messages?.isNotEmpty == true ? messages!.last : null;
}
@override
void bindQueue({
required String queueId,
required String exchangeName,
String? bindingKey,
}) {
// Implement if needed for your tests
}
@override
void unbindQueue({
required String queueId,
required String exchangeName,
String? bindingKey,
}) {
// Implement if needed for your tests
}
@override
void declareExchange({
required String exchangeName,
required ExchangeType exchangeType,
}) {
// Implement if needed for your tests
}
@override
void deleteExchange(String exchangeName) {
// Implement if needed for your tests
}
@override
List<String> listQueues() {
return queuedMessages.keys.toList();
}
@override
void close() {
queuedMessages.clear();
}
@override
void deleteMessage(String queueId, Message message) {
print('Deleting message from queue: $queueId');
queuedMessages[queueId]?.removeWhere((m) => m.id == message.id);
print(
'After deletion, queue $queueId has ${queuedMessages[queueId]?.length} messages');
}
}
class BroadcastTestEvent implements AppEvent, ShouldBroadcast {
@override
List<Object?> get props => [];
@override
bool? get stringify => true;
@override
DateTime get timestamp => DateTime.now();
}
class QueueTestEvent implements AppEvent, ShouldQueue {
@override
List<Object?> get props => [];
@override
bool? get stringify => true;
@override
DateTime get timestamp => DateTime.now();
}
// This is a simple implementation of Reflector that does nothing
class EmptyReflector implements Reflector {
const EmptyReflector();
@override
ReflectedType reflectType(Type type) {
throw UnimplementedError();
}
@override
ReflectedInstance reflectInstance(Object object) {
throw UnimplementedError();
}
@override
ReflectedType reflectFutureOf(Type type) {
throw UnimplementedError();
}
@override
String? getName(Symbol symbol) {
// TODO: implement getName
throw UnimplementedError();
}
@override
ReflectedClass? reflectClass(Type clazz) {
// TODO: implement reflectClass
throw UnimplementedError();
}
@override
ReflectedFunction? reflectFunction(Function function) {
// TODO: implement reflectFunction
throw UnimplementedError();
}
}

7
core/pipeline/.gitignore vendored
View file

@ -1,7 +0,0 @@
# 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

3
core/pipeline/CHANGELOG.md
View file

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

10
core/pipeline/LICENSE.md
View file

@ -1,10 +0,0 @@
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.

1
core/pipeline/README.md
View file

@ -1 +0,0 @@
<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>

30
core/pipeline/analysis_options.yaml
View file

@ -1,30 +0,0 @@
# 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

38
core/pipeline/examples/async_pipeline.dart
View file

@ -1,38 +0,0 @@
import 'package:angel3_framework/angel3_framework.dart';
import 'package:angel3_framework/http.dart';
import 'package:angel3_container/mirrors.dart';
import 'package:angel3_pipeline/pipeline.dart';
class AsyncGreetingPipe {
Future<dynamic> handle(String input, Function next) async {
await Future.delayed(Duration(seconds: 1));
return next('Hello, $input');
}
}
class AsyncExclamationPipe {
Future<dynamic> handle(String input, Function next) async {
await Future.delayed(Duration(seconds: 1));
return next('$input!');
}
}
void main() async {
var app = Angel(reflector: MirrorsReflector());
var http = AngelHttp(app);
app.container.registerSingleton((c) => Pipeline(c));
app.get('/', (req, res) async {
var pipeline = app.container.make<Pipeline>();
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');
}

36
core/pipeline/examples/basic_usage.dart
View file

@ -1,36 +0,0 @@
import 'package:angel3_framework/angel3_framework.dart';
import 'package:angel3_framework/http.dart';
import 'package:angel3_container/mirrors.dart';
import 'package:angel3_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 = Angel(reflector: MirrorsReflector());
var http = AngelHttp(app);
app.container.registerSingleton((c) => Pipeline(c));
app.get('/', (req, res) async {
var pipeline = app.container.make<Pipeline>();
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');
}

34
core/pipeline/examples/error_handling.dart
View file

@ -1,34 +0,0 @@
import 'package:angel3_framework/angel3_framework.dart';
import 'package:angel3_framework/http.dart';
import 'package:angel3_container/mirrors.dart';
import 'package:angel3_pipeline/pipeline.dart';
class ErrorPipe {
dynamic handle(String input, Function next) {
throw Exception('Simulated error');
}
}
void main() async {
var app = Angel(reflector: MirrorsReflector());
var http = AngelHttp(app);
app.container.registerSingleton((c) => Pipeline(c));
app.get('/', (req, res) async {
var pipeline = app.container.make<Pipeline>();
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');
}

35
core/pipeline/examples/mixed_pipes.dart
View file

@ -1,35 +0,0 @@
import 'package:angel3_framework/angel3_framework.dart';
import 'package:angel3_framework/http.dart';
import 'package:angel3_container/mirrors.dart';
import 'package:angel3_pipeline/pipeline.dart';
class GreetingPipe {
dynamic handle(String input, Function next) {
return next('Hello, $input');
}
}
void main() async {
var app = Angel(reflector: MirrorsReflector());
var http = AngelHttp(app);
app.container.registerSingleton((c) => Pipeline(c));
app.get('/', (req, res) async {
var pipeline = app.container.make<Pipeline>();
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');
}

5
core/pipeline/lib/pipeline.dart
View file

@ -1,5 +0,0 @@
library;
export 'src/pipeline.dart';
export 'src/conditionable.dart';
export 'src/pipeline_contract.dart';

16
core/pipeline/lib/src/conditionable.dart
View file

@ -1,16 +0,0 @@
/// 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;
}
}

214
core/pipeline/lib/src/pipeline.dart
View file

@ -1,214 +0,0 @@
import 'dart:async';
import 'dart:mirrors';
import 'package:angel3_container/angel3_container.dart';
import 'package:logging/logging.dart';
import 'pipeline_contract.dart';
import 'conditionable.dart';
/// Defines the signature for a pipe function.
typedef PipeFunction = FutureOr<dynamic> Function(
dynamic passable, FutureOr<dynamic> Function(dynamic) next);
/// The primary class for building and executing pipelines.
class Pipeline with Conditionable<Pipeline> implements PipelineContract {
/// The container implementation.
Container? _container;
final Map<String, Type> _typeMap = {};
/// 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);
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) {
_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(FutureOr<dynamic> Function(dynamic) callback) async {
PipeFunction pipeline = _pipes.fold<PipeFunction>(
(dynamic passable, FutureOr<dynamic> Function(dynamic) next) async =>
await callback(passable),
(PipeFunction next, dynamic pipe) => (dynamic passable,
FutureOr<dynamic> Function(dynamic) nextPipe) async {
return await carry(pipe, passable,
(dynamic result) async => await next(result, nextPipe));
});
return await pipeline(_passable, (dynamic result) async => result);
}
/// 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) {
return await pipe(passable, next);
}
if (pipe is String) {
if (_container == null) {
throw Exception('Container is null, cannot resolve pipe: $pipe');
}
Type? pipeType = _typeMap[pipe];
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]);
}
throw Exception('Unsupported pipe type: ${pipe.runtimeType}');
} 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;
}
Future<dynamic> invokeMethod(
dynamic instance, String methodName, List<dynamic> arguments) async {
var instanceMirror = reflect(instance);
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) {
_logger.severe('Exception occurred in pipeline', e);
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);
});
}
}

9
core/pipeline/lib/src/pipeline_contract.dart
View file

@ -1,9 +0,0 @@
/// 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();
}

19
core/pipeline/pubspec.yaml
View file

@ -1,19 +0,0 @@
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
angel3_framework: ^8.0.0
logging: ^1.1.0
dev_dependencies:
lints: ^3.0.0
test: ^1.24.0

106
core/pipeline/test/pipeline_test.dart
View file

@ -1,106 +0,0 @@
import 'package:test/test.dart';
import 'package:angel3_framework/angel3_framework.dart';
import 'package:angel3_container/angel3_container.dart';
import 'package:angel3_container/mirrors.dart';
import 'package:angel3_pipeline/pipeline.dart';
class AddExclamationPipe {
Future<String> handle(String input, Function next) async {
return await next('$input!');
}
}
class UppercasePipe {
Future<String> handle(String input, Function next) async {
return await next(input.toUpperCase());
}
}
void main() {
late Angel app;
late Container container;
late Pipeline pipeline;
setUp(() {
app = Angel(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<Exception>()),
);
});
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)));
});
}

7
core/process/.gitignore vendored
View file

@ -1,7 +0,0 @@
# 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

3
core/process/CHANGELOG.md
View file

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

10
core/process/LICENSE.md
View file

@ -1,10 +0,0 @@
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.

1
core/process/README.md
View file

@ -1 +0,0 @@
<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>

30
core/process/analysis_options.yaml
View file

@ -1,30 +0,0 @@
# 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

36
core/process/examples/basic_process/main.dart
View file

@ -1,36 +0,0 @@
// 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');
}

37
core/process/examples/process_pipeline/main.dart
View file

@ -1,37 +0,0 @@
// 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');
}

37
core/process/examples/process_pool/main.dart
View file

@ -1,37 +0,0 @@
// 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');
}

67
core/process/examples/web_server_with_processes/main.dart
View file

@ -1,67 +0,0 @@
// 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');
}

39
core/process/examples/web_server_with_processes/views/index.mustache
View file

@ -1,39 +0,0 @@
<!-- 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>

7
core/process/lib/angel3_process.dart
View file

@ -1,7 +0,0 @@
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';

250
core/process/lib/src/process.dart
View file

@ -1,250 +0,0 @@
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)';
}
}

21
core/process/lib/src/process_helper.dart
View file

@ -1,21 +0,0 @@
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,
);
}

148
core/process/lib/src/process_manager.dart
View file

@ -1,148 +0,0 @@
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();
}

50
core/process/lib/src/process_pipeline.dart
View file

@ -1,50 +0,0 @@
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,
'',
);
}
}

62
core/process/lib/src/process_pool.dart
View file

@ -1,62 +0,0 @@
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,
);
}
}

24
core/process/lib/src/process_service_provider.dart
View file

@ -1,24 +0,0 @@
/* 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');
}
} */

25
core/process/pubspec.yaml
View file

@ -1,25 +0,0 @@
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

80
core/process/test/process_test.dart
View file

@ -1,80 +0,0 @@
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)));
}

102
core/process/test/process_test_extended.dart
View file

@ -1,102 +0,0 @@
import 'dart:async';
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>()));
});
}

10
core/queue/.gitignore vendored
View file

@ -1,10 +0,0 @@
# 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
*.mocks.dart
*.reflectable.dart

3
core/queue/CHANGELOG.md
View file

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

10
core/queue/LICENSE.md
View file

@ -1,10 +0,0 @@
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.

1
core/queue/README.md
View file

@ -1 +0,0 @@
<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>

30
core/queue/analysis_options.yaml
View file

@ -1,30 +0,0 @@
# 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

0
core/queue/lib/queue.dart
View file

32
core/queue/lib/src/job_queued_event.dart
View file

@ -1,32 +0,0 @@
import 'package:angel3_event_bus/event_bus.dart';
import 'package:equatable/equatable.dart';
class JobQueuedEvent extends AppEvent {
final String connectionName;
final String? queue;
final dynamic jobId;
final dynamic job;
final String payload;
final Duration? delay;
JobQueuedEvent(this.connectionName, this.queue, this.jobId, this.job,
this.payload, this.delay);
@override
List<Object?> get props =>
[connectionName, queue, jobId, job, payload, delay];
@override
Map<String, dynamic> toJson() {
return {
'connectionName': connectionName,
'queue': queue,
'jobId': jobId,
'job': job.toString(), // or a more appropriate serialization of the job
'payload': payload,
'delay': delay?.inMilliseconds,
};
}
String get name => 'job.queued';
}

29
core/queue/lib/src/job_queueing_event.dart
View file

@ -1,29 +0,0 @@
import 'package:angel3_event_bus/event_bus.dart';
import 'package:equatable/equatable.dart';
class JobQueueingEvent extends AppEvent {
final String connectionName;
final String? queue;
final dynamic job;
final String payload;
final Duration? delay;
JobQueueingEvent(
this.connectionName, this.queue, this.job, this.payload, this.delay);
@override
List<Object?> get props => [connectionName, queue, job, payload, delay];
@override
Map<String, dynamic> toJson() {
return {
'connectionName': connectionName,
'queue': queue,
'job': job.toString(), // or a more appropriate serialization of the job
'payload': payload,
'delay': delay?.inMilliseconds,
};
}
String get name => 'job.queueing';
}

396
core/queue/lib/src/queue.dart
View file

@ -1,396 +0,0 @@
// lib/src/queue.dart
import 'dart:async';
import 'dart:convert';
import 'package:angel3_container/angel3_container.dart';
import 'package:angel3_event_bus/event_bus.dart';
import 'package:angel3_mq/mq.dart';
import 'package:angel3_reactivex/angel3_reactivex.dart';
import 'package:crypto/crypto.dart';
import 'package:uuid/uuid.dart';
import 'job_queueing_event.dart';
import 'job_queued_event.dart';
import 'should_be_encrypted.dart';
import 'should_queue_after_commit.dart';
abstract class Queue with InteractsWithTime {
/// The IoC container instance.
final Container container;
final EventBus eventBus;
final MQClient mq;
final Subject<dynamic> jobSubject;
final Uuid uuid = Uuid();
/// The connection name for the queue.
String _connectionName;
/// Indicates that jobs should be dispatched after all database transactions have committed.
bool dispatchAfterCommit;
/// The create payload callbacks.
static final List<Function> _createPayloadCallbacks = [];
Queue(this.container, this.eventBus, this.mq,
{String connectionName = 'default', this.dispatchAfterCommit = false})
: _connectionName = connectionName,
jobSubject = PublishSubject<dynamic>() {
_setupJobObservable();
}
void _setupJobObservable() {
jobSubject.stream.listen((job) {
// Process the job
print('Processing job: $job');
// Implement your job processing logic here
});
}
Future<dynamic> pushOn(String queue, dynamic job, [dynamic data = '']) {
return push(job, data, queue);
}
Future<dynamic> laterOn(String queue, Duration delay, dynamic job,
[dynamic data = '']) {
return later(delay, job, data, queue);
}
Future<void> bulk(List<dynamic> jobs,
[dynamic data = '', String? queue]) async {
for (var job in jobs) {
await push(job, data, queue);
}
}
// Add this method
void setContainer(Container container) {
// This method might not be necessary in Dart, as we're using final for container
// But we can implement it for API compatibility
throw UnsupportedError(
'Container is final and cannot be changed after initialization');
}
// Update createPayload method to include exception handling
Future<String> createPayload(dynamic job, String queue,
[dynamic data = '']) async {
if (job is Function) {
// TODO: Implement CallQueuedClosure equivalent
throw UnimplementedError('Closure jobs are not yet supported');
}
try {
final payload = jsonEncode(await createPayloadMap(job, queue, data));
return payload;
} catch (e) {
throw InvalidPayloadException('Unable to JSON encode payload: $e');
}
}
Future<Map<String, dynamic>> createPayloadMap(dynamic job, String queue,
[dynamic data = '']) async {
if (job is Object) {
return createObjectPayload(job, queue);
} else {
return createStringPayload(job.toString(), queue, data);
}
}
Future<Map<String, dynamic>> createObjectPayload(
Object job, String queue) async {
final payload = await withCreatePayloadHooks(queue, {
'uuid': const Uuid().v4(),
'displayName': getDisplayName(job),
'job': 'CallQueuedHandler@call', // TODO: Implement CallQueuedHandler
'maxTries': getJobTries(job),
'maxExceptions': job is HasMaxExceptions ? job.maxExceptions : null,
'failOnTimeout': job is HasFailOnTimeout ? job.failOnTimeout : false,
'backoff': getJobBackoff(job),
'timeout': job is HasTimeout ? job.timeout : null,
'retryUntil': getJobExpiration(job),
'data': {
'commandName': job.runtimeType.toString(),
'command': job,
},
});
final command = jobShouldBeEncrypted(job) && container.has<Encrypter>()
? container.make<Encrypter>().encrypt(jsonEncode(job))
: jsonEncode(job);
payload['data'] = {
...payload['data'] as Map<String, dynamic>,
'commandName': job.runtimeType.toString(),
'command': command,
};
return payload;
}
String getDisplayName(Object job) {
if (job is HasDisplayName) {
return job.displayName();
}
return job.runtimeType.toString();
}
int? getJobTries(dynamic job) {
if (job is HasTries) {
return job.tries;
}
return null;
}
String? getJobBackoff(dynamic job) {
if (job is HasBackoff) {
final backoff = job.backoff;
if (backoff == null) return null;
if (backoff is Duration) {
return backoff.inSeconds.toString();
}
if (backoff is List<Duration>) {
return backoff.map((d) => d.inSeconds).join(',');
}
}
return null;
}
int? getJobExpiration(dynamic job) {
if (job is HasRetryUntil) {
final retryUntil = job.retryUntil;
if (retryUntil == null) return null;
return retryUntil.millisecondsSinceEpoch ~/ 1000;
}
return null;
}
bool jobShouldBeEncrypted(Object job) {
return job is ShouldBeEncrypted ||
(job is HasShouldBeEncrypted && job.shouldBeEncrypted);
}
Future<Map<String, dynamic>> createStringPayload(
String job, String queue, dynamic data) async {
return withCreatePayloadHooks(queue, {
'uuid': const Uuid().v4(),
'displayName': job.split('@')[0],
'job': job,
'maxTries': null,
'maxExceptions': null,
'failOnTimeout': false,
'backoff': null,
'timeout': null,
'data': data,
});
}
static void createPayloadUsing(Function? callback) {
if (callback == null) {
_createPayloadCallbacks.clear();
} else {
_createPayloadCallbacks.add(callback);
}
}
Future<Map<String, dynamic>> withCreatePayloadHooks(
String queue, Map<String, dynamic> payload) async {
if (_createPayloadCallbacks.isNotEmpty) {
for (var callback in _createPayloadCallbacks) {
final result = await callback(_connectionName, queue, payload);
if (result is Map<String, dynamic>) {
payload = {...payload, ...result};
}
}
}
return payload;
}
Future<dynamic> enqueueUsing(
dynamic job,
String payload,
String? queue,
Duration? delay,
Future<dynamic> Function(String, String?, Duration?) callback,
) async {
final String jobId = uuid.v4(); // Generate a unique job ID
if (shouldDispatchAfterCommit(job) && container.has<TransactionManager>()) {
return container.make<TransactionManager>().addCallback(() async {
await raiseJobQueueingEvent(queue, job, payload, delay);
final result = await callback(payload, queue, delay);
await raiseJobQueuedEvent(queue, jobId, job, payload, delay);
return result;
});
}
await raiseJobQueueingEvent(queue, job, payload, delay);
final result = await callback(payload, queue, delay);
await raiseJobQueuedEvent(queue, jobId, job, payload, delay);
// Use angel3_mq to publish the job
mq.sendMessage(
message: Message(
headers: {'jobId': jobId}, // Include jobId in headers
payload: payload,
timestamp: DateTime.now().toIso8601String(),
),
exchangeName: '', // Use default exchange
routingKey: queue ?? 'default',
);
// Use angel3_reactivex to add the job to the subject
jobSubject.add(job);
return result;
}
bool shouldDispatchAfterCommit(dynamic job) {
if (job is ShouldQueueAfterCommit) {
return true;
}
if (job is HasAfterCommit) {
return job.afterCommit;
}
return dispatchAfterCommit;
}
Future<void> raiseJobQueueingEvent(
String? queue, dynamic job, String payload, Duration? delay) async {
if (container.has<EventBus>()) {
final eventBus = container.make<EventBus>();
eventBus
.fire(JobQueueingEvent(_connectionName, queue, job, payload, delay));
}
}
Future<void> raiseJobQueuedEvent(String? queue, dynamic jobId, dynamic job,
String payload, Duration? delay) async {
if (container.has<EventBus>()) {
final eventBus = container.make<EventBus>();
eventBus.fire(
JobQueuedEvent(_connectionName, queue, jobId, job, payload, delay));
}
}
String get connectionName => _connectionName;
set connectionName(String name) {
_connectionName = name;
}
Container getContainer() => container;
// Abstract methods to be implemented by subclasses
// Implement the push method
Future<dynamic> push(dynamic job, [dynamic data = '', String? queue]) async {
final payload = await createPayload(job, queue ?? 'default', data);
return enqueueUsing(job, payload, queue, null, (payload, queue, _) async {
final jobId = Uuid().v4();
mq.sendMessage(
message: Message(
id: jobId,
headers: {},
payload: payload,
timestamp: DateTime.now().toIso8601String(),
),
exchangeName: '',
routingKey: queue ?? 'default',
);
return jobId;
});
}
// Implement the later method
Future<dynamic> later(Duration delay, dynamic job,
[dynamic data = '', String? queue]) async {
final payload = await createPayload(job, queue ?? 'default', data);
return enqueueUsing(job, payload, queue, delay,
(payload, queue, delay) async {
final jobId = Uuid().v4();
await Future.delayed(delay!);
mq.sendMessage(
message: Message(
id: jobId,
headers: {},
payload: payload,
timestamp: DateTime.now().toIso8601String(),
),
exchangeName: '',
routingKey: queue ?? 'default',
);
return jobId;
});
}
// Cleanup method
void dispose() {
jobSubject.close();
}
}
// Additional interfaces and classes
abstract class HasMaxExceptions {
int? get maxExceptions;
}
abstract class HasFailOnTimeout {
bool get failOnTimeout;
}
abstract class HasTimeout {
Duration? get timeout;
}
abstract class HasDisplayName {
String displayName();
}
abstract class HasTries {
int? get tries;
}
abstract class HasBackoff {
dynamic get backoff;
}
abstract class HasRetryUntil {
DateTime? get retryUntil;
}
abstract class HasAfterCommit {
bool get afterCommit;
}
abstract class HasShouldBeEncrypted {
bool get shouldBeEncrypted;
}
abstract class Encrypter {
String encrypt(String data);
}
abstract class TransactionManager {
Future<T> addCallback<T>(Future<T> Function() callback);
}
// Add this mixin to the Queue class
mixin InteractsWithTime {
int secondsUntil(DateTime dateTime) {
return dateTime.difference(DateTime.now()).inSeconds;
}
int availableAt(Duration delay) {
return DateTime.now().add(delay).millisecondsSinceEpoch ~/ 1000;
}
}
// First, define the InvalidPayloadException class
class InvalidPayloadException implements Exception {
final String message;
InvalidPayloadException(this.message);
@override
String toString() => 'InvalidPayloadException: $message';
}

1
core/queue/lib/src/should_be_encrypted.dart
View file

@ -1 +0,0 @@
abstract class ShouldBeEncrypted {}

1
core/queue/lib/src/should_queue_after_commit.dart
View file

@ -1 +0,0 @@
abstract class ShouldQueueAfterCommit {}

25
core/queue/pubspec.yaml
View file

@ -1,25 +0,0 @@
name: angel3_queue
description: The Queue 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_mq: ^8.0.0
angel3_event_bus: ^8.0.0
angel3_reactivex: ^8.0.0
uuid: ^4.5.1
crypto: ^3.0.5
dev_dependencies:
build_runner: ^2.3.3
build_test: ^2.1.0
lints: ^3.0.0
mockito: ^5.0.0
test: ^1.24.0

317
core/queue/test/queue_test.dart
View file

@ -1,317 +0,0 @@
import 'package:test/test.dart';
import 'package:mockito/annotations.dart';
import 'package:mockito/mockito.dart';
import 'package:angel3_container/angel3_container.dart';
import 'package:angel3_event_bus/event_bus.dart';
import 'package:angel3_mq/mq.dart';
import 'package:angel3_queue/src/queue.dart';
import 'package:angel3_queue/src/job_queueing_event.dart';
import 'package:angel3_queue/src/job_queued_event.dart';
import 'package:angel3_queue/src/should_queue_after_commit.dart';
import 'queue_test.mocks.dart';
@GenerateMocks([Container, MQClient, TransactionManager, Queue])
void main() {
late MockContainer container;
late EventBus eventBus;
late MockMQClient mq;
late MockQueue queue;
late List<AppEvent> firedEvents;
setUpAll(() {
provideDummy<EventBus>(EventBus());
});
setUp(() {
container = MockContainer();
firedEvents = [];
eventBus = EventBus();
mq = MockMQClient();
queue = MockQueue();
// Inject the other mocks into the queue
// queue.container = container;
// queue.mq = mq;
when(queue.container).thenReturn(container);
when(queue.eventBus).thenReturn(eventBus);
when(queue.mq).thenReturn(mq);
when(queue.connectionName).thenReturn('default');
// Stub for shouldDispatchAfterCommit
when(queue.shouldDispatchAfterCommit(any)).thenReturn(false);
// Modify the createPayload stub
when(queue.createPayload(any, any, any)).thenAnswer((invocation) async {
if (invocation.positionalArguments[0] is Map &&
(invocation.positionalArguments[0] as Map).isEmpty) {
throw InvalidPayloadException('Invalid job: empty map');
}
return 'valid payload';
});
// Modify the push stub
when(queue.push(any, any, any)).thenAnswer((invocation) async {
final job = invocation.positionalArguments[0];
final data = invocation.positionalArguments[1];
final queueName = invocation.positionalArguments[2];
// Simulate firing events asynchronously
Future.microtask(() {
eventBus.fire(JobQueueingEvent(
queue.connectionName, queueName, job, 'payload', null));
eventBus.fire(JobQueuedEvent(
queue.connectionName, queueName, 'job_id', job, 'payload', null));
});
return 'pushed';
});
// Stub for enqueueUsing
when(queue.enqueueUsing(
any,
any,
any,
any,
any,
)).thenAnswer((invocation) async {
final job = invocation.positionalArguments[0];
final payload = invocation.positionalArguments[1];
final queueName = invocation.positionalArguments[2];
final delay = invocation.positionalArguments[3];
final callback = invocation.positionalArguments[4] as Function;
eventBus.fire(JobQueueingEvent(
queue.connectionName, queueName, job, payload, delay));
final result = await callback(payload, queueName, delay);
eventBus.fire(JobQueuedEvent(
queue.connectionName, queueName, result, job, payload, delay));
return result;
});
// Stub for pushOn
when(queue.pushOn(any, any, any)).thenAnswer((invocation) async {
final queueName = invocation.positionalArguments[0];
final job = invocation.positionalArguments[1];
final data = invocation.positionalArguments[2];
return queue.push(job, data, queueName);
});
// Modify the laterOn stub
when(queue.laterOn(any, any, any, any)).thenAnswer((invocation) async {
final queueName = invocation.positionalArguments[0];
final delay = invocation.positionalArguments[1];
final job = invocation.positionalArguments[2];
final data = invocation.positionalArguments[3];
// Directly return 'pushed later' instead of calling later
return 'pushed later';
});
// Add a stub for bulk
when(queue.bulk(any, any, any)).thenAnswer((invocation) async {
final jobs = invocation.positionalArguments[0] as List;
for (var job in jobs) {
await queue.push(job, invocation.positionalArguments[1],
invocation.positionalArguments[2]);
}
});
// Stub for later
when(queue.later(any, any, any, any)).thenAnswer((invocation) async {
final delay = invocation.positionalArguments[0];
final job = invocation.positionalArguments[1];
final data = invocation.positionalArguments[2];
final queueName = invocation.positionalArguments[3];
final payload =
await queue.createPayload(job, queueName ?? 'default', data);
return queue.enqueueUsing(
job, payload, queueName, delay, (p, q, d) async => 'delayed_job_id');
});
when(container.has<EventBus>()).thenReturn(true);
when(container.has<TransactionManager>()).thenReturn(false);
when(container.make<EventBus>()).thenReturn(eventBus);
// Capture fired events
eventBus.on().listen((event) {
firedEvents.add(event);
print("Debug: Event fired - ${event.runtimeType}");
});
// Setup for MQClient mock
when(mq.sendMessage(
message: anyNamed('message'),
exchangeName: anyNamed('exchangeName'),
routingKey: anyNamed('routingKey'),
)).thenAnswer((_) {
print("Debug: Mock sendMessage called");
});
});
test('pushOn calls push with correct arguments', () async {
final result = await queue.pushOn('test_queue', 'test_job', 'test_data');
expect(result, equals('pushed'));
verify(queue.push('test_job', 'test_data', 'test_queue')).called(1);
});
test('laterOn calls later with correct arguments', () async {
final result = await queue.laterOn(
'test_queue', Duration(minutes: 5), 'test_job', 'test_data');
expect(result, equals('pushed later'));
// We're not actually calling 'later' in our stub, so we shouldn't verify it
verify(queue.laterOn(
'test_queue', Duration(minutes: 5), 'test_job', 'test_data'))
.called(1);
});
test('bulk pushes multiple jobs', () async {
await queue.bulk(['job1', 'job2', 'job3'], 'test_data', 'test_queue');
verify(queue.push('job1', 'test_data', 'test_queue')).called(1);
verify(queue.push('job2', 'test_data', 'test_queue')).called(1);
verify(queue.push('job3', 'test_data', 'test_queue')).called(1);
});
test('createPayload throws InvalidPayloadException for invalid job', () {
expect(() => queue.createPayload({}, 'test_queue'),
throwsA(isA<InvalidPayloadException>()));
});
test('shouldDispatchAfterCommit returns correct value', () {
when(queue.shouldDispatchAfterCommit(any)).thenReturn(false);
expect(queue.shouldDispatchAfterCommit({}), isFalse);
when(queue.shouldDispatchAfterCommit(any)).thenReturn(true);
expect(queue.shouldDispatchAfterCommit({}), isTrue);
});
test('push enqueues job and fires events', () async {
final job = 'test_job';
final data = 'test_data';
final queueName = 'test_queue';
print("Debug: Before push");
final result = await queue.push(job, data, queueName);
print("Debug: After push");
// Wait for all events to be processed
await Future.delayed(Duration(milliseconds: 100));
expect(result, equals('pushed'));
verify(queue.push(job, data, queueName)).called(1);
// Filter out EmptyEvents
final significantEvents =
firedEvents.where((event) => event is! EmptyEvent).toList();
// Print fired events for debugging
print("Fired events (excluding EmptyEvents):");
for (var event in significantEvents) {
print("${event.runtimeType}: ${event.toString()}");
}
// Verify fired events
expect(significantEvents.where((event) => event is JobQueueingEvent).length,
equals(1),
reason: "JobQueueingEvent was not fired exactly once");
expect(significantEvents.where((event) => event is JobQueuedEvent).length,
equals(1),
reason: "JobQueuedEvent was not fired exactly once");
});
}
class TestQueue extends Queue {
List<dynamic> pushedJobs = [];
TestQueue(Container container, EventBus eventBus, MQClient mq)
: super(container, eventBus, mq);
@override
Future<dynamic> push(dynamic job, [dynamic data = '', String? queue]) async {
pushedJobs.add(job);
final payload = await createPayload(job, queue ?? 'default', data);
return enqueueUsing(job, payload, queue, null, (payload, queue, _) async {
final jobId = 'test-job-id';
mq.sendMessage(
message: Message(
id: jobId,
headers: {},
payload: payload,
timestamp: DateTime.now().toIso8601String(),
),
exchangeName: '',
routingKey: queue ?? 'default',
);
return jobId;
});
}
@override
Future<dynamic> later(Duration delay, dynamic job,
[dynamic data = '', String? queue]) async {
return 'pushed later';
}
@override
Future<String> createPayload(dynamic job, String queue,
[dynamic data = '']) async {
if (job is Map && job.isEmpty) {
throw InvalidPayloadException('Invalid job: empty map');
}
return 'valid payload';
}
@override
bool shouldDispatchAfterCommit(dynamic job) {
if (job is ShouldQueueAfterCommit) {
return true;
}
return dispatchAfterCommit;
}
@override
Future<dynamic> enqueueUsing(
dynamic job,
String payload,
String? queue,
Duration? delay,
Future<dynamic> Function(String, String?, Duration?) callback,
) async {
eventBus.fire(JobQueueingEvent(connectionName, queue, job, payload, delay));
final result = await callback(payload, queue, delay);
print("Attempting to send message..."); // Debug print
mq.sendMessage(
message: Message(
id: 'test-id',
headers: {},
payload: payload,
timestamp: DateTime.now().toIso8601String(),
),
exchangeName: '',
routingKey: queue ?? 'default',
);
print("Message sent."); // Debug print
eventBus.fire(
JobQueuedEvent(connectionName, queue, result, job, payload, delay));
return result;
}
}
// class DummyEventBus implements EventBus {
// List<AppEvent> firedEvents = [];
// @override
// Future<void> fire(AppEvent event) async {
// firedEvents.add(event);
// }
// @override
// dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
// }
class InvalidPayloadException implements Exception {
final String message;
InvalidPayloadException(this.message);
@override
String toString() => 'InvalidPayloadException: $message';
}
class MockShouldQueueAfterCommit implements ShouldQueueAfterCommit {}

10
core/support/.gitignore vendored
View file

@ -1,10 +0,0 @@
# 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
*.mocks.dart
*.reflectable.dart

3
core/support/CHANGELOG.md
View file

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

10
core/support/LICENSE.md
View file

@ -1,10 +0,0 @@
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.

1
core/support/README.md
View file

@ -1 +0,0 @@
<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>

30
core/support/analysis_options.yaml
View file

@ -1,30 +0,0 @@
# 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

8
core/support/build.yaml
View file

@ -1,8 +0,0 @@
targets:
$default:
builders:
reflectable:
generate_for:
- test/**.dart
options:
formatted: true

13
core/support/lib/service_provider.dart
View file

@ -1,13 +0,0 @@
/*
* This file is part of the Protevus Platform.
*
* (C) Protevus <developers@protevus.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
library;
export '../src/service_provider.dart';
export '../src/discoverable_service_provider.dart';

54
core/support/lib/src/discoverable_service_provider.dart
View file

@ -1,54 +0,0 @@
/*
* This file is part of the Protevus Platform.
*
* (C) Protevus <developers@protevus.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
import 'package:reflectable/reflectable.dart';
/// A reflectable class that provides capabilities for discovering services.
///
/// This class extends [Reflectable] and enables invoking, typing, and reflected type capabilities.
/// It is used to facilitate service discovery and reflection-based operations in the application.
///
/// The capabilities provided by this class are:
/// - [invokingCapability]: Allows invoking methods on reflected instances.
/// - [typingCapability]: Enables access to type information of reflected instances.
/// - [reflectedTypeCapability]: Provides access to the reflected type of instances.
/// - [newInstanceCapability]: Allows creation of new instances of reflected types.
///
/// Usage:
/// ```dart
/// const provider = DiscoverableServiceProvider();
/// // Use the provider to perform reflection-based operations
/// ```
///
/// This class is typically used in conjunction with the [discoverableServiceProvider]
/// constant to provide a shared instance for reflection operations throughout the application.
class DiscoverableServiceProvider extends Reflectable {
const DiscoverableServiceProvider()
: super(
invokingCapability,
typingCapability,
declarationsCapability,
reflectedTypeCapability,
newInstanceCapability,
superclassQuantifyCapability);
}
/// A constant instance of [DiscoverableServiceProvider].
///
/// This constant provides a single, shared instance of the [DiscoverableServiceProvider] class.
/// It can be used throughout the application to access the reflection capabilities
/// defined in the [DiscoverableServiceProvider] class, such as invoking methods,
/// accessing type information, and working with reflected types.
///
/// Usage:
/// ```dart
/// var reflector = discoverableServiceProvider;
/// // Use the reflector to perform reflection operations
/// ```
const discoverableServiceProvider = DiscoverableServiceProvider();

887
core/support/lib/src/service_provider.dart
View file

@ -1,887 +0,0 @@
/*
* This file is part of the Protevus Platform.
*
* (C) Protevus <developers@protevus.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
import 'package:angel3_framework/angel3_framework.dart';
import 'package:angel3_container/angel3_container.dart';
import 'package:meta/meta.dart';
import 'package:logging/logging.dart';
import 'package:reflectable/reflectable.dart';
import 'discoverable_service_provider.dart';
@discoverableServiceProvider
abstract class ServiceProvider {
/// Static variable to hold the logger instance for the ServiceProvider class.
/// This logger is lazily initialized and can be accessed via the `logger` getter.
/// It can also be set explicitly for testing purposes using the `setLogger` method.
static Logger? _logger;
/// Returns the logger for the ServiceProvider class.
///
/// If a logger has not been explicitly set using the `setLogger` method,
/// this getter creates and returns a new Logger instance with the name 'ServiceProvider'.
///
/// If a logger has been set, it returns that logger instance.
///
/// This approach allows for lazy initialization of the logger and provides
/// flexibility for testing by allowing the logger to be explicitly set when needed.
static Logger get logger => _logger ?? Logger('ServiceProvider');
/// Constructs a [ServiceProvider] instance.
///
/// The [priority] parameter determines the order in which service providers are registered.
/// Service providers with higher priority values are registered before those with lower priority values.
/// By default, the priority is set to 0.
///
/// Example:
/// ```dart
/// class MyServiceProvider extends ServiceProvider {
/// MyServiceProvider() : super(priority: 10);
/// }
/// ```
const ServiceProvider({this.priority = 0});
/// The priority of this service provider. Higher priority providers are registered first.
///
/// This integer value determines the order in which service providers are registered
/// within the Angel. Service providers with higher priority values are processed
/// before those with lower priority values. This allows for fine-grained control over
/// the initialization sequence of services, ensuring that dependencies are properly
/// set up before they are needed by other components.
///
/// For example, a service provider with a priority of 10 will be registered before
/// a service provider with a priority of 5. If two service providers have the same
/// priority, their registration order is not guaranteed.
final int priority;
/// Sets a custom logger for the ServiceProvider class.
///
/// This method allows changing the logger used by the ServiceProvider class,
/// which is particularly useful for testing purposes. It replaces the default
/// logger with the provided custom logger.
///
/// Parameters:
/// - logger: The custom Logger instance to be used.
///
/// Example:
/// ```dart
/// ServiceProvider.setLogger(MockLogger());
/// ```
///
/// Note: This method is marked as `@visibleForTesting` to indicate that it
/// should only be used in test environments.
@visibleForTesting
static void setLogger(Logger logger) {
_logger = logger;
print('Logger set to: $logger'); // Add this line
}
/// Registers the service provider with the given container and Angel.
///
/// This method is responsible for registering the service provider's dependencies
/// and services into the provided [container]. It also has access to the [app]
/// instance for any Angel-specific configurations.
///
/// The method is annotated with [@mustCallSuper], indicating that subclasses
/// overriding this method must call the superclass implementation.
///
/// This implementation logs the registration process, performs a sample
/// registration (which should be replaced with actual logic in subclasses),
/// and handles any exceptions that occur during the registration process.
///
/// Parameters:
/// - [container]: The dependency injection container to register with.
/// - [app]: The Angel instance, providing access to app-wide configurations.
///
/// Throws:
/// Any exceptions caught during the registration process are logged
/// and then re-thrown.
@mustCallSuper
void register(Container container, Angel app) {
try {
logger.info('Registering ${runtimeType.toString()}');
// Actual registration logic
container.registerSingleton<String>('test', as: null);
} catch (e, stackTrace) {
print('Caught exception in register: $e'); // Add this line
logger.severe(
'Error registering ${runtimeType.toString()}', e, stackTrace);
print('Logged severe message in register'); // Add this line
rethrow;
}
}
/// Boots the service provider, performing necessary setup tasks.
///
/// This method is responsible for initializing the service provider and
/// performing any required setup tasks. It's called during the Angel's
/// boot process.
///
/// The method is annotated with [@mustCallSuper], indicating that subclasses
/// overriding this method must call the superclass implementation.
///
/// This implementation logs the booting process, performs a sample
/// configuration (which should be replaced with actual logic in subclasses),
/// and handles any exceptions that occur during the booting process.
///
/// Parameters:
/// - [app]: The Angel instance, providing access to app-wide configurations.
///
/// Throws:
/// Any exceptions caught during the booting process are logged
/// and then re-thrown.
///
/// Returns:
/// A [Future] that completes when the boot process is finished.
@mustCallSuper
Future<void> boot(Angel app) async {
try {
logger.info('Booting ${runtimeType.toString()}');
// Actual booting logic
await app.configure((_) => null);
} catch (e, stackTrace) {
print('Caught exception in boot: $e'); // Add this line
logger.severe('Error booting ${runtimeType.toString()}', e, stackTrace);
print('Logged severe message in boot'); // Add this line
rethrow;
}
}
/// Indicates whether the service provider is deferred.
///
/// This getter returns a boolean value that determines if the service provider
/// should be loaded in a deferred manner. By default, it returns `false`,
/// meaning the service provider is not deferred.
///
/// Deferred loading can be useful for optimizing Angel startup time
/// by delaying the loading of certain services until they are actually needed.
///
/// Subclasses can override this getter to implement custom deferral logic.
///
/// Returns:
/// A boolean value. `false` indicates the service provider is not deferred,
/// while `true` would indicate it is deferred.
bool get isDeferred => false;
/// Returns a list of types that this service provider provides.
///
/// This getter is intended to be overridden by subclasses to specify
/// the types of services or dependencies that the service provider
/// makes available to the Angel.
///
/// By default, it returns an empty list, indicating that the base
/// ServiceProvider class doesn't provide any specific types.
///
/// Subclasses should override this getter to return a list of
/// [Type] objects representing the services they provide.
///
/// Example:
/// ```dart
/// @override
/// List<Type> get provides => [DatabaseService, LoggingService];
/// ```
///
/// Returns:
/// A [List] of [Type] objects representing the provided services.
List<Type> get provides => const [];
/// Determines whether the service provider should be registered.
///
/// This method is used to conditionally register the service provider based on
/// the current state of the Angel. By default, it always returns `true`,
/// meaning the service provider will be registered in all cases.
///
/// Subclasses can override this method to implement custom logic for determining
/// when the service provider should be registered. This can be useful for
/// conditional registration based on environment variables, configuration settings,
/// or other Angel-specific criteria.
///
/// Parameters:
/// - [app]: The current [Angel] instance, which can be used to access
/// Angel-wide settings or configurations.
///
/// Returns:
/// A boolean value. `true` indicates that the service provider should be
/// registered, while `false` indicates it should not be registered.
///
/// Example:
/// ```dart
/// @override
/// bool when(Angel app) {
/// return app.environment == 'production';
/// }
/// ```
bool when(Angel app) => true;
/// Determines if the service provider is ready to be registered with the Angel.
///
/// This method checks two conditions:
/// 1. The service provider is not deferred (checked via the [isDeferred] property).
/// 2. The [when] method returns true for the given [app].
///
/// If both conditions are met, the method returns true, indicating that the
/// service provider is ready to be registered.
///
/// Parameters:
/// - [app]: The [Angel] instance to check against.
///
/// Returns:
/// A boolean value. `true` if the service provider is ready to be registered,
/// `false` otherwise.
bool isReadyToRegister(Angel app) => !isDeferred && when(app);
/// Registers a singleton instance of type [T] with the provided [container].
///
/// This method ensures that the same [instance] of type [T] is always returned
/// when requested from the container. This is useful for objects that should
/// maintain a single state throughout the Angel's lifecycle.
///
/// Parameters:
/// - [container]: The dependency injection container to register with.
/// - [instance]: The instance of type [T] to be registered as a singleton.
///
/// Example:
/// ```dart
/// final myService = MyService();
/// singleton<MyService>(container, myService);
/// ```
///
/// After registration, any request for type [T] from the container will return
/// the same [instance].
void singleton<T>(Container container, T instance) {
container.registerSingleton<T>(instance);
}
/// Registers a factory that always returns the same instance of type [T].
///
/// This method creates a singleton-like behavior using a factory function.
/// The first time the type [T] is requested from the container, the [factory]
/// function is called to create an instance. This instance is then cached and
/// returned for all subsequent requests.
///
/// Parameters:
/// - [container]: The dependency injection container to register with.
/// - [factory]: A function that takes a [Container] and returns an instance of [T].
///
/// Example:
/// ```dart
/// singletonFactory<Database>(container, (c) => Database.connect());
/// ```
///
/// After registration, any request for type [T] from the container will return
/// the same instance, created by the [factory] function on the first request.
void singletonFactory<T>(Container container, T Function(Container) factory) {
T? instance;
container.registerFactory<T>((c) {
instance ??= factory(c);
return instance!;
});
}
/// Registers a factory function for type [T] with the provided [container].
///
/// This method allows you to register a factory function that creates instances
/// of type [T]. The factory function will be called each time an instance of [T]
/// is requested from the container.
///
/// Parameters:
/// - [container]: The dependency injection container to register with.
/// - [factory]: A function that takes a [Container] and returns an instance of [T].
///
/// Example:
/// ```dart
/// bind<Logger>(container, (c) => Logger('MyLogger'));
/// ```
///
/// After registration, you can request an instance of [T] from the container,
/// and the [factory] function will be called to create a new instance each time.
void bind<T>(Container container, T Function(Container) factory) {
container.registerFactory<T>(factory);
}
/// Registers an instance of type [T] with the provided [container].
///
/// This method creates a factory that always returns the same [instance] of type [T].
/// It's useful for registering existing objects into the dependency injection container.
///
/// Parameters:
/// - [container]: The dependency injection container to register with.
/// - [instance]: The instance of type [T] to be registered.
///
/// Example:
/// ```dart
/// final myService = MyService();
/// bindInstance<MyService>(container, myService);
/// ```
///
/// After registration, any request for type [T] from the container will return
/// the same [instance].
void bindInstance<T>(Container container, T instance) {
container.registerFactory<T>((Container c) => instance);
}
/// Registers a factory function for type [T] with the provided [container], but only if a binding for [T] doesn't already exist.
///
/// This method checks if the container already has a binding for type [T]. If it doesn't,
/// it registers the provided [factory] function. This is useful for providing default
/// implementations while allowing them to be overridden.
///
/// Parameters:
/// - [container]: The dependency injection container to register with.
/// - [factory]: A function that takes a [Container] and returns an instance of [T].
///
/// Example:
/// ```dart
/// bindIf<Logger>(container, (c) => Logger('DefaultLogger'));
/// ```
///
/// If [T] is not already registered in the container, the [factory] function will be
/// registered. Otherwise, the existing binding will be preserved.
void bindIf<T>(Container container, T Function(Container) factory) {
if (!container.has<T>()) {
container.registerFactory<T>(factory);
}
}
/// Registers multiple factory bindings at once with the provided container.
///
/// This method allows for batch registration of multiple factory functions
/// for different types in a single call. It iterates through the provided
/// [bindings] map and registers each factory function with its corresponding type.
///
/// Parameters:
/// - [container]: The dependency injection container to register with.
/// - [bindings]: A map where each key is a Type and each value is a factory function.
///
/// Example:
/// ```dart
/// bindMultiple(container, {
/// Logger: (c) => Logger('MyLogger'),
/// Database: (c) => Database.connect(),
/// UserService: (c) => UserService(c.make<Database>())
/// });
/// ```
///
/// Note: The factory functions in the [bindings] map should match the
/// signature `T Function(Container)` where T is the type being registered.
void bindMultiple(Container container, Map<Type, Function> bindings) {
bindings.forEach((type, factory) {
container.registerFactory(factory as Function(Container));
});
}
/// Registers an alias for an existing binding in the container.
///
/// This method creates a new binding of type [A] that resolves to the same
/// instance as an existing binding of type [T]. This is useful when you want
/// to refer to the same service using different types, such as when working
/// with interfaces and implementations.
///
/// Parameters:
/// - [container]: The dependency injection container to register the alias with.
///
/// Type Parameters:
/// - [T]: The original type that is already bound in the container.
/// - [A]: The alias type that will be registered to resolve to the same instance as [T].
///
/// Example:
/// ```dart
/// // Assuming UserService is already registered
/// alias<UserService, IUserService>(container);
/// ```
///
/// After calling this method, requests for type [A] from the container will
/// return the same instance as requests for type [T].
///
/// Note: This method assumes that [T] is already registered in the container
/// and that instances of [T] can be cast to [A]. If these conditions are not met,
/// it may result in runtime errors.
void alias<T, A>(Container container) {
container.registerFactory<A>((c) => c.make<T>() as A);
}
/// Registers a scoped instance of type [T] with the provided [container].
///
/// A scoped instance is created once per scope and reused within that scope.
/// This is useful for objects that should have a single instance per request
/// or transaction, but not shared across the entire Angel.
///
/// Parameters:
/// - [container]: The dependency injection container to register with.
/// - [factory]: A function that takes a [Container] and returns an instance of [T].
///
/// Example:
/// ```dart
/// scoped<RequestContext>(container, (c) => RequestContext());
/// ```
///
/// After registration, a new instance of [T] will be created for each scope,
/// but within the same scope, the same instance will be reused.
void scoped<T>(Container container, T Function(Container) factory) {
container.registerScoped<T>(factory);
}
/// Registers a transient factory for type [T] with the provided [container].
///
/// A transient factory creates a new instance every time it is requested, even within the same scope.
/// This is useful for objects that should not be shared and need a fresh instance each time they are used.
///
/// Parameters:
/// - [container]: The dependency injection container to register with.
/// - [factory]: A function that takes a [Container] and returns a new instance of [T].
///
/// Example:
/// ```dart
/// transient<Logger>(container, (c) => Logger('NewLogger'));
/// ```
///
/// After registration, each request for type [T] from the container will return a new instance
/// created by the [factory] function, regardless of the scope or previous requests.
void transient<T>(Container container, T Function(Container) factory) {
container.registerTransient<T>(factory);
}
/// Registers a constant value of type [T] with the provided [container].
///
/// This method allows you to register a constant value that will always be
/// returned when requesting an instance of type [T] from the container.
/// The registered value cannot be changed after registration.
///
/// Parameters:
/// - [container]: The dependency injection container to register with.
/// - [value]: The constant value of type [T] to be registered.
///
/// Example:
/// ```dart
/// constant<int>(container, 42);
/// constant<String>(container, 'AppName');
/// ```
///
/// After registration, any request for type [T] from the container will
/// always return the registered [value].
void constant<T>(Container container, T value) {
container.registerConstant<T>(value);
}
/// Registers a lazy-loaded singleton of type [T] with the provided [container].
///
/// This method registers a factory function that creates a singleton instance of type [T]
/// only when it is first requested from the container. Subsequent requests will return
/// the same instance.
///
/// Parameters:
/// - [container]: The dependency injection container to register with.
/// - [factory]: A function that takes a [Container] and returns an instance of [T].
///
/// The lazy initialization can help improve performance by deferring the creation
/// of the singleton until it's actually needed.
///
/// Example:
/// ```dart
/// lazySingleton<Database>(container, (c) => Database.connect());
/// ```
///
/// After registration, the first request for type [T] from the container will trigger
/// the [factory] function to create the instance. All subsequent requests will return
/// the same instance without calling the factory again.
void lazySingleton<T>(Container container, T Function(Container) factory) {
container.registerLazySingleton<T>(factory);
}
}
/// A container that manages service providers for dependency injection.
///
/// This class extends the base [Container] and adds functionality to handle
/// [ServiceProvider] instances. It allows for adding, configuring, booting,
/// and discovering service providers.
///
/// The container maintains an internal list of providers, sorted by their priority.
/// It provides methods to add providers manually, configure and boot them with
/// an [Angel] instance, and automatically discover providers using reflection.
///
/// Key features:
/// - Add providers with priority-based sorting
/// - Configure providers with the Angel instance
/// - Boot providers when the Angel starts
/// - Automatically discover and instantiate providers using reflection
///
/// Usage:
/// ```dart
/// var container = ServiceProviderContainer(myReflector);
/// container.addProvider(MyServiceProvider());
/// container.configureProviders(app);
/// container.bootProviders(app);
/// container.discoverProviders();
/// ```
///
/// This container is designed to work with the [ServiceProvider] abstract class
/// and integrates with the Angel framework's [Angel] class.
class ServiceProviderContainer extends Container {
/// Static logger instance for the ServiceProviderContainer class.
///
/// This logger is used throughout the ServiceProviderContainer class to log
/// information, warnings, and errors related to service provider operations.
/// It is initialized with the name 'ServiceProviderContainer' to easily
/// identify log messages originating from this class.
static final Logger _logger = Logger('ServiceProviderContainer');
/// A list to store all registered [ServiceProvider] instances.
///
/// This private field maintains an ordered collection of service providers
/// that have been added to the container. The order of providers in this list
/// is significant, as it determines the sequence in which they are configured
/// and booted.
///
/// Service providers are typically added to this list through the [addProvider]
/// method, which also sorts them based on their priority.
final List<ServiceProvider> _providers = [];
/// Constructs a new [ServiceProviderContainer] instance.
///
/// This constructor initializes a [ServiceProviderContainer] with the provided [reflector].
/// The [reflector] is passed to the superclass constructor and is used for reflection-based
/// operations, such as discovering and instantiating service providers.
///
/// Parameters:
/// - [reflector]: An instance of a reflector that complies with the requirements of the
/// superclass [Container]. This reflector is used for introspection and dynamic
/// instantiation of service providers.
///
/// Example:
/// ```dart
/// var container = ServiceProviderContainer(myReflector);
/// ```
ServiceProviderContainer(super.reflector);
/// Adds a [ServiceProvider] to the container and sorts the providers by priority.
///
/// This method performs the following operations:
/// 1. Adds the given [provider] to the internal list of providers.
/// 2. Sorts the list of providers based on their priority in descending order.
/// 3. Logs the addition of the provider.
///
/// If any exception occurs during this process, it logs the error and rethrows the exception.
///
/// Parameters:
/// - [provider]: The [ServiceProvider] instance to be added to the container.
///
/// Throws:
/// Any exception that occurs during the addition or sorting process.
///
/// Example:
/// ```dart
/// container.addProvider(MyServiceProvider());
/// ```
void addProvider(ServiceProvider provider) {
try {
_providers.add(provider);
_providers.sort((a, b) => b.priority.compareTo(a.priority));
_logger.info('Added provider: ${provider.runtimeType.toString()}');
} catch (e, stackTrace) {
_logger.severe('Error adding provider', e, stackTrace);
rethrow;
}
}
/// Boots all registered service providers that are ready to be registered.
///
/// This method iterates through all registered service providers and calls
/// their `boot` method if they are ready to be registered. It handles the
/// booting process for each provider and logs any errors that occur.
///
/// Parameters:
/// - [app]: The [Angel] instance that providers will use for booting.
///
/// Throws:
/// Any exception that occurs during the booting process of a provider.
/// The exception is logged before being rethrown.
///
/// Note:
/// - Only providers that return true for `isReadyToRegister(app)` are booted.
/// - If an error occurs while booting a provider, it's logged and rethrown,
/// which may interrupt the booting process for subsequent providers.
void bootProviders(Angel app) {
for (var provider in _providers) {
if (provider.isReadyToRegister(app)) {
try {
provider.boot(app);
} catch (e, stackTrace) {
_logger.severe(
'Error booting provider: ${provider.runtimeType}', e, stackTrace);
rethrow;
}
}
}
}
/// Configures all registered service providers that are ready to be registered.
///
/// This method iterates through all registered service providers and calls
/// their `register` method if they are ready to be registered. It handles the
/// configuration process for each provider and logs any errors that occur.
///
/// Parameters:
/// - [app]: The [Angel] instance that providers will use for configuration.
///
/// Throws:
/// Any exception that occurs during the configuration process of a provider.
/// The exception is logged before being rethrown.
///
/// Note:
/// - Only providers that return true for `isReadyToRegister(app)` are configured.
/// - If an error occurs while configuring a provider, it's logged and rethrown,
/// which may interrupt the configuration process for subsequent providers.
void configureProviders(Angel app) {
for (var provider in _providers) {
if (provider.isReadyToRegister(app)) {
try {
provider.register(this, app);
} catch (e, stackTrace) {
_logger.severe('Error configuring provider: ${provider.runtimeType}',
e, stackTrace);
rethrow;
}
}
}
}
/// Discovers and adds service providers to the container using reflection.
///
/// This method iterates through all classes annotated with [discoverableServiceProvider],
/// checks if they are non-abstract subclasses of [ServiceProvider], and if so,
/// instantiates and adds them to the container.
///
/// The discovery process includes:
/// 1. Iterating through annotated classes.
/// 2. Checking if each class is a non-abstract ServiceProvider.
/// 3. Instantiating the provider and adding it to the container.
/// 4. Logging successful additions and any errors encountered.
///
/// If an error occurs during the instantiation of a specific provider,
/// it is logged as a warning, and the method continues with the next provider.
/// If a general error occurs during the discovery process, it is logged
/// as a severe error and rethrown.
///
/// This method is useful for automatically registering service providers
/// without manual instantiation and addition.
///
/// Throws:
/// Any exception that occurs during the general discovery process,
/// excluding individual provider instantiation errors.
void discoverProviders() {
try {
for (var classMirror in discoverableServiceProvider.annotatedClasses) {
if (_isServiceProvider(classMirror) && !classMirror.isAbstract) {
try {
var provider = classMirror.newInstance('', []) as ServiceProvider;
addProvider(provider);
_logger.info(
'Discovered and added provider: ${classMirror.simpleName}');
} catch (e, stackTrace) {
_logger.warning(
'Error instantiating provider: ${classMirror.simpleName}',
e,
stackTrace);
// Continue to the next provider instead of rethrowing
}
}
}
} catch (e, stackTrace) {
_logger.severe('Error discovering providers', e, stackTrace);
rethrow;
}
}
/// Determines if a given ClassMirror represents a subclass of ServiceProvider.
///
/// This method traverses the inheritance hierarchy of the provided [classMirror]
/// to check if it is a subclass of ServiceProvider. It does this by comparing
/// the simple name of each superclass with 'ServiceProvider'.
///
/// Parameters:
/// - [classMirror]: The ClassMirror to check.
///
/// Returns:
/// true if the class is a subclass of ServiceProvider, false otherwise.
///
/// Note:
/// This method assumes that the ServiceProvider class is named exactly
/// 'ServiceProvider'. It will not detect classes that implement an interface
/// named ServiceProvider or classes that are renamed through imports.
bool _isServiceProvider(ClassMirror classMirror) {
ClassMirror? currentMirror = classMirror;
while (currentMirror != null) {
if (currentMirror.simpleName == 'ServiceProvider') {
return true;
}
currentMirror = currentMirror.superclass;
}
return false;
}
}
/// Extension on [Angel] to provide service provider functionality.
extension ServiceProviderExtension on Angel {
/// A static logger instance for the ServiceProviderExtension.
///
/// This logger is used throughout the ServiceProviderExtension to log
/// information, warnings, and errors related to service provider operations.
/// It is initialized with the name 'ServiceProviderExtension' to easily
/// identify log messages originating from this extension.
static final Logger _logger = Logger('ServiceProviderExtension');
/// Adds a [ServiceProvider] to the Angel's container.
///
/// This method attempts to add the given [provider] to the Angel's
/// container. It performs the following steps:
///
/// 1. Checks if the Angel's container is an instance of [ServiceProviderContainer].
/// 2. If it is, adds the provider to the container using the [addProvider] method.
/// 3. If it's not, throws an exception indicating that the container must be a [ServiceProviderContainer].
///
/// If any error occurs during this process, it logs a severe error message
/// and rethrows the exception.
///
/// Parameters:
/// - [provider]: The [ServiceProvider] instance to be added to the container.
///
/// Throws:
/// - [Exception] if the Angel's container is not a [ServiceProviderContainer].
/// - Any other exception that occurs during the process of adding the provider.
///
/// Usage:
/// ```dart
/// app.useServiceProvider(MyServiceProvider());
/// ```
void useServiceProvider(ServiceProvider provider) {
try {
if (container is ServiceProviderContainer) {
(container as ServiceProviderContainer).addProvider(provider);
} else {
throw Exception('Angel container must be a ServiceProviderContainer');
}
} catch (e, stackTrace) {
_logger.severe('Error using service provider', e, stackTrace);
rethrow;
}
}
/// Configures all registered service providers in the Angel's container.
///
/// This method attempts to configure the service providers by calling the
/// `configureProviders` method on the Angel's container. It performs
/// the following steps:
///
/// 1. Checks if the Angel's container is an instance of [ServiceProviderContainer].
/// 2. If it is, calls the `configureProviders` method on the container, passing `this` (the Angel).
/// 3. If it's not, throws an exception indicating that the container must be a [ServiceProviderContainer].
///
/// If any error occurs during this process, it logs a severe error message
/// and rethrows the exception.
///
/// Throws:
/// - [Exception] if the Angel's container is not a [ServiceProviderContainer].
/// - Any other exception that occurs during the configuration process.
///
/// Usage:
/// ```dart
/// app.configureProviders();
/// ```
void configureProviders() {
try {
if (container is ServiceProviderContainer) {
(container as ServiceProviderContainer).configureProviders(this);
} else {
throw Exception('Angel container must be a ServiceProviderContainer');
}
} catch (e, stackTrace) {
_logger.severe('Error configuring providers', e, stackTrace);
rethrow;
}
}
/// Registers and boots all service providers in the Angel's container.
///
/// This method performs the following operations:
/// 1. Checks if the Angel's container is an instance of [ServiceProviderContainer].
/// 2. If it is, it casts the container to [ServiceProviderContainer].
/// 3. Calls [configureProviders] on the container to register the providers.
/// 4. Calls [bootProviders] on the container to initialize the providers.
///
/// If the container is not a [ServiceProviderContainer], it throws an exception.
///
/// This method is typically called after all service providers have been added
/// to the container, either manually or through discovery.
///
/// Throws:
/// - [Exception] if the Angel's container is not a [ServiceProviderContainer].
/// - Any other exception that occurs during the registration or booting process.
/// These exceptions are logged before being rethrown.
///
/// Usage:
/// ```dart
/// app.registerAndBootProviders();
/// ```
void registerAndBootProviders() {
try {
if (container is ServiceProviderContainer) {
var providerContainer = container as ServiceProviderContainer;
providerContainer.configureProviders(this);
providerContainer.bootProviders(this);
} else {
throw Exception('Angel container must be a ServiceProviderContainer');
}
} catch (e, stackTrace) {
_logger.severe('Error registering and booting providers', e, stackTrace);
rethrow;
}
}
/// Discovers, registers, and boots all service providers in the Angel.
///
/// This method performs a complete lifecycle for service providers:
/// 1. Discovers providers using reflection (if the container supports it).
/// 2. Configures the discovered providers.
/// 3. Registers and boots the providers.
///
/// The process includes:
/// - Checking if the Angel's container is a [ServiceProviderContainer].
/// - Calling [discoverProviders] to find and instantiate providers.
/// - Calling [configureProviders] to set up the providers.
/// - Calling [registerAndBootProviders] to finalize provider initialization.
///
/// If the container is not a [ServiceProviderContainer], an exception is thrown.
///
/// This method is typically called once during Angel startup to set up
/// all service providers automatically.
///
/// Throws:
/// - [Exception] if the Angel's container is not a [ServiceProviderContainer].
/// - Any exceptions that occur during the discovery, configuration, or booting process.
/// These exceptions are logged before being rethrown.
///
/// Usage:
/// ```dart
/// app.discoverAndRegisterProviders();
/// ```
void discoverAndRegisterProviders() {
try {
if (container is ServiceProviderContainer) {
(container as ServiceProviderContainer).discoverProviders();
configureProviders();
registerAndBootProviders();
} else {
throw Exception('Angel container must be a ServiceProviderContainer');
}
} catch (e, stackTrace) {
_logger.severe(
'Error discovering and registering providers', e, stackTrace);
rethrow;
}
}
}

34
core/support/pubspec.yaml
View file

@ -1,34 +0,0 @@
name: angel3_support
description: The Support 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_framework: ^7.0.0
angel3_configuration: ^9.0.0
angel3_container: ^9.0.0
angel3_migration: ^9.0.0
angel3_orm: ^9.0.0
angel3_route: ^9.0.0
angel3_validate: ^9.0.0
collection: ^1.15.0
file: ^7.0.0
get_it: ^7.7.0
injectable: ^2.4.4
logging: ^1.0.0
meta: ^1.3.0
mockito: ^5.4.4
path: ^1.8.0
reflectable: ^4.0.9
dev_dependencies:
build_runner: ^2.3.3
build_test: ^2.1.0
lints: ^3.0.0
test: ^1.24.0

0
core/support/test/.gitkeep
View file

167
core/support/test/service_provider_test.dart
View file

@ -1,167 +0,0 @@
/*
* This file is part of the Protevus Platform.
*
* (C) Protevus <developers@protevus.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
import 'package:mockito/annotations.dart';
import 'package:mockito/mockito.dart';
import 'package:test/test.dart';
import 'package:angel3_container/angel3_container.dart';
import 'package:angel3_framework/angel3_framework.dart';
import 'package:logging/logging.dart';
import 'package:angel3_support/service_provider.dart';
@GenerateMocks([Logger, Container, Angel])
import 'service_provider_test.mocks.dart';
class TestServiceProvider extends ServiceProvider {
@override
void register(Container container, Angel app) {
try {
ServiceProvider.logger.info('Registering TestServiceProvider');
container.registerSingleton<String>('test', as: null);
} catch (e, stackTrace) {
ServiceProvider.logger
.severe('Error registering TestServiceProvider', e, stackTrace);
rethrow;
}
}
@override
Future<void> boot(Angel app) async {
try {
ServiceProvider.logger.info('Booting TestServiceProvider');
await app.configure((_) => null);
} catch (e, stackTrace) {
ServiceProvider.logger
.severe('Error booting TestServiceProvider', e, stackTrace);
rethrow;
}
}
}
void main() {
late MockContainer mockContainer;
late MockAngel mockAngel;
late MockLogger mockLogger;
late TestServiceProvider serviceProvider;
setUp(() {
mockContainer = MockContainer();
mockAngel = MockAngel();
mockLogger = MockLogger();
ServiceProvider.setLogger(mockLogger); // Set the mock logger
serviceProvider = TestServiceProvider();
// Setup default stubs
when(mockContainer.registerSingleton<dynamic>(any, as: anyNamed('as')))
.thenReturn('default');
when(mockContainer.registerFactory<dynamic>(any, as: anyNamed('as')))
.thenReturn((Container c) => 'default');
when(mockAngel.configure(any)).thenAnswer((_) async {});
// Allow any severe logging
when(mockLogger.info(any)).thenReturn(null);
when(mockLogger.severe(any, any, any)).thenReturn(null);
});
test('register method logs info and calls actual registration logic', () {
serviceProvider.register(mockContainer, mockAngel);
verify(mockLogger.info('Registering TestServiceProvider')).called(1);
});
test('boot method logs info and calls actual booting logic', () async {
await serviceProvider.boot(mockAngel);
verify(mockLogger.info('Booting TestServiceProvider')).called(1);
});
test('isDeferred returns false by default', () {
expect(serviceProvider.isDeferred, isFalse);
});
test('provides returns an empty list by default', () {
expect(serviceProvider.provides, isEmpty);
});
test('when returns true by default', () {
expect(serviceProvider.when(mockAngel), isTrue);
});
test(
'isReadyToRegister returns true when not deferred and when condition is met',
() {
expect(serviceProvider.isReadyToRegister(mockAngel), isTrue);
});
test('singleton method registers a singleton in the container', () {
final instance = 'test instance';
serviceProvider.singleton<String>(mockContainer, instance);
verify(mockContainer.registerSingleton<String>(instance, as: null))
.called(1);
});
test('bind method registers a factory in the container', () {
String factory(Container c) => 'test instance';
serviceProvider.bind<String>(mockContainer, factory);
verify(mockContainer.registerFactory<String>(any, as: null)).called(1);
});
test('error in register method is thrown', () {
when(mockContainer.registerSingleton<String>('test', as: null))
.thenThrow(Exception('Test error'));
expect(
() => serviceProvider.register(mockContainer, mockAngel),
throwsException,
);
});
test('error in boot method is thrown', () async {
when(mockAngel.configure(any)).thenThrow(Exception('Test error'));
await expectLater(
() => serviceProvider.boot(mockAngel),
throwsException,
);
});
test('error in register method is logged', () {
when(mockContainer.registerSingleton<String>('test', as: null))
.thenThrow(Exception('Test error'));
expect(
() => serviceProvider.register(mockContainer, mockAngel),
throwsException,
);
final captured =
verify(mockLogger.severe(captureAny, captureAny, captureAny)).captured;
expect(captured[0], 'Error registering TestServiceProvider');
expect(captured[1], isA<Exception>());
expect(captured[2], isA<StackTrace>());
});
test('error in boot method is logged', () async {
when(mockAngel.configure(any)).thenThrow(Exception('Test error'));
await expectLater(
() => serviceProvider.boot(mockAngel),
throwsException,
);
final captured =
verify(mockLogger.severe(captureAny, captureAny, captureAny)).captured;
expect(captured[0], 'Error booting TestServiceProvider');
expect(captured[1], isA<Exception>());
expect(captured[2], isA<StackTrace>());
});
tearDown(() {
// Reset the logger to its original state after each test
ServiceProvider.setLogger(Logger('ServiceProvider'));
});
}