// 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();