feat(operators): startWithFuture

lib/src/operators/index.dart

@@ -5,5 +5,6 @@ export 'do_on.dart';
 export 'done_on_error.dart';
 export 'flat_map_batches.dart';
 export 'ignore.dart';
+export 'start_with_future.dart';
 export 'to_single_subscription.dart';
 export 'void.dart';

lib/src/operators/start_with_future.dart

@@ -0,0 +1,22 @@
+import 'package:rxdart/rxdart.dart';
+
+/// Just like [startWith], but accepts a [Future]
+///
+/// ### Example
+///
+///    Stream.fromIterable([1, 2, 3])
+///      .startWithFuture(Future(() async => 0))
+///      .listen(print); // prints 0, 1, 2, 3
+///
+extension StartWithFuture<T> on Stream<T> {
+  /// Just like [startWith], but accepts a [Future]
+  ///
+  /// ### Example
+  ///
+  ///    Stream.fromIterable([1, 2, 3])
+  ///      .startWithFuture(Future(() async => 0))
+  ///      .listen(print); // prints 0, 1, 2, 3
+  ///
+  Stream<T> startWithFuture(Future<T> startFuture) =>
+      Rx.concatEager([startFuture.asStream(), this]);
+}

test/operators/start_with_future_test.dart

@@ -0,0 +1,29 @@
+import 'dart:async';
+
+import 'package:rxdart_ext/rxdart_ext.dart';
+import 'package:test/test.dart';
+
+void main() {
+  test(
+    'Stream.startWithFuture',
+    () async {
+      final stream = _createStreamForTest().startWithFuture(_startFuture());
+
+      expect(stream, emitsInOrder(<int>[0, 1, 2, 3]));
+    },
+  );
+}
+
+Future<int> _startFuture() async {
+  await Future<void>.delayed(Duration(seconds: 1, milliseconds: 500));
+  return 0;
+}
+
+Stream<int> _createStreamForTest() async* {
+  yield 1;
+
+  await Future<void>.delayed(Duration(seconds: 1));
+
+  yield 2;
+  yield 3;
+}