Use StreamChannel.

lib/src/utils.dart

Index: lib/src/utils.dart
diff --git a/lib/src/utils.dart b/lib/src/utils.dart
index dde6b4af98680e976bc0c096de32e8798f583e81..8871b56c24c43d7a95cc13394bb4b1901ae85971 100644
--- a/lib/src/utils.dart
+++ b/lib/src/utils.dart
@@ -5,6 +5,10 @@
 import 'dart:async';
 
 import 'package:stack_trace/stack_trace.dart';
+import 'package:stream_channel/stream_channel.dart';
+
+import '../error_code.dart' as error_code;
+import 'exception.dart';
 
 typedef ZeroArgumentFunction();
 
@@ -64,6 +68,35 @@ tryFinally(body(), whenComplete()) {
   }
 }
 
+/// A transformer that silently drops [FormatException]s.
+final ignoreFormatExceptions = new StreamTransformer.fromHandlers(
+    handleError: (error, stackTrace, sink) {
+  if (error is FormatException) return;
+  sink.addError(error, stackTrace);
+});
+
+/// A transformer that sends error responses on [FormatException]s.
+final StreamChannelTransformer respondToFormatExceptions =
+    new _RespondToFormatExceptionsTransformer();
+
+/// The implementation of [respondToFormatExceptions].
+class _RespondToFormatExceptionsTransformer
+    implements StreamChannelTransformer {
+  StreamChannel bind(StreamChannel channel) {
+    var transformed;
+    transformed = channel.changeStream((stream) {
+      return stream.handleError((error) {
+        if (error is! FormatException) throw error;
+
+        var exception = new RpcException(
+            error_code.PARSE_ERROR, 'Invalid JSON: ${error.message}');
+        transformed.sink.add(exception.serialize(error.source));
+      });
+    });
+    return transformed;
+  }
+}
+
 /// Returns a [StreamSink] that wraps [sink] and maps each event added using
 /// [callback].
 StreamSink mapStreamSink(StreamSink sink, callback(event)) =>