final _email = StreamController<String>();

final _email = StreamController<String>.broadcast();

class Bloc with Validators {
  final _email = StreamController<String>();
  final _password = StreamController<String>();

  Function(String) get sinkEmail => _email.sink.add;

  Stream<String> get email => _email.stream.transform(validateEmail);
//  Stream<String> get email => _email.stream; //THIS IS WORKING

  Function(String) get sinkPass => _password.sink.add;

  Stream<String> get pass => _password.stream.transform(validatePassword);
//  Stream<String> get pass => _password.stream; //THIS TOO OFC

  dispose() {
    _email.close();
    _password.close();
  }
}

Widget _buildEmailField() {
  return StreamBuilder(
      stream: bloc.email,
      builder: (context, snapshot) {
        return TextField(
          keyboardType: TextInputType.emailAddress,
          onChanged: bloc.sinkEmail,
          decoration: InputDecoration(
            border: OutlineInputBorder(),
            labelText: 'Email Adress',
            hintText: 'you@example.com',
            errorText: snapshot.error,
          ),
        );
      });
}

class Validators {
  final validateEmail =
      StreamTransformer<String, String>.fromHandlers(handleData: (email, sink) {
    if (email.contains('@')) {
      sink.add(email);
    } else {
      sink.addError('Enter a valid email');
    }
  });
}

flutter: The following StateError was thrown building NotificationListener<KeepAliveNotification>:
flutter: Bad state: Stream has already been listened to.
flutter:
flutter: When the exception was thrown, this was the stack: