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

1.0.0

Browse source
This commit is contained in:
Tobe O 2018-11-08 10:09:32 -05:00
parent d6d56f1f68
commit bc079d3735
11 changed files with 91 additions and 151 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -62,3 +62,4 @@ pubspec.lock
# Directory created by dartdoc
# If you don't generate documentation locally you can remove this line.
doc/api/
.dart_tool

31
README.md
View file

@ -1,2 +1,33 @@
# eventsource
Server-sent Events (SSE) plugin for Angel.
## Installation
In your `pubspec.yaml`:
```yaml
dependencies:
angel_eventsource: ^1.0.0
```
## Usage
SSE and WebSockets are somewhat similar in that they allow pushing of events from server
to client. SSE is not bi-directional, but the same abstractions used for WebSockets can be
applied to SSE easily.
For this reason, the `AngelEventSourcePublisher` class is a simple adapter that
hands control of SSE requests to an existing `AngelWebSocket` driver.
So, using this is pretty straightforward. You can dispatch events
via WebSocket as per usual, and have them propagated to SSE clients
as well.
```dart
var app = new Angel();
var ws = new AngelWebSocket(app);
var events = new AngelEventSourcePublisher(ws);
await app.configure(ws.configureServer);
app.all('/ws', ws.handleRequest);
app.get('/events', events.handleRequest);
```

3
analysis_options.yaml
View file

@ -1,2 +1,3 @@
analyzer:
strong-mode: true
strong-mode:
implicit-casts: false

21
example/main.dart
View file

@ -1,25 +1,26 @@
import 'dart:convert';
import 'dart:io';
import 'package:angel_eventsource/server.dart';
import 'package:angel_framework/angel_framework.dart';
import 'package:angel_framework/http.dart';
import 'package:angel_websocket/server.dart';
import 'package:eventsource/eventsource.dart';
import 'package:eventsource/publisher.dart';
import 'package:logging/logging.dart';
import 'pretty_logging.dart';
main() async {
var app = new Angel();
var ws = new AngelWebSocket(app);
var events = new AngelEventSourcePublisher(ws);
await app.configure(ws.configureServer);
app.use('/api/todos', new MapService());
var publisher = new AngelEventSourcePublisher(new EventSourcePublisher());
await app.configure(publisher.configureServer);
app.get('/sse', publisher.handleRequest);
app.all('/ws', ws.handleRequest);
app.get('/events', events.handleRequest);
app.logger = new Logger('angel_eventsource')..onRecord.listen(prettyLog);
var server = await app.startServer('127.0.0.1', 3000);
var http = new AngelHttp(app);
var server = await http.startServer('127.0.0.1', 3000);
var url = Uri.parse('http://${server.address.address}:${server.port}');
print('Listening at $url');
@ -41,11 +42,9 @@ main() async {
rs.transform(UTF8.decoder).transform(const LineSplitter()).listen(print);
*/
var eventSource = await EventSource.connect(url);
await for (var event in eventSource) {
print(event.data);
}
}

9
example/pretty_logging.dart
View file

@ -7,10 +7,8 @@ import 'package:logging/logging.dart';
chooseLogColor(pen.reset(), record.level);
pen(record.toString());
if (record.error != null)
pen(record.error.toString());
if (record.stackTrace != null)
pen(record.stackTrace.toString());
if (record.error != null) pen(record.error.toString());
if (record.stackTrace != null) pen(record.stackTrace.toString());
pen();
}
@ -27,6 +25,5 @@ void chooseLogColor(TextPen pen, Level level) {
pen.magenta();
else if (level == Level.FINER)
pen.blue();
else if (level == Level.FINEST)
pen.darkBlue();
else if (level == Level.FINEST) pen.darkBlue();
}

112
lib/server.dart
View file

