Remove RxDart dependency from bloc package

packages/bloc/lib/src/bloc.dart

@@ -1,7 +1,6 @@
 import 'dart:async';
 
 import 'package:meta/meta.dart';
-import 'package:rxdart/rxdart.dart';
 
 import '../bloc.dart';
 
@@ -10,24 +9,25 @@ import '../bloc.dart';
 /// and transforms them into a `Stream` of `States` as output.
 /// {@endtemplate}
 abstract class Bloc<Event, State> extends Stream<State> implements Sink<Event> {
-  final PublishSubject<Event> _eventSubject = PublishSubject<Event>();
+  final _eventController = StreamController<Event>.broadcast();
+  final _stateController = StreamController<State>.broadcast();
 
-  BehaviorSubject<State> _stateSubject;
+  State _state;
 
   /// Returns the current [state] of the [bloc].
-  State get state => _stateSubject.value;
+  State get state => _state;
 
   /// Returns the [state] before any `events` have been [add]ed.
   State get initialState;
 
   /// Returns whether the `Stream<State>` is a broadcast stream.
   @override
-  bool get isBroadcast => _stateSubject.isBroadcast;
+  bool get isBroadcast => true;
 
   /// {@macro bloc}
   Bloc() {
-    _stateSubject = BehaviorSubject<State>.seeded(initialState);
-    _bindStateSubject();
+    _state = initialState;
+    _bindEventsToStates();
   }
 
   /// Adds a subscription to the `Stream<State>`.
@@ -41,7 +41,7 @@ abstract class Bloc<Event, State> extends Stream<State> implements Sink<Event> {
     void Function() onDone,
     bool cancelOnError,
   }) {
-    return _stateSubject.listen(
+    return _buildStateStream().listen(
       onData,
       onError: onError,
       onDone: onDone,
@@ -77,7 +77,7 @@ abstract class Bloc<Event, State> extends Stream<State> implements Sink<Event> {
     try {
       BlocSupervisor.delegate.onEvent(this, event);
       onEvent(event);
-      _eventSubject.sink.add(event);
+      _eventController.add(event);
     } on dynamic catch (error) {
       _handleError(error);
     }
@@ -93,8 +93,8 @@ abstract class Bloc<Event, State> extends Stream<State> implements Sink<Event> {
   @override
   @mustCallSuper
   Future<void> close() async {
-    await _eventSubject.close();
-    await _stateSubject.close();
+    await _eventController.close();
+    await _stateController.close();
   }
 
   /// Transforms the [events] stream along with a [next] function into
@@ -157,15 +157,20 @@ abstract class Bloc<Event, State> extends Stream<State> implements Sink<Event> {
   /// ```
   Stream<State> transformStates(Stream<State> states) => states;
 
-  void _bindStateSubject() {
+  Stream<State> _buildStateStream() async* {
+    yield _state;
+    yield* _stateController.stream;
+  }
+
+  void _bindEventsToStates() {
     Event currentEvent;
 
-    transformStates(transformEvents(_eventSubject, (event) {
+    transformStates(transformEvents(_eventController.stream, (event) {
       currentEvent = event;
       return mapEventToState(currentEvent).handleError(_handleError);
     })).forEach(
       (nextState) {
-        if (state == nextState || _stateSubject.isClosed) return;
+        if (state == nextState || _stateController.isClosed) return;
         final transition = Transition(
           currentState: state,
           event: currentEvent,
@@ -174,7 +179,8 @@ abstract class Bloc<Event, State> extends Stream<State> implements Sink<Event> {
         try {
           BlocSupervisor.delegate.onTransition(this, transition);
           onTransition(transition);
-          _stateSubject.add(nextState);
+          _state = nextState;
+          _stateController.add(nextState);
         } on dynamic catch (error) {
           _handleError(error);
         }

packages/bloc/pubspec.yaml

@@ -10,10 +10,11 @@ environment:
 
 dependencies:
   meta: ^1.1.6
-  rxdart: ^0.23.0
 
 dev_dependencies:
   test: ">=1.3.0 <2.0.0"
   test_coverage: ^0.2.0
   mockito: ^4.0.0
   effective_dart: ^1.2.0
+  rxdart: ^0.23.0
+