platform/lib/src/http/hooked_service.dart
thosakwe 8187045754 +1
2017-04-24 22:44:22 -04:00

719 lines
23 KiB
Dart

library angel_framework.http;
import 'dart:async';
import 'package:merge_map/merge_map.dart';
import '../util.dart';
import 'angel_http_exception.dart';
import 'request_context.dart';
import 'response_context.dart';
import 'metadata.dart';
import 'service.dart';
/// Wraps another service in a service that broadcasts events on actions.
class HookedService extends Service {
final List<StreamController<HookedServiceEvent>> _ctrl = [];
/// Tbe service that is proxied by this hooked one.
final Service inner;
final HookedServiceEventDispatcher beforeIndexed =
new HookedServiceEventDispatcher();
final HookedServiceEventDispatcher beforeRead =
new HookedServiceEventDispatcher();
final HookedServiceEventDispatcher beforeCreated =
new HookedServiceEventDispatcher();
final HookedServiceEventDispatcher beforeModified =
new HookedServiceEventDispatcher();
final HookedServiceEventDispatcher beforeUpdated =
new HookedServiceEventDispatcher();
final HookedServiceEventDispatcher beforeRemoved =
new HookedServiceEventDispatcher();
final HookedServiceEventDispatcher afterIndexed =
new HookedServiceEventDispatcher();
final HookedServiceEventDispatcher afterRead =
new HookedServiceEventDispatcher();
final HookedServiceEventDispatcher afterCreated =
new HookedServiceEventDispatcher();
final HookedServiceEventDispatcher afterModified =
new HookedServiceEventDispatcher();
final HookedServiceEventDispatcher afterUpdated =
new HookedServiceEventDispatcher();
final HookedServiceEventDispatcher afterRemoved =
new HookedServiceEventDispatcher();
HookedService(Service this.inner) {
// Clone app instance
if (inner.app != null) this.app = inner.app;
}
RequestContext _getRequest(Map params) {
if (params == null) return null;
return params['__requestctx'];
}
ResponseContext _getResponse(Map params) {
if (params == null) return null;
return params['__responsectx'];
}
Map _stripReq(Map params) {
if (params == null)
return params;
else
return params.keys
.where((key) => key != '__requestctx' && key != '__responsectx')
.fold({}, (map, key) => map..[key] = params[key]);
}
/// Closes any open [StreamController]s on this instance. **Internal use only**.
@override
Future close() async {
_ctrl.forEach((c) => c.close());
beforeIndexed._close();
beforeRead._close();
beforeCreated._close();
beforeModified._close();
beforeUpdated._close();
beforeRemoved._close();
afterIndexed._close();
afterRead._close();
afterCreated._close();
afterModified._close();
afterUpdated._close();
afterRemoved._close();
if (inner is HookedService) inner.close();
}
/// Adds hooks to this instance.
void addHooks() {
Hooks hooks = getAnnotation(inner, Hooks);
final before = [];
final after = [];
if (hooks != null) {
before.addAll(hooks.before);
after.addAll(hooks.after);
}
void applyListeners(Function fn, HookedServiceEventDispatcher dispatcher,
[bool isAfter]) {
Hooks hooks = getAnnotation(fn, Hooks);
final listeners = []..addAll(isAfter == true ? after : before);
if (hooks != null)
listeners.addAll(isAfter == true ? hooks.after : hooks.before);
listeners.forEach(dispatcher.listen);
}
applyListeners(inner.index, beforeIndexed);
applyListeners(inner.read, beforeRead);
applyListeners(inner.created, beforeCreated);
applyListeners(inner.modify, beforeModified);
applyListeners(inner.updated, beforeUpdated);
applyListeners(inner.removed, beforeRemoved);
applyListeners(inner.index, afterIndexed, true);
applyListeners(inner.read, afterRead, true);
applyListeners(inner.created, afterCreated, true);
applyListeners(inner.modify, afterModified, true);
applyListeners(inner.updated, afterUpdated, true);
applyListeners(inner.removed, afterRemoved, true);
}
/// Adds routes to this instance.
@override
void addRoutes() {
// Set up our routes. We still need to copy middleware from inner service
Map restProvider = {'provider': Providers.REST};
// Add global middleware if declared on the instance itself
Middleware before = getAnnotation(inner, Middleware);
final handlers = [
(RequestContext req, ResponseContext res) async {
req.serviceParams
..['__requestctx'] = req
..['__responsectx'] = res;
return true;
}
];
if (before != null) handlers.addAll(before.handlers);
Middleware indexMiddleware = getAnnotation(inner.index, Middleware);
get('/', (req, res) async {
return await this.index(mergeMap([
{'query': req.query},
restProvider,
req.serviceParams
]));
},
middleware: []
..addAll(handlers)
..addAll((indexMiddleware == null) ? [] : indexMiddleware.handlers));
Middleware createMiddleware = getAnnotation(inner.create, Middleware);
post('/', (req, res) async {
var r = await this.create(
await req.lazyBody(),
mergeMap([
{'query': req.query},
restProvider,
req.serviceParams
]));
res.statusCode = 201;
return r;
},
middleware: []
..addAll(handlers)
..addAll(
(createMiddleware == null) ? [] : createMiddleware.handlers));
Middleware readMiddleware = getAnnotation(inner.read, Middleware);
get(
'/:id',
(req, res) async => await this.read(
toId(req.params['id']),
mergeMap([
{'query': req.query},
restProvider,
req.serviceParams
])),
middleware: []
..addAll(handlers)
..addAll((readMiddleware == null) ? [] : readMiddleware.handlers));
Middleware modifyMiddleware = getAnnotation(inner.modify, Middleware);
patch(
'/:id',
(req, res) async => await this.modify(
toId(req.params['id']),
await req.lazyBody(),
mergeMap([
{'query': req.query},
restProvider,
req.serviceParams
])),
middleware: []
..addAll(handlers)
..addAll(
(modifyMiddleware == null) ? [] : modifyMiddleware.handlers));
Middleware updateMiddleware = getAnnotation(inner.update, Middleware);
post(
'/:id',
(req, res) async => await this.update(
toId(req.params['id']),
await req.lazyBody(),
mergeMap([
{'query': req.query},
restProvider,
req.serviceParams
])),
middleware: []
..addAll(handlers)
..addAll(
(updateMiddleware == null) ? [] : updateMiddleware.handlers));
put(
'/:id',
(req, res) async => await this.update(
toId(req.params['id']),
await req.lazyBody(),
mergeMap([
{'query': req.query},
restProvider,
req.serviceParams
])),
middleware: []
..addAll(handlers)
..addAll(
(updateMiddleware == null) ? [] : updateMiddleware.handlers));
Middleware removeMiddleware = getAnnotation(inner.remove, Middleware);
delete(
'/',
(req, res) async => await this.remove(
null,
mergeMap([
{'query': req.query},
restProvider,
req.serviceParams
])),
middleware: []
..addAll(handlers)
..addAll(
(removeMiddleware == null) ? [] : removeMiddleware.handlers));
delete(
'/:id',
(req, res) async => await this.remove(
toId(req.params['id']),
mergeMap([
{'query': req.query},
restProvider,
req.serviceParams
])),
middleware: []
..addAll(handlers)
..addAll(
(removeMiddleware == null) ? [] : removeMiddleware.handlers));
// REST compliance
put('/', () => throw new AngelHttpException.notFound());
patch('/', () => throw new AngelHttpException.notFound());
addHooks();
}
/// Runs the [listener] before every service method specified.
void before(
Iterable<String> eventNames, HookedServiceEventListener listener) {
eventNames.map((name) {
switch (name) {
case HookedServiceEvent.INDEXED:
return beforeIndexed;
case HookedServiceEvent.READ:
return beforeRead;
case HookedServiceEvent.CREATED:
return beforeCreated;
case HookedServiceEvent.MODIFIED:
return beforeModified;
case HookedServiceEvent.UPDATED:
return beforeUpdated;
case HookedServiceEvent.REMOVED:
return beforeRemoved;
default:
throw new ArgumentError('Invalid service method: ${name}');
}
}).forEach((HookedServiceEventDispatcher dispatcher) =>
dispatcher.listen(listener));
}
/// Runs the [listener] after every service method specified.
void after(Iterable<String> eventNames, HookedServiceEventListener listener) {
eventNames.map((name) {
switch (name) {
case HookedServiceEvent.INDEXED:
return afterIndexed;
case HookedServiceEvent.READ:
return afterRead;
case HookedServiceEvent.CREATED:
return afterCreated;
case HookedServiceEvent.MODIFIED:
return afterModified;
case HookedServiceEvent.UPDATED:
return afterUpdated;
case HookedServiceEvent.REMOVED:
return afterRemoved;
default:
throw new ArgumentError('Invalid service method: ${name}');
}
}).forEach((HookedServiceEventDispatcher dispatcher) =>
dispatcher.listen(listener));
}
/// Runs the [listener] before every service method.
void beforeAll(HookedServiceEventListener listener) {
beforeIndexed.listen(listener);
beforeRead.listen(listener);
beforeCreated.listen(listener);
beforeModified.listen(listener);
beforeUpdated.listen(listener);
beforeRemoved.listen(listener);
}
/// Runs the [listener] after every service method.
void afterAll(HookedServiceEventListener listener) {
afterIndexed.listen(listener);
afterRead.listen(listener);
afterCreated.listen(listener);
afterModified.listen(listener);
afterUpdated.listen(listener);
afterRemoved.listen(listener);
}
/// Returns a [Stream] of all events fired before every service method.
///
/// *NOTE*: Only use this if you do not plan to modify events. There is no guarantee
/// that events coming out of this [Stream] will see changes you make within the [Stream]
/// callback.
Stream<HookedServiceEvent> beforeAllStream() {
var ctrl = new StreamController<HookedServiceEvent>();
_ctrl.add(ctrl);
before(HookedServiceEvent.ALL, ctrl.add);
return ctrl.stream;
}
/// Returns a [Stream] of all events fired after every service method.
///
/// *NOTE*: Only use this if you do not plan to modify events. There is no guarantee
/// that events coming out of this [Stream] will see changes you make within the [Stream]
/// callback.
Stream<HookedServiceEvent> afterAllStream() {
var ctrl = new StreamController<HookedServiceEvent>();
_ctrl.add(ctrl);
before(HookedServiceEvent.ALL, ctrl.add);
return ctrl.stream;
}
/// Returns a [Stream] of all events fired before every service method specified.
///
/// *NOTE*: Only use this if you do not plan to modify events. There is no guarantee
/// that events coming out of this [Stream] will see changes you make within the [Stream]
/// callback.
Stream<HookedServiceEvent> beforeStream(Iterable<String> eventNames) {
var ctrl = new StreamController<HookedServiceEvent>();
_ctrl.add(ctrl);
before(eventNames, ctrl.add);
return ctrl.stream;
}
/// Returns a [Stream] of all events fired AFTER every service method specified.
///
/// *NOTE*: Only use this if you do not plan to modify events. There is no guarantee
/// that events coming out of this [Stream] will see changes you make within the [Stream]
/// callback.
Stream<HookedServiceEvent> afterStream(Iterable<String> eventNames) {
var ctrl = new StreamController<HookedServiceEvent>();
_ctrl.add(ctrl);
after(eventNames, ctrl.add);
return ctrl.stream;
}
/// Runs the [listener] before [create], [modify] and [update].
void beforeModify(HookedServiceEventListener listener) {
beforeCreated.listen(listener);
beforeModified.listen(listener);
beforeUpdated.listen(listener);
}
@override
Future index([Map _params]) async {
var params = _stripReq(_params);
HookedServiceEvent before = await beforeIndexed._emit(
new HookedServiceEvent(false, _getRequest(_params),
_getResponse(_params), inner, HookedServiceEvent.INDEXED,
params: params));
if (before._canceled) {
HookedServiceEvent after = await beforeIndexed._emit(
new HookedServiceEvent(true, _getRequest(_params),
_getResponse(_params), inner, HookedServiceEvent.INDEXED,
params: params, result: before.result));
return after.result;
}
var result = await inner.index(params);
HookedServiceEvent after = await afterIndexed._emit(new HookedServiceEvent(
true,
_getRequest(_params),
_getResponse(_params),
inner,
HookedServiceEvent.INDEXED,
params: params,
result: result));
return after.result;
}
@override
Future read(id, [Map _params]) async {
var params = _stripReq(_params);
HookedServiceEvent before = await beforeRead._emit(new HookedServiceEvent(
false,
_getRequest(_params),
_getResponse(_params),
inner,
HookedServiceEvent.READ,
id: id,
params: params));
if (before._canceled) {
HookedServiceEvent after = await afterRead._emit(new HookedServiceEvent(
true,
_getRequest(_params),
_getResponse(_params),
inner,
HookedServiceEvent.READ,
id: id,
params: params,
result: before.result));
return after.result;
}
var result = await inner.read(id, params);
HookedServiceEvent after = await afterRead._emit(new HookedServiceEvent(
true,
_getRequest(_params),
_getResponse(_params),
inner,
HookedServiceEvent.READ,
id: id,
params: params,
result: result));
return after.result;
}
@override
Future create(data, [Map _params]) async {
var params = _stripReq(_params);
HookedServiceEvent before = await beforeCreated._emit(
new HookedServiceEvent(false, _getRequest(_params),
_getResponse(_params), inner, HookedServiceEvent.CREATED,
data: data, params: params));
if (before._canceled) {
HookedServiceEvent after = await afterCreated._emit(
new HookedServiceEvent(true, _getRequest(_params),
_getResponse(_params), inner, HookedServiceEvent.CREATED,
data: data, params: params, result: before.result));
return after.result;
}
var result = await inner.create(data, params);
HookedServiceEvent after = await afterCreated._emit(new HookedServiceEvent(
true,
_getRequest(_params),
_getResponse(_params),
inner,
HookedServiceEvent.CREATED,
data: data,
params: params,
result: result));
return after.result;
}
@override
Future modify(id, data, [Map _params]) async {
var params = _stripReq(_params);
HookedServiceEvent before = await beforeModified._emit(
new HookedServiceEvent(false, _getRequest(_params),
_getResponse(_params), inner, HookedServiceEvent.MODIFIED,
id: id, data: data, params: params));
if (before._canceled) {
HookedServiceEvent after = await afterModified._emit(
new HookedServiceEvent(true, _getRequest(_params),
_getResponse(_params), inner, HookedServiceEvent.MODIFIED,
id: id, data: data, params: params, result: before.result));
return after.result;
}
var result = await inner.modify(id, data, params);
HookedServiceEvent after = await afterModified._emit(new HookedServiceEvent(
true,
_getRequest(_params),
_getResponse(_params),
inner,
HookedServiceEvent.MODIFIED,
id: id,
data: data,
params: params,
result: result));
return after.result;
}
@override
Future update(id, data, [Map _params]) async {
var params = _stripReq(_params);
HookedServiceEvent before = await beforeUpdated._emit(
new HookedServiceEvent(false, _getRequest(_params),
_getResponse(_params), inner, HookedServiceEvent.UPDATED,
id: id, data: data, params: params));
if (before._canceled) {
HookedServiceEvent after = await afterUpdated._emit(
new HookedServiceEvent(true, _getRequest(_params),
_getResponse(_params), inner, HookedServiceEvent.UPDATED,
id: id, data: data, params: params, result: before.result));
return after.result;
}
var result = await inner.update(id, data, params);
HookedServiceEvent after = await afterUpdated._emit(new HookedServiceEvent(
true,
_getRequest(_params),
_getResponse(_params),
inner,
HookedServiceEvent.UPDATED,
id: id,
data: data,
params: params,
result: result));
return after.result;
}
@override
Future remove(id, [Map _params]) async {
var params = _stripReq(_params);
HookedServiceEvent before = await beforeRemoved._emit(
new HookedServiceEvent(false, _getRequest(_params),
_getResponse(_params), inner, HookedServiceEvent.REMOVED,
id: id, params: params));
if (before._canceled) {
HookedServiceEvent after = await afterRemoved._emit(
new HookedServiceEvent(true, _getRequest(_params),
_getResponse(_params), inner, HookedServiceEvent.REMOVED,
id: id, params: params, result: before.result));
return after.result;
}
var result = await inner.remove(id, params);
HookedServiceEvent after = await afterRemoved._emit(new HookedServiceEvent(
true,
_getRequest(_params),
_getResponse(_params),
inner,
HookedServiceEvent.REMOVED,
id: id,
params: params,
result: result));
return after.result;
}
/// Fires an `after` event. This will not be propagated to clients,
/// but will be broadcasted to WebSockets, etc.
Future<HookedServiceEvent> fire(String eventName, result,
[HookedServiceEventListener callback]) async {
HookedServiceEventDispatcher dispatcher;
switch (eventName) {
case HookedServiceEvent.INDEXED:
dispatcher = afterIndexed;
break;
case HookedServiceEvent.READ:
dispatcher = afterRead;
break;
case HookedServiceEvent.CREATED:
dispatcher = afterCreated;
break;
case HookedServiceEvent.MODIFIED:
dispatcher = afterModified;
break;
case HookedServiceEvent.UPDATED:
dispatcher = afterUpdated;
break;
case HookedServiceEvent.REMOVED:
dispatcher = afterRemoved;
break;
default:
throw new ArgumentError("Invalid service event name: '$eventName'");
}
var ev = new HookedServiceEvent(true, null, null, this, eventName);
return await fireEvent(dispatcher, ev, callback);
}
/// Sends an arbitrary event down the hook chain.
Future<HookedServiceEvent> fireEvent(
HookedServiceEventDispatcher dispatcher, HookedServiceEvent event,
[HookedServiceEventListener callback]) async {
if (callback != null && event?._canceled != true) await callback(event);
return await dispatcher._emit(event);
}
}
/// Fired when a hooked service is invoked.
class HookedServiceEvent {
static const String INDEXED = "indexed";
static const String READ = "read";
static const String CREATED = "created";
static const String MODIFIED = "modified";
static const String UPDATED = "updated";
static const String REMOVED = "removed";
static const List<String> ALL = const [
INDEXED,
READ,
CREATED,
MODIFIED,
UPDATED,
REMOVED
];
/// Use this to end processing of an event.
void cancel([result]) {
_canceled = true;
this.result = result ?? this.result;
}
/// Resolves a service from the application.
///
/// Shorthand for `e.service.app.service(...)`.
Service getService(Pattern path) => service.app.service(path);
bool _canceled = false;
String _eventName;
var _id;
bool _isAfter;
var data;
Map _params;
RequestContext _request;
ResponseContext _response;
var result;
String get eventName => _eventName;
get id => _id;
bool get isAfter => _isAfter == true;
bool get isBefore => !isAfter;
Map get params => _params;
RequestContext get request => _request;
ResponseContext get response => _response;
/// The inner service whose method was hooked.
Service service;
HookedServiceEvent(this._isAfter, this._request, this._response,
Service this.service, String this._eventName,
{id, this.data, Map params, this.result}) {
_id = id;
_params = params ?? {};
}
}
/// Triggered on a hooked service event.
typedef HookedServiceEventListener(HookedServiceEvent event);
/// Can be listened to, but events may be canceled.
class HookedServiceEventDispatcher {
final List<StreamController<HookedServiceEvent>> _ctrl = [];
final List<HookedServiceEventListener> listeners = [];
void _close() {
_ctrl.forEach((c) => c.close());
}
/// Fires an event, and returns it once it is either canceled, or all listeners have run.
Future<HookedServiceEvent> _emit(HookedServiceEvent event) async {
if (event?._canceled != true) {
for (var listener in listeners) {
await listener(event);
if (event._canceled) return event;
}
}
return event;
}
/// Returns a [Stream] containing all events fired by this dispatcher.
///
/// *NOTE*: Callbacks on the returned [Stream] cannot be guaranteed to run before other [listeners].
/// Use this only if you need a read-only stream of events.
Stream<HookedServiceEvent> asStream() {
var ctrl = new StreamController<HookedServiceEvent>();
_ctrl.add(ctrl);
listen(ctrl.add);
return ctrl.stream;
}
/// Registers the listener to be called whenever an event is triggered.
void listen(HookedServiceEventListener listener) {
listeners.add(listener);
}
}