Add distinctBy

lib/src/transformers/distinct_by.dart

@@ -0,0 +1,14 @@
+import 'dart:async';
+
+/// Extends the Stream class with the ability to skip items that have previously
+/// been emitted.
+extension DistinctByExtension<T> on Stream<T> {
+  /// Creates a Stream where data events are skipped if they have already
+  /// been emitted before.
+  ///
+  /// [Interactive marble diagram](http://rxmarbles.com/#distinct)
+  Stream<T> distinctBy<S>(
+    S Function(T it) distincter,
+  ) =>
+      distinct((prev, next) => distincter(prev) == distincter(next));
+}

lib/transformers.dart

@@ -3,6 +3,7 @@ library rx_transformers;
 export 'src/transformers/default_if_empty.dart';
 export 'src/transformers/delay.dart';
 export 'src/transformers/dematerialize.dart';
+export 'src/transformers/distinct_by.dart';
 export 'src/transformers/distinct_unique.dart';
 export 'src/transformers/do.dart';
 export 'src/transformers/exhaust_map.dart';

test/transformers/distinct_by_test.dart

@@ -0,0 +1,31 @@
+import 'dart:async';
+
+import 'package:rxdart/rxdart.dart';
+import 'package:test/test.dart';
+
+void main() {
+  test('Rx.distinctBy', () async {
+    await expectLater(
+        Stream.fromIterable(['123', 'abc']).distinctBy((value) => value.length),
+        emitsInOrder(<dynamic>[
+          '123',
+          emitsDone
+        ]));
+    await expectLater(
+        Stream.fromIterable(['123', 'abc', '1234']).distinctBy((value) => value.length),
+        emitsInOrder(<dynamic>[
+          '123',
+          '1234',
+          emitsDone
+        ]));
+    await expectLater(
+        Stream.fromIterable(['1234', 'abc', '1234', 'abcd']).distinctBy((value) => value.length),
+        emitsInOrder(<dynamic>[
+          '1234',
+          'abc',
+          '1234',
+          emitsDone
+        ]));
+  });
+
+}