@ -1,119 +1,51 @@
import 'dart:async';
import 'package:angel_framework/angel_framework.dart';
import 'package:angel_framework/hooks.dart' as hooks;
import 'package:angel_websocket/server.dart';
import 'package:eventsource/eventsource.dart';
import 'package:eventsource/src/encoder.dart';
import 'package:eventsource/publisher.dart';
import 'package:json_god/json_god.dart' as god;
import 'package:stream_channel/stream_channel.dart';
class AngelEventSourcePublisher {
final EventSourcePublisher eventSourcePublisher;
final AngelWebSocket webSocketDriver;
/// Used to notify other nodes of an event's firing. Good for scaled applications.
final WebSocketSynchronizer synchronizer;
final String channel;
/// Serializes a [WebSocketEvent] to JSON.
///
/// Defaults to [god.serialize].
final String Function(WebSocketEvent) serializer;
int _count = 0;
AngelEventSourcePublisher(this.eventSourcePublisher,
{this.synchronizer, this.serializer});
AngelEventSourcePublisher(this.webSocketDriver, {this.channel: ''});
Future configureServer(Angel app) async {
await app.configure(hooks.hookAllServices((service) {
if (service is HookedService) {
var path =
app.services.keys.firstWhere((p) => app.services[p] == service);
service.after([
HookedServiceEvent.created,
HookedServiceEvent.modified,
HookedServiceEvent.updated,
HookedServiceEvent.removed,
], (e) {
var event = new WebSocketEvent(
eventName: '${path.toString()}::${e.eventName}',
data: e.result,
);
_filter(RequestContext req, ResponseContext res) {
if (e.service.configuration.containsKey('sse:filter'))
return e.service.configuration['sse:filter'](e, req, res);
else if (e.params != null && e.params.containsKey('sse:filter'))
return e.params['sse:filter'](e, req, res);
else
return true;
}
var canSend = _filter(e.request, e.response);
if (canSend) {
batchEvent(event, e.request.properties['channel'] ?? '');
}
});
}
}));
if (synchronizer != null) {
var sub = synchronizer.stream.listen((e) => batchEvent(e, ''));
app.shutdownHooks.add((_) async {
sub.cancel();
});
}
}
Future batchEvent(WebSocketEvent event, String channel) async {
eventSourcePublisher.add(
new Event(
id: (_count++).toString(),
data: (serializer ?? god.serialize)(event),
),
channels: [channel],
);
}
Future<bool> handleRequest(RequestContext req, ResponseContext res) async {
Future handleRequest(RequestContext req, ResponseContext res) async {
if (!req.accepts('text/event-stream', strict: false))
throw new AngelHttpException.badRequest();
res
..headers.addAll({
res.headers.addAll({
'cache-control': 'no-cache, no-store, must-revalidate',
'content-type': 'text/event-stream',
'connection': 'keep-alive',
})
..willCloseItself = true
..end();
});
var acceptsGzip =
(req.headers['accept-encoding']?.contains('gzip') == true);
if (acceptsGzip) res.io.headers.set('content-encoding', 'gzip');
res.headers.forEach(res.io.headers.set);
var sock = res.io ?? await res.io.detachSocket();
sock.flush();
if (acceptsGzip) res.headers['content-encoding'] = 'gzip';
var eventSink = new EventSourceEncoder(compressed: acceptsGzip)
.startChunkedConversion(sock);
.startChunkedConversion(res);
eventSourcePublisher.newSubscription(
onEvent: (e) {
try {
eventSink.add(e);
sock.flush();
} catch (_) {
// Ignore disconnect
}
},
onClose: eventSink.close,
channel: req.properties['channel'] ?? '',
lastEventId: req.headers.value('last-event-id'),
);
// Listen for events.
var ctrl = new StreamChannelController();
return false;
// Incoming events are strings, and should be sent via the eventSink.
ctrl.local.stream.cast<String>().listen((data) {
eventSink.add(new Event(
id: (_count++).toString(),
data: data,
));
});
// Create a new WebSocketContext, and hand it off to the driver.
var socket = new WebSocketContext(ctrl.foreign, req, res);
return await webSocketDriver.handleClient(socket);
}
}

26
pubspec.yaml
View file

@ -1,14 +1,16 @@
name: "angel_eventsource"
name: angel_eventsource
version: 1.0.0
description: Server-sent Events (SSE) plugin for Angel.
homepage: https://github.com/angel-dart/eventsource
author: Tobe O <thosakwe@gmail.com>
environment:
sdk: ">=2.0.0-dev <3.0.0"
dependencies:
angel_client: "^1.0.0"
angel_framework: "^1.0.0-dev"
angel_model: "^1.0.0"
angel_serialize: "^1.0.0-alpha"
angel_websocket: "^1.0.0"
eventsource: "^0.1.0"
json_god: ^2.0.0-beta
angel_framework: ^2.0.0-alpha
angel_websocket: ^2.0.0-alpha
eventsource: ^0.2.0
stream_channel: ^1.0.0
dev_dependencies:
angel_serialize_generator: "^1.0.0-alpha"
build_runner: "^0.5.0"
console: "^2.2.4"
test: "^0.12.0"
console: ^3.0.0
logging:
test: ^1.0.0

4
tool/build.dart
View file

@ -1,4 +0,0 @@
import 'package:build_runner/build_runner.dart';
import 'build_actions.dart';
main() => build(buildActions, deleteFilesByDefault: true);

15
tool/build_actions.dart
View file

@ -1,15 +0,0 @@
import 'package:angel_serialize_generator/angel_serialize_generator.dart';
import 'package:build_runner/build_runner.dart';
import 'package:source_gen/source_gen.dart';
final List<BuildAction> buildActions = [
new BuildAction(
new PartBuilder([
const JsonModelGenerator(autoIdAndDateFields: false),
]),
'angel_eventsource',
inputs: const [
'lib/angel_eventsource.dart',
],
),
];

4
tool/watch.dart
View file

@ -1,4 +0,0 @@
import 'package:build_runner/build_runner.dart';
import 'build_actions.dart';
main() => watch(buildActions, deleteFilesByDefault: true);