Fix Stream.distinct.

sdk/lib/async/stream_pipe.dart

 // Copyright (c) 2012, 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;
 
 /** Runs user code and takes actions depending on success or failure. */
 _runUserCode<T>(
     T userCode(), onSuccess(T value), onError(error, StackTrace stackTrace)) {
   try {
@@ skipping 320 matching lines @@
         // this from source.
         sink._close();
       }
     }
   }
 }
 
 /**
  * A [_ForwardingStreamSubscription] with one extra state field.
  *
- * Use by several different classes, some storing an integer, others a bool.
+ * Use by several different classes, storing an integer, bool or general.
  */
 class _StateStreamSubscription<T> extends _ForwardingStreamSubscription<T, T> {
   // Raw state field. Typed access provided by getters and setters below.
   var _sharedState;
 
   _StateStreamSubscription(_ForwardingStream<T, T> stream, void onData(T data),
       Function onError, void onDone(), bool cancelOnError, this._sharedState)
       : super(stream, onData, onError, onDone, cancelOnError);
 
   bool get _flag => _sharedState;
   void set _flag(bool flag) {
     _sharedState = flag;
   }
 
   int get _count => _sharedState;
   void set _count(int count) {
     _sharedState = count;
   }
+
+  Object get _value => _sharedState;
+  void set _value(Object value) {
+    _sharedState = value;
+  }
 }
 
 class _TakeWhileStream<T> extends _ForwardingStream<T, T> {
   final _Predicate<T> _test;
 
   _TakeWhileStream(Stream<T> source, bool test(T value))
       : this._test = test,
         super(source);
 
   void _handleData(T inputEvent, _EventSink<T> sink) {
@@ skipping 76 matching lines @@
       sink._add(inputEvent);
     }
   }
 }
 
 typedef bool _Equality<T>(T a, T b);
 
 class _DistinctStream<T> extends _ForwardingStream<T, T> {
   static var _SENTINEL = new Object();
 
-  _Equality<T> _equals;
-  var _previous = _SENTINEL;
+  final _Equality<T> _equals;
 
   _DistinctStream(Stream<T> source, bool equals(T a, T b))
       : _equals = equals,
         super(source);
 
+  StreamSubscription<T> _createSubscription(void onData(T data),
+      Function onError, void onDone(), bool cancelOnError) {
+    return new _StateStreamSubscription<T>(
+        this, onData, onError, onDone, cancelOnError, _SENTINEL);
+  }
+
   void _handleData(T inputEvent, _EventSink<T> sink) {
-    if (identical(_previous, _SENTINEL)) {
-      _previous = inputEvent;
-      return sink._add(inputEvent);
+    _StateStreamSubscription<T> subscription = sink;
+    var previous = subscription._value;
+    if (identical(previous, _SENTINEL)) {
+      // First event.
+      subscription._value = inputEvent;
+      sink._add(inputEvent);
     } else {
+      T previousEvent = previous;
       bool isEqual;
       try {
         if (_equals == null) {
-          isEqual = (_previous == inputEvent);
+          isEqual = (previousEvent == inputEvent);
         } else {
-          isEqual = _equals(_previous as Object/*=T*/, inputEvent);
+          isEqual = _equals(previousEvent, inputEvent);
         }
       } catch (e, s) {
         _addErrorWithReplacement(sink, e, s);
-        return null;
+        return;
       }
       if (!isEqual) {
         sink._add(inputEvent);
-        _previous = inputEvent;
+        subscription._value = inputEvent;
       }
     }
   }
 }