This commit is contained in:
thosakwe 2016-12-23 05:47:21 -05:00
parent 57e1516b0e
commit 7957a4beee
6 changed files with 198 additions and 16 deletions

View file

@ -0,0 +1,133 @@
import 'dart:async';
import 'package:angel_client/angel_client.dart';
import 'package:http/src/base_client.dart' as http;
import 'package:web_socket_channel/web_socket_channel.dart';
import 'package:web_socket_channel/status.dart';
import 'angel_websocket.dart';
export 'package:angel_client/angel_client.dart';
import 'package:angel_client/base_angel_client.dart';
final RegExp _straySlashes = new RegExp(r"(^/)|(/+$)");
abstract class BaseWebSocketClient extends BaseAngelClient {
WebSocketChannel _socket;
/// The [WebSocketChannel] underneath this instance.
WebSocketChannel get socket => _socket;
BaseWebSocketClient(http.BaseClient client, String basePath)
: super(client, basePath);
Future<WebSocketChannel> connect();
@override
BaseWebSocketService service<T>(String path,
{Type type, AngelDeserializer deserializer}) {
String uri = path.toString().replaceAll(_straySlashes, '');
return new BaseWebSocketService(socket, this, uri,
deserializer: deserializer)..listen();
}
}
class BaseWebSocketService extends Service {
@override
final Angel app;
final AngelDeserializer deserializer;
final WebSocketChannel socket;
final String uri;
final StreamController<WebSocketEvent> _onMessage =
new StreamController<WebSocketEvent>();
final StreamController<WebSocketEvent> _onError =
new StreamController<WebSocketEvent>();
final StreamController<WebSocketEvent> _onIndexed =
new StreamController<WebSocketEvent>();
final StreamController<WebSocketEvent> _onRead =
new StreamController<WebSocketEvent>();
final StreamController<WebSocketEvent> _onCreated =
new StreamController<WebSocketEvent>();
final StreamController<WebSocketEvent> _onModified =
new StreamController<WebSocketEvent>();
final StreamController<WebSocketEvent> _onUpdated =
new StreamController<WebSocketEvent>();
final StreamController<WebSocketEvent> _onRemoved =
new StreamController<WebSocketEvent>();
final WebSocketExtraneousEventHandler _on =
new WebSocketExtraneousEventHandler();
/// Use this to handle events that are not standard.
WebSocketExtraneousEventHandler get on => _on;
/// Fired on all events.
Stream<WebSocketEvent> get onMessage => _onMessage.stream;
/// Fired on errors.
Stream<WebSocketEvent> get onError => _onError.stream;
/// Fired on `index` events.
Stream<WebSocketEvent> get onIndexed => _onIndexed.stream;
/// Fired on `read` events.
Stream<WebSocketEvent> get onRead => _onRead.stream;
/// Fired on `created` events.
Stream<WebSocketEvent> get onCreated => _onCreated.stream;
/// Fired on `modified` events.
Stream<WebSocketEvent> get onModified => _onModified.stream;
/// Fired on `updated` events.
Stream<WebSocketEvent> get onUpdated => _onUpdated.stream;
/// Fired on `removed` events.
Stream<WebSocketEvent> get onRemoved => _onRemoved.stream;
BaseWebSocketService(this.socket, this.app, this.uri, {this.deserializer});
void listen() {
socket.stream.listen((message) {
print('Message: ${message.runtimeType}');
});
}
@override
Future<List> index([Map params]) {
// TODO: implement index
}
@override
Future read(id, [Map params]) {
// TODO: implement read
}
@override
Future create(data, [Map params]) {
// TODO: implement create
}
@override
Future modify(id, data, [Map params]) {
// TODO: implement modify
}
@override
Future update(id, data, [Map params]) {
// TODO: implement update
}
@override
Future remove(id, [Map params]) {
// TODO: implement remove
}
}
class WebSocketExtraneousEventHandler {
Map<String, StreamController<WebSocketEvent>> _events = {};
operator [](String index) {
if (_events[index] == null)
_events[index] = new StreamController<WebSocketEvent>();
return _events[index].stream;
}
}

38
lib/io.dart Normal file
View file

