Use StreamChannel.

lib/src/utils.dart

 // Copyright (c) 2014, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 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();
 
 /// Like [new Future.sync], but automatically wraps the future in a
 /// [Chain.track] call.
 Future syncFuture(callback()) => Chain.track(new Future.sync(callback));
 
 /// Returns a sentence fragment listing the elements of [iter].
 ///
 /// This converts each element of [iter] to a string and separates them with
@@ skipping 39 matching lines @@
   }
 
   if (result is! Future) {
     whenComplete();
     return result;
   } else {
     return result.whenComplete(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)) =>
     new _MappedStreamSink(sink, callback);
 
 /// A [StreamSink] wrapper that maps each event added to the sink.
 class _MappedStreamSink implements StreamSink {
   final StreamSink _inner;
   final Function _callback;
 
   Future get done => _inner.done;
 
   _MappedStreamSink(this._inner, this._callback);
 
   void add(event) => _inner.add(_callback(event));
   void addError(error, [StackTrace stackTrace]) =>
       _inner.addError(error, stackTrace);
   Future addStream(Stream stream) => _inner.addStream(stream.map(_callback));
   Future close() => _inner.close();
 }