Update: adding bus package not production grade zero test passing

This commit is contained in:
Patrick Stewart 2024-10-06 09:50:42 -07:00
parent 802de34a33
commit 37c2720969
11 changed files with 1024 additions and 0 deletions

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -0,0 +1,201 @@
// 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';
class Dispatcher implements QueueingDispatcher {
final Container container;
final EventBus _eventBus;
final Subject<Command> _commandSubject;
final MQClient _queue;
final Map<Type, Type> _handlers = {};
Dispatcher(this.container)
: _eventBus = container.make<EventBus>(),
_commandSubject = BehaviorSubject<Command>(),
_queue = container.make<MQClient>() {
_setupCommandProcessing();
}
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);
});
}
@override
Future<dynamic> dispatch(Command command) {
if (command is ShouldQueue) {
return dispatchToQueue(command);
} else {
return dispatchNow(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;
}
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}');
}
}
@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;
}
@override
Future<dynamic> dispatchSync(Command command, [Handler? handler]) {
return dispatchNow(command, handler);
}
@override
Future<Batch?> findBatch(String batchId) async {
// Implement batch finding logic
throw UnimplementedError();
}
@override
PendingBatch batch(List<Command> commands) {
return PendingBatch(this, commands);
}
@override
PendingChain chain(List<Command> commands) {
return PendingChain(this, commands);
}
@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;
}
@override
Dispatcher map(Map<Type, Type> handlers) {
_handlers.addAll(handlers);
return this;
}
@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];
}

View file

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

View file

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

View file

@ -10,8 +10,15 @@ environment:
# 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

View file

@ -0,0 +1,68 @@
import 'package:angel3_bus/angel3_bus.dart';
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 'package:test/test.dart';
import 'package:mockito/annotations.dart';
import 'package:mockito/mockito.dart';
import 'dispatcher_test.mocks.dart';
@GenerateMocks([Container, EventBus, MQClient])
// class MockContainer extends Mock implements Container {}
class MockEventBus extends Mock implements EventBus {}
class MockMQClient extends Mock implements MQClient {}
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();
}
}
void main() {
late MockContainer container;
//late MockEventBus eventBus;
//late MockMQClient mqClient;
late Dispatcher dispatcher;
setUpAll(() {
provideDummy<EventBus>(MockEventBus());
provideDummy<MQClient>(MockMQClient());
});
setUp(() {
container = MockContainer();
//eventBus = MockEventBus();
//mqClient = MockMQClient();
dispatcher = Dispatcher(container);
});
group('Dispatcher', () {
test('dispatchNow should handle command and return result', () async {
final command = TestCommand('test data');
final handler = TestHandler();
when(container.make(TestHandler)).thenReturn(handler);
dispatcher.map({TestCommand: TestHandler});
final result = await dispatcher.dispatchNow(command);
expect(result, equals('Handled: test data'));
});
;
});
}

View file