@ -0,0 +1,38 @@
/// Command-line WebSocket client library for the Angel framework.
library angel_client.cli;
import 'dart:async';
import 'package:angel_client/angel_client.dart';
import 'package:http/http.dart' as http;
import 'package:json_god/json_god.dart' as god;
import 'package:web_socket_channel/web_socket_channel.dart';
import 'package:web_socket_channel/io.dart';
import 'base_websocket_client.dart';
export 'package:angel_client/angel_client.dart';
export 'angel_websocket.dart';
final RegExp _straySlashes = new RegExp(r"(^/)|(/+$)");
/// Queries an Angel server via WebSockets.
class WebSockets extends BaseWebSocketClient {
WebSockets(String path) : super(new http.Client(), path);
@override
Future<WebSocketChannel> connect() async {
return new IOWebSocketChannel.connect(basePath);
}
@override
WebSocketsService service<T>(String path,
{Type type, AngelDeserializer deserializer}) {
String uri = path.replaceAll(_straySlashes, "");
return new WebSocketsService(socket, this, uri, T != dynamic ? T : type);
}
}
class WebSocketsService extends BaseWebSocketService {
final Type type;
WebSocketsService(WebSocketChannel socket, Angel app, String uri, this.type)
: super(socket, app, uri);
}

View file

@ -17,13 +17,23 @@ class AngelWebSocket extends AngelPlugin {
Angel _app;
List<WebSocket> _clients = [];
StreamController<WebSocketContext> _onConnection =
new StreamController<WebSocketContext>.broadcast();
new StreamController<WebSocketContext>();
StreamController<WebSocketContext> _onDisconnect =
new StreamController<WebSocketContext>.broadcast();
List<WebSocket> get clients => new List.from(_clients, growable: false);
List<String> servicesAlreadyWired = [];
String endpoint;
new StreamController<WebSocketContext>();
final List<String> _servicesAlreadyWired = [];
/// A list of clients currently connected to this server via WebSockets.
List<WebSocket> get clients => new List.unmodifiable(_clients);
/// Services that have already been hooked to fire socket events.
List<String> get servicesAlreadyWired => new List.unmodifiable(_servicesAlreadyWired);
final String endpoint;
/// Fired on incoming connections.
Stream<WebSocketContext> get onConnection => _onConnection.stream;
/// Fired when a user disconnects.
Stream<WebSocketContext> get onDisconnection => _onDisconnect.stream;
AngelWebSocket(String this.endpoint);
@ -108,7 +118,7 @@ class AngelWebSocket extends AngelPlugin {
..afterUpdated.listen(batch)
..afterRemoved.listen(batch);
servicesAlreadyWired.add(path);
_servicesAlreadyWired.add(path);
}
Future onConnect(WebSocketContext socket) async {}
@ -170,7 +180,7 @@ class AngelWebSocket extends AngelPlugin {
wireAllServices(Angel app) {
for (Pattern key in app.services.keys.where((x) {
return !servicesAlreadyWired.contains(x) &&
return !_servicesAlreadyWired.contains(x) &&
app.services[x] is HookedService;
})) {
hookupService(key, app.services[key]);

View file

@ -18,7 +18,7 @@ class WebSocketContext {
god.serialize(new WebSocketEvent(eventName: eventName, data: data)));
}
sendError(AngelHttpException error) => send("error", error);
sendError(AngelHttpException error) => send("error", error.toJson());
}
class _WebSocketEventTable {

View file

@ -1,14 +1,16 @@
name: angel_websocket
description: WebSocket plugin for Angel
environment:
sdk: ">=1.19.0"
version: 1.0.0-dev+5
author: Tobe O <thosakwe@gmail.com>
homepage: https://github.com/angel-dart/angel_websocket
dependencies:
angel_client: ">=1.0.0-dev <2.0.0"
angel_framework: ">=1.0.0-dev < 2.0.0"
json_god: ">=2.0.0-beta <3.0.0"
jwt: ">=0.1.4 <1.0.0"
uuid: ">=0.5.3 <1.0.0"
angel_auth: "^1.0.0-dev"
angel_client: "^1.0.0-dev"
angel_framework: "^1.0.0-dev"
uuid: "^0.5.3"
web_socket_channel: "^1.0.0"
dev_dependencies:
http: ">= 0.11.3 < 0.12.0"
test: ">= 0.12.13 < 0.13.0"
http: "^0.11.3"
test: "^0.12.15"

View file

@ -1 +0,0 @@
../packages