// Stream controllers
final emailS = BehaviorSubject.seeded('');
final passwordS = BehaviorSubject.seeded('');
final isLoadingS = BehaviorSubject.seeded(false);
final submitLoginS = StreamController<void>();
final subjects = [emailS, passwordS, isLoadingS, submitLoginS];

// Email error and password error stream
final emailError$ = emailS.map(validator.validateEmail).distinct().share();
 
final passwordError$ =
    passwordS.map(validator.validatePassword).distinct().share();

// Submit stream
final submit$ = submitLoginS.stream
    .throttleTime(const Duration(milliseconds: 500))
    .withLatestFrom<bool, bool>(
      Rx.combineLatest<Set<ValidationError>, bool>(
        [emailError$, passwordError$],
        (listOfSets) => listOfSets.every((errorsSet) => errorsSet.isEmpty),
      ),
      (_, isValid) => isValid,
    )
    .share();

// Message stream
final message$ = Rx.merge(
  [
    submit$
        .where((isValid) => isValid)
        .withLatestFrom2(
          emailS,
          passwordS,
          (_, email, password) => Credential(
            email: email,
            password: password,
          ),
        )
        .exhaustMap(
          (credential) => interactor.performLogin(
            credential,
            isLoadingS,
          ),
        ),
    submit$
        .where((isValid) => !isValid)
        .map((_) => const InvalidInformationMessage()),
  ],
).publish();