@ -0,0 +1,627 @@
// Mocks generated by Mockito 5.4.4 from annotations
// in angel3_bus/test/dispatcher_test.dart.
// Do not manually edit this file.
// ignore_for_file: no_leading_underscores_for_library_prefixes
import 'dart:async' as _i4;
import 'package:angel3_container/src/container.dart' as _i3;
import 'package:angel3_container/src/reflector.dart' as _i2;
import 'package:angel3_event_bus/res/app_event.dart' as _i8;
import 'package:angel3_event_bus/res/event_bus.dart' as _i7;
import 'package:angel3_event_bus/res/history_entry.dart' as _i9;
import 'package:angel3_event_bus/res/subscription.dart' as _i5;
import 'package:angel3_mq/src/core/constants/enums.dart' as _i12;
import 'package:angel3_mq/src/message/message.dart' as _i11;
import 'package:angel3_mq/src/mq/mq.dart' as _i10;
import 'package:mockito/mockito.dart' as _i1;
import 'package:mockito/src/dummies.dart' as _i6;
// ignore_for_file: type=lint
// ignore_for_file: avoid_redundant_argument_values
// ignore_for_file: avoid_setters_without_getters
// ignore_for_file: comment_references
// ignore_for_file: deprecated_member_use
// ignore_for_file: deprecated_member_use_from_same_package
// ignore_for_file: implementation_imports
// ignore_for_file: invalid_use_of_visible_for_testing_member
// ignore_for_file: prefer_const_constructors
// ignore_for_file: unnecessary_parenthesis
// ignore_for_file: camel_case_types
// ignore_for_file: subtype_of_sealed_class
class _FakeReflector_0 extends _i1.SmartFake implements _i2.Reflector {
_FakeReflector_0(
Object parent,
Invocation parentInvocation,
) : super(
parent,
parentInvocation,
);
}
class _FakeContainer_1 extends _i1.SmartFake implements _i3.Container {
_FakeContainer_1(
Object parent,
Invocation parentInvocation,
) : super(
parent,
parentInvocation,
);
}
class _FakeFuture_2<T1> extends _i1.SmartFake implements _i4.Future<T1> {
_FakeFuture_2(
Object parent,
Invocation parentInvocation,
) : super(
parent,
parentInvocation,
);
}
class _FakeSubscription_3 extends _i1.SmartFake implements _i5.Subscription {
_FakeSubscription_3(
Object parent,
Invocation parentInvocation,
) : super(
parent,
parentInvocation,
);
}
/// A class which mocks [Container].
///
/// See the documentation for Mockito's code generation for more information.
class MockContainer extends _i1.Mock implements _i3.Container {
MockContainer() {
_i1.throwOnMissingStub(this);
}
@override
_i2.Reflector get reflector => (super.noSuchMethod(
Invocation.getter(#reflector),
returnValue: _FakeReflector_0(
this,
Invocation.getter(#reflector),
),
) as _i2.Reflector);
@override
bool get isRoot => (super.noSuchMethod(
Invocation.getter(#isRoot),
returnValue: false,
) as bool);
@override
_i3.Container createChild() => (super.noSuchMethod(
Invocation.method(
#createChild,
[],
),
returnValue: _FakeContainer_1(
this,
Invocation.method(
#createChild,
[],
),
),
) as _i3.Container);
@override
bool has<T>([Type? t]) => (super.noSuchMethod(
Invocation.method(
#has,
[t],
),
returnValue: false,
) as bool);
@override
bool hasNamed(String? name) => (super.noSuchMethod(
Invocation.method(
#hasNamed,
[name],
),
returnValue: false,
) as bool);
@override
_i4.Future<T> makeAsync<T>([Type? type]) => (super.noSuchMethod(
Invocation.method(
#makeAsync,
[type],
),
returnValue: _i6.ifNotNull(
_i6.dummyValueOrNull<T>(
this,
Invocation.method(
#makeAsync,
[type],
),
),
(T v) => _i4.Future<T>.value(v),
) ??
_FakeFuture_2<T>(
this,
Invocation.method(
#makeAsync,
[type],
),
),
) as _i4.Future<T>);
@override
T make<T>([Type? type]) => (super.noSuchMethod(
Invocation.method(
#make,
[type],
),
returnValue: _i6.dummyValue<T>(
this,
Invocation.method(
#make,
[type],
),
),
) as T);
@override
T Function(_i3.Container) registerLazySingleton<T>(
T Function(_i3.Container)? f, {
Type? as,
}) =>
(super.noSuchMethod(
Invocation.method(
#registerLazySingleton,
[f],
{#as: as},
),
returnValue: (_i3.Container __p0) => _i6.dummyValue<T>(
this,
Invocation.method(
#registerLazySingleton,
[f],
{#as: as},
),
),
) as T Function(_i3.Container));
@override
T Function(_i3.Container) registerFactory<T>(
T Function(_i3.Container)? f, {
Type? as,
}) =>
(super.noSuchMethod(
Invocation.method(
#registerFactory,
[f],
{#as: as},
),
returnValue: (_i3.Container __p0) => _i6.dummyValue<T>(
this,
Invocation.method(
#registerFactory,
[f],
{#as: as},
),
),
) as T Function(_i3.Container));
@override
T registerSingleton<T>(
T? object, {
Type? as,
}) =>
(super.noSuchMethod(
Invocation.method(
#registerSingleton,
[object],
{#as: as},
),
returnValue: _i6.dummyValue<T>(
this,
Invocation.method(
#registerSingleton,
[object],
{#as: as},
),
),
) as T);
@override
T findByName<T>(String? name) => (super.noSuchMethod(
Invocation.method(
#findByName,
[name],
),
returnValue: _i6.dummyValue<T>(
this,
Invocation.method(
#findByName,
[name],
),
),
) as T);
@override
T registerNamedSingleton<T>(
String? name,
T? object,
) =>
(super.noSuchMethod(
Invocation.method(
#registerNamedSingleton,
[
name,
object,
],
),
returnValue: _i6.dummyValue<T>(
this,
Invocation.method(
#registerNamedSingleton,
[
name,
object,
],
),
),
) as T);
@override
void registerScoped<T>(T Function(_i3.Container)? factory) =>
super.noSuchMethod(
Invocation.method(
#registerScoped,
[factory],
),
returnValueForMissingStub: null,
);
@override
void registerTransient<T>(T Function(_i3.Container)? factory) =>
super.noSuchMethod(
Invocation.method(
#registerTransient,
[factory],
),
returnValueForMissingStub: null,
);
@override
void registerConstant<T>(T? value) => super.noSuchMethod(
Invocation.method(
#registerConstant,
[value],
),
returnValueForMissingStub: null,
);
}
/// A class which mocks [EventBus].
///
/// See the documentation for Mockito's code generation for more information.
class MockEventBus extends _i1.Mock implements _i7.EventBus {
MockEventBus() {
_i1.throwOnMissingStub(this);
}
@override
int get maxHistoryLength => (super.noSuchMethod(
Invocation.getter(#maxHistoryLength),
returnValue: 0,
) as int);
@override
bool get allowLogging => (super.noSuchMethod(
Invocation.getter(#allowLogging),
returnValue: false,
) as bool);
@override
Map<Type, List<_i8.AppEvent Function(_i8.AppEvent)>> get map =>
(super.noSuchMethod(
Invocation.getter(#map),
returnValue: <Type, List<_i8.AppEvent Function(_i8.AppEvent)>>{},
) as Map<Type, List<_i8.AppEvent Function(_i8.AppEvent)>>);
@override
bool get isBusy => (super.noSuchMethod(
Invocation.getter(#isBusy),
returnValue: false,
) as bool);
@override
_i4.Stream<bool> get isBusy$ => (super.noSuchMethod(
Invocation.getter(#isBusy$),
returnValue: _i4.Stream<bool>.empty(),
) as _i4.Stream<bool>);
@override
_i4.Stream<_i8.AppEvent?> get last$ => (super.noSuchMethod(
Invocation.getter(#last$),
returnValue: _i4.Stream<_i8.AppEvent?>.empty(),
) as _i4.Stream<_i8.AppEvent?>);
@override
_i4.Stream<List<_i8.AppEvent>> get inProgress$ => (super.noSuchMethod(
Invocation.getter(#inProgress$),
returnValue: _i4.Stream<List<_i8.AppEvent>>.empty(),
) as _i4.Stream<List<_i8.AppEvent>>);
@override
List<_i9.EventBusHistoryEntry> get history => (super.noSuchMethod(
Invocation.getter(#history),
returnValue: <_i9.EventBusHistoryEntry>[],
) as List<_i9.EventBusHistoryEntry>);
@override
void fire(_i8.AppEvent? event) => super.noSuchMethod(
Invocation.method(
#fire,
[event],
),
returnValueForMissingStub: null,
);
@override
void watch(_i8.AppEvent? event) => super.noSuchMethod(
Invocation.method(
#watch,
[event],
),
returnValueForMissingStub: null,
);
@override
void complete(
_i8.AppEvent? event, {
_i8.AppEvent? nextEvent,
}) =>
super.noSuchMethod(
Invocation.method(
#complete,
[event],
{#nextEvent: nextEvent},
),
returnValueForMissingStub: null,
);
@override
bool isInProgress<T>() => (super.noSuchMethod(
Invocation.method(
#isInProgress,
[],
),
returnValue: false,
) as bool);
@override
_i4.Stream<T> on<T extends _i8.AppEvent>() => (super.noSuchMethod(
Invocation.method(
#on,
[],
),
returnValue: _i4.Stream<T>.empty(),
) as _i4.Stream<T>);
@override
_i5.Subscription respond<T>(_i5.Responder<T>? responder) =>
(super.noSuchMethod(
Invocation.method(
#respond,
[responder],
),
returnValue: _FakeSubscription_3(
this,
Invocation.method(
#respond,
[responder],
),
),
) as _i5.Subscription);
@override
_i4.Stream<bool> whileInProgress<T extends _i8.AppEvent>() =>
(super.noSuchMethod(
Invocation.method(
#whileInProgress,
[],
),
returnValue: _i4.Stream<bool>.empty(),
) as _i4.Stream<bool>);
@override
void clearHistory() => super.noSuchMethod(
Invocation.method(
#clearHistory,
[],
),
returnValueForMissingStub: null,
);
@override
void reset() => super.noSuchMethod(
Invocation.method(
#reset,
[],
),
returnValueForMissingStub: null,
);
@override
void dispose() => super.noSuchMethod(
Invocation.method(
#dispose,
[],
),
returnValueForMissingStub: null,
);
}
/// A class which mocks [MQClient].
///
/// See the documentation for Mockito's code generation for more information.
class MockMQClient extends _i1.Mock implements _i10.MQClient {
MockMQClient() {
_i1.throwOnMissingStub(this);
}
@override
String declareQueue(String? queueId) => (super.noSuchMethod(
Invocation.method(
#declareQueue,
[queueId],
),
returnValue: _i6.dummyValue<String>(
this,
Invocation.method(
#declareQueue,
[queueId],
),
),
) as String);
@override
void deleteQueue(String? queueId) => super.noSuchMethod(
Invocation.method(
#deleteQueue,
[queueId],
),
returnValueForMissingStub: null,
);
@override
_i4.Stream<_i11.Message> fetchQueue(String? queueId) => (super.noSuchMethod(
Invocation.method(
#fetchQueue,
[queueId],
),
returnValue: _i4.Stream<_i11.Message>.empty(),
) as _i4.Stream<_i11.Message>);
@override
List<String> listQueues() => (super.noSuchMethod(
Invocation.method(
#listQueues,
[],
),
returnValue: <String>[],
) as List<String>);
@override
void deleteMessage(
String? queueId,
_i11.Message? message,
) =>
super.noSuchMethod(
Invocation.method(
#deleteMessage,
[
queueId,
message,
],
),
returnValueForMissingStub: null,
);
@override
void sendMessage({
required _i11.Message? message,
String? exchangeName,
String? routingKey,
}) =>
super.noSuchMethod(
Invocation.method(
#sendMessage,
[],
{
#message: message,
#exchangeName: exchangeName,
#routingKey: routingKey,
},
),
returnValueForMissingStub: null,
);
@override
_i11.Message? getLatestMessage(String? queueId) =>
(super.noSuchMethod(Invocation.method(
#getLatestMessage,
[queueId],
)) as _i11.Message?);
@override
void bindQueue({
required String? queueId,
required String? exchangeName,
String? bindingKey,
}) =>
super.noSuchMethod(
Invocation.method(
#bindQueue,
[],
{
#queueId: queueId,
#exchangeName: exchangeName,
#bindingKey: bindingKey,
},
),
returnValueForMissingStub: null,
);
@override
void unbindQueue({
required String? queueId,
required String? exchangeName,
String? bindingKey,
}) =>
super.noSuchMethod(
Invocation.method(
#unbindQueue,
[],
{
#queueId: queueId,
#exchangeName: exchangeName,
#bindingKey: bindingKey,
},
),
returnValueForMissingStub: null,
);
@override
void declareExchange({
required String? exchangeName,
required _i12.ExchangeType? exchangeType,
}) =>
super.noSuchMethod(
Invocation.method(
#declareExchange,
[],
{
#exchangeName: exchangeName,
#exchangeType: exchangeType,
},
),
returnValueForMissingStub: null,
);
@override
void deleteExchange(String? exchangeName) => super.noSuchMethod(
Invocation.method(
#deleteExchange,
[exchangeName],
),
returnValueForMissingStub: null,
);
@override
void close() => super.noSuchMethod(
Invocation.method(
#close,
[],
),
returnValueForMissingStub: null,
);
}