Renamed StreamSink to EventSink. Renamed signalError to addError.

sdk/lib/async/stream.dart

 // Copyright (c) 2013, 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.
 
 part of dart.async;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 50 matching lines @@
    * When the future completes, the stream will fire one event, either
    * data or error, and then close with a done-event.
    */
   factory Stream.fromFuture(Future<T> future) {
     _StreamImpl<T> stream = new _SingleStreamImpl<T>();
     future.then((value) {
         stream._add(value);
         stream._close();
       },
       onError: (error) {
-        stream._signalError(error);
+        stream._addError(error);
         stream._close();
       });
     return stream;
   }
 
   /**
    * Creates a single-subscription stream that gets its data from [data].
    */
   factory Stream.fromIterable(Iterable<T> data) {
     _PendingEvents iterableEvents = new _IterablePendingEvents<T>(data);
@@ skipping 125 matching lines @@
         result._setError(e);
       },
       onDone: () {
         result._setValue(value);
       },
       unsubscribeOnError: true);
     return result;
   }
 
   // Deprecated method, previously called 'pipe', retained for compatibility.
-  Future pipeInto(StreamSink<T> sink,
+  Future pipeInto(EventSink<T> sink,
                   {void onError(AsyncError error),
                    bool unsubscribeOnError}) {
     _FutureImpl<T> result = new _FutureImpl<T>();
     this.listen(
         sink.add,
-        onError: sink.signalError,
+        onError: sink.addError,
         onDone: () {
           sink.close();
           result._setValue(null);
         },
         unsubscribeOnError: unsubscribeOnError);
     return result;
   }
 
 
   /**
@@ skipping 605 matching lines @@
   /**
    * Resume after a pause.
    */
   void resume();
 }
 
 
 /**
  * An interface that abstracts sending events into a [Stream].
  */
