Client Error Stream

lib/src/client/client.dart

@@ -30,4 +30,9 @@ abstract class Client {
    * the new [Channel]
    */
   Future<Channel> channel();
+
+  /**
+   * Register listener for errors
+   */
+  StreamSubscription<Exception> errorListener(void onData(Exception error), { Function onError, void onDone(), bool cancelOnError });
 }

lib/src/client/impl/client_impl.dart

@@ -19,6 +19,9 @@ class _ClientImpl implements Client {
   Completer _connected;
   Completer _clientClosed;
 
+  //Error Stream
+  StreamController _error = new StreamController.broadcast();
+
   _ClientImpl({ConnectionSettings settings}) {
 
     // Use defaults if no settings specified
@@ -166,6 +169,10 @@ class _ClientImpl implements Client {
       return;
     }
 
+    if (_error != null && _error.hasListener && !_error.isClosed) {
+      _error.add(ex);
+    }
+
     switch (ex.runtimeType) {
       case FatalException:
       case ConnectionException:
@@ -236,6 +243,7 @@ class _ClientImpl implements Client {
         _socket.destroy();
         _socket = null;
         _connected = null;
+        _error.close();
         _clientClosed.complete();
       });
 
@@ -270,4 +278,7 @@ class _ClientImpl implements Client {
       return userChannel._channelOpened.future;
     });
   }
+
+  StreamSubscription<Exception> errorListener(void onData(Exception error), { Function onError, void onDone(), bool cancelOnError}) => _error.stream.listen(onData, onError : onError, onDone : onDone, cancelOnError : cancelOnError);
+
 }

test/lib/exception_handling_test.dart

@@ -1,6 +1,7 @@
 library dart_amqp.test.exceptions;
 
 import "dart:typed_data";
+import "dart:async";
 
 import "../packages/unittest/unittest.dart";
 import "../packages/mock/mock.dart";
@@ -350,6 +351,26 @@ main({bool enableLogger : true}) {
         }).catchError(expectAsync(handleError));
       });
     });
-
+    group("error stream:", () {
+      test("fatal exception", () async {
+        void handleError(ex) {
+          expect(ex, new isInstanceOf<FatalException>());
+        }
+        server.shutdown()
+            .then((_) => server.listen(client.settings.host, client.settings.port))
+            .then((_) {
+              generateHandshakeMessages(frameWriter, server);
+              return client
+                  .connect()
+                  .then((_) {
+                    client.errorListener((ex) => handleError(ex));
+                    return server.shutdown()
+                        .then((_) => new Future.delayed(
+                        new Duration(seconds: 5) + server.responseDelay))
+                        .then((_) => fail("Expected an exception to be thrown"));
+          });
+        });
+      });
+    });
   });
 }