diff --git a/README.md b/README.md
index 07f70ac6..4f5d527b 100644
--- a/README.md
+++ b/README.md
@@ -1,6 +1,6 @@
 # angel_client
 
-[![pub 1.0.4](https://img.shields.io/badge/pub-1.0.4-brightgreen.svg)](https://pub.dartlang.org/packages/angel_client)
+[![pub 1.0.5](https://img.shields.io/badge/pub-1.0.5-brightgreen.svg)](https://pub.dartlang.org/packages/angel_client)
 [![build status](https://travis-ci.org/angel-dart/client.svg)](https://travis-ci.org/angel-dart/client)
 
 Client library for the Angel framework.
diff --git a/lib/angel_client.dart b/lib/angel_client.dart
index 2061f125..04ed80c8 100644
--- a/lib/angel_client.dart
+++ b/lib/angel_client.dart
@@ -22,6 +22,9 @@ abstract class Angel {
 
   Angel(String this.basePath);
 
+  /// Fired whenever a WebSocket is successfully authenticated.
+  Stream<AngelAuthResult> get onAuthenticated;
+
   Future<AngelAuthResult> authenticate(
       {String type,
       credentials,
@@ -41,7 +44,7 @@ abstract class Angel {
   /// Logs the current user out of the application.
   Future logout();
 
-  Service service<T>(String path, {Type type, AngelDeserializer deserializer});
+  Service<T> service<T>(String path, {Type type, AngelDeserializer deserializer});
 
   Future<http.Response> delete(String url, {Map<String, String> headers});
 
@@ -87,10 +90,30 @@ class AngelAuthResult {
 }
 
 /// Queries a service on an Angel server, with the same API.
-abstract class Service {
+abstract class Service<T> {
+  /// Fired on `indexed` events.
+  Stream<T> get onIndexed;
+
+  /// Fired on `read` events.
+  Stream<T> get onRead;
+
+  /// Fired on `created` events.
+  Stream<T> get onCreated;
+
+  /// Fired on `modified` events.
+  Stream<T> get onModified;
+
+  /// Fired on `updated` events.
+  Stream<T> get onUpdated;
+
+  /// Fired on `removed` events.
+  Stream<T> get onRemoved;
+
   /// The Angel instance powering this service.
   Angel get app;
 
+  Future close();
+
   /// Retrieves all resources.
   Future index([Map params]);
 
diff --git a/lib/base_angel_client.dart b/lib/base_angel_client.dart
index 7fadd73a..e9552ff1 100644
--- a/lib/base_angel_client.dart
+++ b/lib/base_angel_client.dart
@@ -54,8 +54,14 @@ AngelHttpException failure(http.Response response, {error, StackTrace stack}) {
 }
 
 abstract class BaseAngelClient extends Angel {
+  final StreamController<AngelAuthResult> _onAuthenticated =
+      new StreamController<AngelAuthResult>();
+  final List<Service> _services = [];
   final http.BaseClient client;
 
+  @override
+  Stream<AngelAuthResult> get onAuthenticated => _onAuthenticated.stream;
+
   BaseAngelClient(this.client, String basePath) : super(basePath);
 
   @override
@@ -87,7 +93,9 @@ abstract class BaseAngelClient extends Angel {
                   "Auth endpoint '$url' did not return a proper response.");
         }
 
-        return new AngelAuthResult.fromMap(json);
+        var r = new AngelAuthResult.fromMap(json);
+        _onAuthenticated.add(r);
+        return r;
       } catch (e, st) {
         throw failure(response, error: e, stack: st);
       }
@@ -117,7 +125,9 @@ abstract class BaseAngelClient extends Angel {
                   "Auth endpoint '$url' did not return a proper response.");
         }
 
-        return new AngelAuthResult.fromMap(json);
+        var r = new AngelAuthResult.fromMap(json);
+        _onAuthenticated.add(r);
+        return r;
       } catch (e, st) {
         throw failure(response, error: e, stack: st);
       }
@@ -126,6 +136,10 @@ abstract class BaseAngelClient extends Angel {
 
   Future close() async {
     client.close();
+    _onAuthenticated.close();
+    Future.wait(_services.map((s) => s.close())).then((_) {
+      _services.clear();
+    });
   }
 
   Future logout() async {
@@ -161,10 +175,13 @@ abstract class BaseAngelClient extends Angel {
   }
 
   @override
-  Service service<T>(String path, {Type type, AngelDeserializer deserializer}) {
+  Service<T> service<T>(String path,
+      {Type type, AngelDeserializer deserializer}) {
     String uri = path.toString().replaceAll(straySlashes, "");
-    return new BaseAngelService(client, this, '$basePath/$uri',
+    var s = new BaseAngelService<T>(client, this, '$basePath/$uri',
         deserializer: deserializer);
+    _services.add(s);
+    return s;
   }
 
   String _join(url) {
@@ -208,13 +225,48 @@ abstract class BaseAngelClient extends Angel {
   }
 }
 
-class BaseAngelService extends Service {
+class BaseAngelService<T> extends Service<T> {
   @override
   final BaseAngelClient app;
   final String basePath;
   final http.BaseClient client;
   final AngelDeserializer deserializer;
 
+  final StreamController<T> _onIndexed = new StreamController<T>(),
+      _onRead = new StreamController<T>(),
+      _onCreated = new StreamController<T>(),
+      _onModified = new StreamController<T>(),
+      _onUpdated = new StreamController<T>(),
+      _onRemoved = new StreamController<T>();
+
+  @override
+  Stream<T> get onIndexed => _onIndexed.stream;
+
+  @override
+  Stream<T> get onRead => _onRead.stream;
+
+  @override
+  Stream<T> get onCreated => _onCreated.stream;
+
+  @override
+  Stream<T> get onModified => _onModified.stream;
+
+  @override
+  Stream<T> get onUpdated => _onUpdated.stream;
+
+  @override
+  Stream<T> get onRemoved => _onRemoved.stream;
+
+  @override
+  Future close() async {
+    _onIndexed.close();
+    _onRead.close();
+    _onCreated.close();
+    _onModified.close();
+    _onUpdated.close();
+    _onRemoved.close();
+  }
+
   BaseAngelService(this.client, this.app, this.basePath, {this.deserializer});
 
   deserialize(x) {
@@ -244,8 +296,15 @@ class BaseAngelService extends Service {
       }
 
       final json = JSON.decode(response.body);
-      if (json is! List) return json;
-      return json.map(deserialize).toList();
+
+      if (json is! List) {
+        _onIndexed.add(json);
+        return json;
+      }
+
+      var r = json.map(deserialize).toList();
+      _onIndexed.add(r);
+      return r;
     } catch (e, st) {
       throw failure(response, error: e, stack: st);
     }
@@ -261,7 +320,9 @@ class BaseAngelService extends Service {
         throw failure(response);
       }
 
-      return deserialize(JSON.decode(response.body));
+      var r = deserialize(JSON.decode(response.body));
+      _onRead.add(r);
+      return r;
     } catch (e, st) {
       throw failure(response, error: e, stack: st);
     }
@@ -277,7 +338,9 @@ class BaseAngelService extends Service {
         throw failure(response);
       }
 
-      return deserialize(JSON.decode(response.body));
+      var r = deserialize(JSON.decode(response.body));
+      _onCreated.add(r);
+      return r;
     } catch (e, st) {
       throw failure(response, error: e, stack: st);
     }
@@ -293,7 +356,9 @@ class BaseAngelService extends Service {
         throw failure(response);
       }
 
-      return deserialize(JSON.decode(response.body));
+      var r = deserialize(JSON.decode(response.body));
+      _onModified.add(r);
+      return r;
     } catch (e, st) {
       throw failure(response, error: e, stack: st);
     }
@@ -309,7 +374,9 @@ class BaseAngelService extends Service {
         throw failure(response);
       }
 
-      return deserialize(JSON.decode(response.body));
+      var r = deserialize(JSON.decode(response.body));
+      _onUpdated.add(r);
+      return r;
     } catch (e, st) {
       throw failure(response, error: e, stack: st);
     }
@@ -325,7 +392,9 @@ class BaseAngelService extends Service {
         throw failure(response);
       }
 
-      return deserialize(JSON.decode(response.body));
+      var r = deserialize(JSON.decode(response.body));
+      _onRemoved.add(r);
+      return r;
     } catch (e, st) {
       throw failure(response, error: e, stack: st);
     }
diff --git a/lib/io.dart b/lib/io.dart
index b1ec9f1f..1d450681 100644
--- a/lib/io.dart
+++ b/lib/io.dart
@@ -10,27 +10,40 @@ export 'angel_client.dart';
 
 /// Queries an Angel server via REST.
 class Rest extends BaseAngelClient {
+  final List<Service> _services = [];
+
   Rest(String path) : super(new http.Client(), path);
 
   @override
-  Service service<T>(String path, {Type type, AngelDeserializer deserializer}) {
+  Service<T> service<T>(String path, {Type type, AngelDeserializer deserializer}) {
     String uri = path.replaceAll(straySlashes, "");
-    return new RestService(
+    var s = new RestService<T>(
         client, this, "$basePath/$uri", T != dynamic ? T : type);
+    _services.add(s);
+    return s;
   }
+
   @override
   Stream<String> authenticateViaPopup(String url, {String eventName: 'token'}) {
     throw new UnimplementedError('Opening popup windows is not supported in the `dart:io` client.');
   }
+
+  Future close() async {
+    super.close();
+    Future.wait(_services.map((s) => s.close())).then((_) {
+      _services.clear();
+    });
+  }
 }
 
 /// Queries an Angel service via REST.
-class RestService extends BaseAngelService {
+class RestService<T> extends BaseAngelService<T> {
   final Type type;
 
   RestService(http.BaseClient client, Angel app, String url, this.type)
       : super(client, app, url);
 
+  @override
   deserialize(x) {
     if (type != null) {
       return x.runtimeType == type
@@ -49,29 +62,4 @@ class RestService extends BaseAngelService {
 
     return super.makeBody(x);
   }
-
-  @override
-  Future index([Map params]) async {
-    final items = await super.index(params);
-    if (items is! List) return items;
-    return items.map(deserialize).toList();
-  }
-
-  @override
-  Future read(id, [Map params]) => super.read(id, params).then(deserialize);
-
-  @override
-  Future create(data, [Map params]) =>
-      super.create(data, params).then(deserialize);
-
-  @override
-  Future modify(id, data, [Map params]) =>
-      super.modify(id, data, params).then(deserialize);
-
-  @override
-  Future update(id, data, [Map params]) =>
-      super.update(id, data, params).then(deserialize);
-
-  @override
-  Future remove(id, [Map params]) => super.remove(id, params).then(deserialize);
 }
diff --git a/pubspec.yaml b/pubspec.yaml
index 963444d6..0ff7ee9d 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -1,5 +1,5 @@
 name: angel_client
-version: 1.0.4
+version: 1.0.5
 description: Client library for the Angel framework.
 author: Tobe O <thosakwe@gmail.com>
 homepage: https://github.com/angel-dart/angel_client
diff --git a/test/io_test.dart b/test/io_test.dart
index fc7056a7..61c8882e 100644
--- a/test/io_test.dart
+++ b/test/io_test.dart
@@ -19,7 +19,7 @@ main() {
       httpServer =
           await serverApp.startServer(InternetAddress.LOOPBACK_IP_V4, 0);
       url = "http://localhost:${httpServer.port}";
-      serverApp.use("/postcards", new server.MemoryService<Postcard>());
+      serverApp.use("/postcards", new server.TypedService<Postcard>(new server.MapService()));
       serverPostcards = serverApp.service("postcards");
 
       clientApp = new client.Rest(url);
@@ -94,7 +94,7 @@ main() {
 
     test("modify/update", () async {
       var innerPostcards =
-          serverPostcards.inner as server.MemoryService<Postcard>;
+          serverPostcards.inner as server.TypedService<Postcard>;
       print(innerPostcards.items);
       Postcard mecca = await clientTypedPostcards
           .create(new Postcard(location: "Mecca", message: "Pilgrimage"));