-abstract class StreamSink<T> {
+abstract class EventSink<T> {
   void add(T event);
   /** Signal an async error to the receivers of this sink's values. */
-  void signalError(AsyncError errorEvent);
+  void addError(AsyncError errorEvent);
+  /** *Deprecated*. Use [addError] instead. */
+  void signalError(AsyncError errorEvent) {

[floitsch, 2013/03/07 14:19:34]

Move this method to StreamSink below. Users that use the new EventSink should
also use 'addError'.

Also: I know you don't like them, but if @deprecated works, please use it. It
will go away soon anyways.

[Lasse Reichstein Nielsen, 2013/03/08 10:18:25]

That's problematic. This is very much a consumed interface.
Users create event handlers that expect a StreamSink - and we currently give
them an EventListener. I'll change the structure here, so that code *expecting*
a StreamSink will still work. 
Code creating a StreamSink and sending it to code expecting an EventSink will
fail. I can't find a way to support both.
Done.

+    addError(errorEvent);
+  }
   void close();
 }
 
+/**
+ * *Deprecated*. Use [EventSink] instead.
+ */
+abstract class StreamSink<T> extends EventSink<T> {}

[floitsch, 2013/03/07 14:19:34]

ditto for @deprecated.

[Lasse Reichstein Nielsen, 2013/03/08 10:18:25]

Done.

+
 /** [Stream] wrapper that only exposes the [Stream] interface. */
 class StreamView<T> extends Stream<T> {
   Stream<T> _stream;
 
   StreamView(this._stream);
 
   bool get isBroadcast => _stream.isBroadcast;
 
   Stream<T> asBroadcastStream() => _stream.asBroadcastStream();
 
   StreamSubscription<T> listen(void onData(T value),
                                { void onError(AsyncError error),
                                  void onDone(),
                                  bool unsubscribeOnError }) {
     return _stream.listen(onData, onError: onError, onDone: onDone,
                           unsubscribeOnError: unsubscribeOnError);
   }
 }
 
 /**
- * [StreamSink] wrapper that only exposes the [StreamSink] interface.
+ * [EventSink] wrapper that only exposes the [EventSink] interface.
  */
-class StreamSinkView<T> implements StreamSink<T> {
-  final StreamSink<T> _sink;
+class EventSinkView<T> implements EventSink<T> {
+  final EventSink<T> _sink;
 
-  StreamSinkView(this._sink);
+  EventSinkView(this._sink);
 
   void add(T value) { _sink.add(value); }
-  void signalError(AsyncError error) { _sink.signalError(error); }
+  void addError(AsyncError error) { _sink.addError(error); }
+  void signalError(AsyncError error) { addError(error); }

[floitsch, 2013/03/07 14:19:34]

Add TODO?
Alternatively add comment to EventSink.signalError that instructs to 'grep' for
subclasses that implement signalError.

[Lasse Reichstein Nielsen, 2013/03/08 10:18:25]

Added TODO(8997) where relevant (and created bug for it).

   void close() { _sink.close(); }
 }
 
 
 /**
  * The target of a [Stream.pipe] call.
  *
  * The [Stream.pipe] call will pass itself to this object, and then return
  * the resulting [Future]. The pipe should complete the future when it's
  * done.
@@ skipping 14 matching lines @@
    * Create a [StreamTransformer] that delegates events to the given functions.
    *
    * This is actually a [StreamEventTransformer] where the event handling is
    * performed by the function arguments.
    * If an argument is omitted, it acts as the corresponding default method from
    * [StreamEventTransformer].
    *
    * Example use:
    *
    *     stringStream.transform(new StreamTransformer<String, String>(
-   *         handleData: (Strung value, StreamSink<String> sink) {
+   *         handleData: (Strung value, EventSink<String> sink) {
    *           sink.add(value);
    *           sink.add(value);  // Duplicate the incoming events.
    *         }));
    *
    */
   factory StreamTransformer({
-      void handleData(S data, StreamSink<T> sink),
-      void handleError(AsyncError error, StreamSink<T> sink),
-      void handleDone(StreamSink<T> sink)}) {
+      void handleData(S data, EventSink<T> sink),
+      void handleError(AsyncError error, EventSink<T> sink),
+      void handleDone(EventSink<T> sink)}) {
     return new _StreamTransformerImpl<S, T>(handleData,
                                             handleError,
                                             handleDone);
   }
 
   Stream<T> bind(Stream<S> stream);
 }
 
 
 /**
  * Base class for transformers that modifies stream events.
  *
  * A [StreamEventTransformer] transforms incoming Stream
  * events of one kind into outgoing events of (possibly) another kind.
  *
  * Subscribing on the stream returned by [bind] is the same as subscribing on
  * the source stream, except that events are passed through the [transformer]
  * before being emitted. The transformer may generate any number and
  * types of events for each incoming event. Pauses on the returned
  * subscription are forwarded to this stream.
  *
  * An example that duplicates all data events:
  *
  *     class DoubleTransformer<T> extends StreamEventTransformerBase<T, T> {
- *       void handleData(T data, StreamSink<T> sink) {
+ *       void handleData(T data, EventSink<T> sink) {
  *         sink.add(value);
  *         sink.add(value);
  *       }
  *     }
  *     someTypeStream.transform(new DoubleTransformer<Type>());
  *
  * The default implementations of the "handle" methods forward
  * the events unmodified. If using the default [handleData] the generic type [T]
  * needs to be assignable to [S].
  */
 abstract class StreamEventTransformer<S, T> implements StreamTransformer<S, T> {
   const StreamEventTransformer();
 
   Stream<T> bind(Stream<S> source) {
     return new EventTransformStream<S, T>(source, this);
   }
 
   /**
    * Act on incoming data event.
    *
    * The method may generate any number of events on the sink, but should
    * not throw.
    */
-  void handleData(S event, StreamSink<T> sink) {
+  void handleData(S event, EventSink<T> sink) {
     var data = event;
     sink.add(data);
   }
 
   /**
    * Act on incoming error event.
    *
    * The method may generate any number of events on the sink, but should
    * not throw.
    */
-  void handleError(AsyncError error, StreamSink<T> sink) {
-    sink.signalError(error);
+  void handleError(AsyncError error, EventSink<T> sink) {
+    sink.addError(error);
   }
 
   /**
    * Act on incoming done event.
    *
    * The method may generate any number of events on the sink, but should
    * not throw.
    */
-  void handleDone(StreamSink<T> sink){
+  void handleDone(EventSink<T> sink){
     sink.close();
   }
 }
 
 
 /**
  * Stream that transforms another stream by intercepting and replacing events.
  *
  * This [Stream] is a transformation of a source stream. Listening on this
  * stream is the same as listening on the source stream, except that events
@@ skipping 13 matching lines @@
                                  bool unsubscribeOnError }) {
     unsubscribeOnError = identical(true, unsubscribeOnError);
     return new _EventTransformStreamSubscription(_source, _transformer,
                                                  onData, onError, onDone,
                                                  unsubscribeOnError);
   }
 }
 
 class _EventTransformStreamSubscription<S, T>
     extends _BaseStreamSubscription<T>
-    implements _StreamOutputSink<T> {
+    implements _EventOutputSink<T> {
   /** The transformer used to transform events. */
   final StreamEventTransformer<S, T> _transformer;
   /** Whether to unsubscribe when emitting an error. */
   final bool _unsubscribeOnError;
   /** Source of incoming events. */
   StreamSubscription<S> _subscription;
-  /** Cached StreamSink wrapper for this class. */
-  StreamSink<T> _sink;
+  /** Cached EventSink wrapper for this class. */
+  EventSink<T> _sink;
 
   _EventTransformStreamSubscription(Stream<S> source,
                                     this._transformer,
                                     void onData(T data),
                                     void onError(AsyncError error),
                                     void onDone(),
                                     this._unsubscribeOnError)
       : super(onData, onError, onDone) {
-    _sink = new _StreamOutputSinkWrapper<T>(this);
+    _sink = new _EventOutputSinkWrapper<T>(this);
     _subscription = source.listen(_handleData,
                                   onError: _handleError,
                                   onDone: _handleDone);
   }
 
   void pause([Future pauseSignal]) {
     if (_subscription != null) _subscription.pause(pauseSignal);
   }
 
   void resume() {
@@ skipping 25 matching lines @@
 
   void _handleDone() {
     _subscription = null;
     try {
       _transformer.handleDone(_sink);
     } catch (e, s) {
       _sendError(_asyncError(e, s));
     }
   }
 
-  // StreamOutputSink interface.
+  // EventOutputSink interface.
   void _sendData(T data) {
     _onData(data);
   }
 
   void _sendError(AsyncError error) {
     _onError(error);
     if (_unsubscribeOnError) {
       cancel();
     }
   }
 
   void _sendDone() {
     // It's ok to cancel even if we have been unsubscribed already.
     cancel();
     _onDone();
   }
 }
 
-class _StreamOutputSinkWrapper<T> implements StreamSink<T> {
-  _StreamOutputSink _sink;
-  _StreamOutputSinkWrapper(this._sink);
+class _EventOutputSinkWrapper<T> implements EventSink<T> {
+  _EventOutputSink _sink;
+  _EventOutputSinkWrapper(this._sink);
 
-  void add(T data) => _sink._sendData(data);
-  void signalError(AsyncError error) => _sink._sendError(error);
-  void close() => _sink._sendDone();
+  void add(T data) { _sink._sendData(data); }
+  void addError(AsyncError error) { _sink._sendError(error); }
+  void signalError(AsyncError error) { addError(error); }
+  void close() { _sink._sendDone(); }
 }