Java 类io.reactivex.CompletableEmitter 实例源码

public static Completable userlessAuthentication(
        final RedditClient reddit,
        final Credentials credentials) {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(CompletableEmitter e) throws Exception {
            try {
                OAuthData oAuthData = reddit.getOAuthHelper().easyAuth(credentials);
                reddit.authenticate(oAuthData);
                e.onComplete();
            } catch (Exception ex) {
                e.onError(ex);
            }
        }
    });
}

public static Completable userAuthentication(
        final RedditClient reddit,
        final Credentials credentials,
        final String url) {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(CompletableEmitter e) throws Exception {
            OAuthHelper oAuthHelper = reddit.getOAuthHelper();

            try {
                OAuthData oAuthData = oAuthHelper.onUserChallenge(url, credentials);
                reddit.authenticate(oAuthData);
                AuthenticationManager.get().onAuthenticated(oAuthData);
                e.onComplete();
            } catch (Exception ex) {
                e.onError(ex);
            }
        }
    });
}

public static Completable logout(final Credentials credentials) {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(CompletableEmitter e) throws Exception {
            try {
                AuthenticationManager.get().getRedditClient().getOAuthHelper()
                        .revokeAccessToken(credentials);
                AuthenticationManager.get().getRedditClient().getOAuthHelper()
                        .revokeRefreshToken(credentials);
                // Calling deauthenticate() isn't really necessary, since revokeAccessToken()
                // already calls it.
                // AuthenticationManager.get().getRedditClient().deauthenticate();

                // As of JRAW 9.0.0, revoking the access/refresh token does not update the
                // auth state to NONE (it instead remains as NEEDS_REFRESH), so to completely
                // restart the session to a blank state you should re-instantiate the
                // AuthenticationManager. See https://github.com/mattbdean/JRAW/issues/196

                // AuthenticationManager.get().init(...., ....); uncomment this line.
                e.onComplete();
            } catch (Exception ex) {
                e.onError(ex);
            }
        }
    });
}

@Override
public Completable deleteProfile(final String uid) {
    return Completable.create(
            new CompletableOnSubscribe() {
                @Override
                public void subscribe(final CompletableEmitter e) throws Exception {
                    final DatabaseReference rootRef = FirebaseDatabase.getInstance().getReference();
                    rootRef.child(USER_PROFILES)
                            .child(uid)
                            .setValue(null)
                            .addOnCompleteListener(new OnCompleteListener<Void>() {
                                @Override
                                public void onComplete(@NonNull Task<Void> task) {
                                    if (task.isSuccessful()) {
                                        e.onComplete();
                                    } else {
                                        e.onError(task.getException());
                                    }
                                }
                            });
                }

            });
}

@Override
public Completable updateProfile(final Profile profile) {
    return Completable.create(
            new CompletableOnSubscribe() {
                @Override
                public void subscribe(final CompletableEmitter e) throws Exception {
                    final DatabaseReference rootRef = FirebaseDatabase.getInstance().getReference();
                    rootRef.child(USER_PROFILES)
                            .child(profile.getUid())
                            .setValue(profile)
                            .addOnCompleteListener(new OnCompleteListener<Void>() {
                                @Override
                                public void onComplete(@NonNull Task<Void> task) {
                                    if (task.isSuccessful()) {
                                        e.onComplete();
                                    } else {
                                        e.onError(task.getException());
                                    }
                                }
                            });
                }

            });
}

@Override
public Completable createAccount(final Credentials cred) {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(final CompletableEmitter e) throws Exception {
            if (auth == null) {
                auth = FirebaseAuth.getInstance();
            }

            auth.createUserWithEmailAndPassword(cred.getEmail(), cred.getPassword())
                    .addOnCompleteListener(new OnCompleteListener<AuthResult>() {
                        @Override
                        public void onComplete(@NonNull Task<AuthResult> task) {
                            if (task.isSuccessful()) {
                                e.onComplete();
                            } else {
                                e.onError(task.getException());
                            }
                        }
                    });

        }
    });
}

@Override
public Completable attemptLogin(final Credentials cred) {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(final CompletableEmitter e) throws Exception {
            if (auth == null) {
                auth = FirebaseAuth.getInstance();
            }

            auth.signInWithEmailAndPassword(cred.getEmail(), cred.getPassword())
                    .addOnCompleteListener(new OnCompleteListener<AuthResult>() {
                        @Override
                        public void onComplete(@NonNull Task<AuthResult> task) {
                            if (task.isSuccessful()) {
                                e.onComplete();
                            } else {
                                e.onError(task.getException());
                            }
                        }
                    });
        }
    });
}

@Override
public Completable deleteUser() {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(final CompletableEmitter e) throws Exception {
            if (auth == null) {
                auth = FirebaseAuth.getInstance();
            }

            final FirebaseUser user = auth.getCurrentUser();

            user.delete().addOnCompleteListener(new OnCompleteListener<Void>() {
                @Override
                public void onComplete(@NonNull Task<Void> task) {
                    if (task.isSuccessful()) {
                        e.onComplete();
                    } else {
                        e.onError(task.getException());
                    }
                }
            });

        }
    });
}

/**
 * Set the given value on the specified {@link DatabaseReference}.
 *
 * @param ref reference represents a particular location in your database.
 * @param value value to update.
 * @return a {@link Completable} which is complete when the set value call finish successfully.
 */
@NonNull
public static Completable setValue(@NonNull final DatabaseReference ref,
                                   final Object value) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(@io.reactivex.annotations.NonNull final CompletableEmitter e) throws Exception {
         ref.setValue(value).addOnSuccessListener(new OnSuccessListener<Void>() {
            @Override public void onSuccess(Void aVoid) {
               e.onComplete();
            }
         }).addOnFailureListener(new OnFailureListener() {
            @Override public void onFailure(@NonNull Exception exception) {
               e.onError(exception);
            }
         });
      }
   });
}

/**
 * Update the specific child keys to the specified values.
 *
 * @param ref        reference represents a particular location in your database.
 * @param updateData The paths to update and their new values
 * @return a {@link Completable} which is complete when the update children call finish successfully.
 */
@NonNull
public static Completable updateChildren(@NonNull final DatabaseReference ref,
                                         @NonNull final Map<String, Object> updateData) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(final CompletableEmitter emitter) throws Exception {
         ref.updateChildren(updateData, new DatabaseReference.CompletionListener() {
            @Override
            public void onComplete(DatabaseError error, DatabaseReference databaseReference) {
               if (error != null) {
                  emitter.onError(new RxFirebaseDataException(error));
               } else {
                  emitter.onComplete();
               }
            }
         });
      }
   });
}

@Override
public void subscribe(final CompletableEmitter emitter) throws Exception {

    final OnCompleteListener<T> listener = new OnCompleteListener<T>() {
        @Override
        public void onComplete(@NonNull Task<T> task) {

            if (!emitter.isDisposed()) {
                if (!task.isSuccessful()) {
                    emitter.onError(task.getException());
                } else {
                    emitter.onComplete();
                }
            }

        }
    };

    database.runTransaction(transaction).addOnCompleteListener(listener);
}

@Override
public void subscribe(final CompletableEmitter emitter) throws Exception {
    final OnCompleteListener<Void> listener = new OnCompleteListener<Void>() {
        @Override
        public void onComplete(@NonNull Task<Void> task) {

            if (!emitter.isDisposed()) {
                if (!task.isSuccessful()) {
                    emitter.onError(task.getException());
                } else {
                    emitter.onComplete();
                }
            }

        }
    };

    batch.commit().addOnCompleteListener(listener);
}

@Override
public void subscribe(final CompletableEmitter emitter) throws Exception {
    final OnCompleteListener<Void> listener = new OnCompleteListener<Void>() {
        @Override
        public void onComplete(@NonNull Task<Void> task) {

            if (!emitter.isDisposed()) {
                if (!task.isSuccessful()) {
                    emitter.onError(task.getException());
                } else {
                    emitter.onComplete();
                }
            }

        }
    };

    reference.set(value).addOnCompleteListener(listener);

}

@Override
public void subscribe(final CompletableEmitter emitter) throws Exception {
    final OnCompleteListener<Void> listener = new OnCompleteListener<Void>() {
        @Override
        public void onComplete(@NonNull Task<Void> task) {

            if (!emitter.isDisposed()) {
                if (!task.isSuccessful()) {
                    emitter.onError(task.getException());
                } else {
                    emitter.onComplete();
                }
            }

        }
    };

    reference.delete().addOnCompleteListener(listener);
}

@Override
public void subscribe(final CompletableEmitter emitter) throws Exception {
    final OnCompleteListener<DocumentReference> listener =
            new OnCompleteListener<DocumentReference>() {
                @Override
                public void onComplete(@NonNull Task<DocumentReference> task) {

                    if (!emitter.isDisposed()) {
                        if (!task.isSuccessful()) {
                            emitter.onError(task.getException());
                        } else {
                            emitter.onComplete();
                        }
                    }

                }
            };

    reference.add(value).addOnCompleteListener(listener);
}

@Override
public void subscribe(final CompletableEmitter emitter) throws Exception {
    final OnCompleteListener<Void> listener = new OnCompleteListener<Void>() {
        @Override
        public void onComplete(@NonNull Task<Void> task) {

            if (!emitter.isDisposed()) {
                if (!task.isSuccessful()) {
                    emitter.onError(task.getException());
                } else {
                    emitter.onComplete();
                }
            }

        }
    };

    reference.update(updates).addOnCompleteListener(listener);

}

/**
 * @param emit
 * @param <R>
 * @return
 */
@NonNull
@CheckReturnValue
public static <R> OnCompleteListener<R> listener(@NonNull final CompletableEmitter emit) {
    return new OnCompleteListener<R>() {
        @Override
        public void onComplete(@NonNull final Task<R> task) {
            if (!emit.isDisposed()) {
                if (task.isSuccessful()) {
                    emit.onComplete();
                } else {
                    Exception e = task.getException();
                    emit.onError(e != null ? e : new RuntimeException());
                }
            }
        }
    };
}

public static OnCompleteListener<Void> listener(@NonNull final CompletableEmitter emitter) {
  return new OnCompleteListener<Void>() {
    @Override public void onComplete(@NonNull Task<Void> task) {
      if (!task.isSuccessful()) {
        if (!emitter.isDisposed()) {
          emitter.onError(task.getException());
        }
        return;
      }

      if (!emitter.isDisposed()) {
        emitter.onComplete();
      }
    }
  };
}

@Override public void subscribe(final CompletableEmitter emitter) {
  final Transaction.Handler handler = new Transaction.Handler() {
    @Override public Transaction.Result doTransaction(MutableData mutableData) {
      try {
        return task.apply(mutableData);
      } catch (Exception e) {
        //TODO: Is this enough?
        throw new RuntimeException(e);
      }
    }

    @Override public void onComplete(DatabaseError databaseError, boolean committed,
        DataSnapshot dataSnapshot) {
      if (!emitter.isDisposed()) {
        if (null == databaseError) {
          emitter.onComplete();
        } else {
          emitter.onError(databaseError.toException());
        }
      }
    }
  };

  ref.runTransaction(handler, fireLocalEvents);
}

/**
 * Given an operation that takes a {@link MaybeConsumer<Success>}, create a JavaRX
 * {@link Completable} that succeeds iff the operation does.
 *
 * Example:
 * <pre>
 *     // update the experiment, and then log that it was successful
 *     DataController dc = getDataController();
 *     MaybeConsumers.buildCompleteable(mc -> dc.updateExperiment(e.getExperimentId(), mc))
 *                   .subscribe(() -> log("successfully updated!"));
 * </pre>
 */
public static Completable buildCompleteable(
        io.reactivex.functions.Consumer<MaybeConsumer<Success>> c) {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(CompletableEmitter emitter) throws Exception {
            c.accept(new MaybeConsumer<Success>() {
                @Override
                public void success(Success value) {
                    emitter.onComplete();
                }

                @Override
                public void fail(Exception e) {
                    emitter.onError(e);
                }
            });
        }
    });
}

@Test
public void whenTheConstructorIsCalledWithAValidEmitterThenGetOnErrorReturnsTheEmitter() {

    //Given
    CompletableEmitter emitter = Mockito.mock(CompletableEmitter.class);
    Throwable ex = Mockito.mock(Throwable.class);
    CompletableEmitterMqttActionListener listener = new CompletableEmitterMqttActionListener(emitter) {

        @Override
        public void onSuccess(IMqttToken asyncActionToken) {
            // Not invoked
        }
    };

    // When
    OnError onError = listener.getOnError();
    onError.onError(ex);

    // Then
    Mockito.verify(emitter).onError(ex);
}

@Test public void verifyCancellation() {
  final AtomicInteger i = new AtomicInteger();
  //noinspection unchecked because Java
  Completable source = Completable.create(new CompletableOnSubscribe() {
    @Override public void subscribe(CompletableEmitter e) {
      e.setCancellable(new Cancellable() {
        @Override public void cancel() {
          i.incrementAndGet();
        }
      });
    }
  });
  MaybeSubject<Integer> lifecycle = MaybeSubject.create();
  source.as(autoDisposable(lifecycle))
      .subscribe();

  assertThat(i.get()).isEqualTo(0);
  assertThat(lifecycle.hasObservers()).isTrue();

  lifecycle.onSuccess(0);

  // Verify cancellation was called
  assertThat(i.get()).isEqualTo(1);
  assertThat(lifecycle.hasObservers()).isFalse();
}

/**
 * Delete File from internal Storage
 *
 * @param context  Context, not null
 * @param filename  Filename of the File that will be deleted
 * @return <code>Completable</code> that completes with <code>onComplete()</code> when the File is deleted.
 *         <code>onError()</code> is emitted if the given Context is null, filename is null or File is not deleted.
 */
public static Completable deleteInternal(@Nullable final Context context, final String filename) {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(CompletableEmitter emitter) throws Exception {
            if (context == null) {
                emitter.onError(new IllegalArgumentException("Context must not be null"));
                return;
            }
            if (filename == null) {
                emitter.onError(new IllegalArgumentException("Filename must not be null"));
                return;
            }

            File directory = context.getFilesDir();
            File file = new File(directory, filename);
            boolean deleted = file.delete();

            if (deleted) {
                emitter.onComplete();
            } else {
                emitter.onError(new FileNotFoundException());
            }
        }
    });
}

/**
 * Converts an {@link ApolloPrefetch} to a synchronous Completable
 *
 * @param prefetch the ApolloPrefetch to convert
 * @return the converted Completable
 * @throws NullPointerException if prefetch == null
 */
@Nonnull public static Completable from(@Nonnull final ApolloPrefetch prefetch) {
  checkNotNull(prefetch, "prefetch == null");

  return Completable.create(new CompletableOnSubscribe() {
    @Override public void subscribe(final CompletableEmitter emitter) {
      cancelOnCompletableDisposed(emitter, prefetch);
      prefetch.enqueue(new ApolloPrefetch.Callback() {
        @Override public void onSuccess() {
          if (!emitter.isDisposed()) {
            emitter.onComplete();
          }
        }

        @Override public void onFailure(@Nonnull ApolloException e) {
          Exceptions.throwIfFatal(e);
          if (!emitter.isDisposed()) {
            emitter.onError(e);
          }
        }
      });
    }
  });
}

/**
 * Set the given value on the specified {@link DatabaseReference}.
 *
 * @param ref   reference represents a particular location in your database.
 * @param value value to update.
 * @return a {@link Completable} which is complete when the set value call finish successfully.
 */
@NonNull
public static Completable setValue(@NonNull final DatabaseReference ref,
                                   final Object value) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(@NonNull final CompletableEmitter e) throws Exception {
         ref.setValue(value).addOnSuccessListener(new OnSuccessListener<Void>() {
            @Override public void onSuccess(Void aVoid) {
               e.onComplete();
            }
         }).addOnFailureListener(new OnFailureListener() {
            @Override public void onFailure(@NonNull Exception exception) {
               e.onError(exception);
            }
         });
      }
   });
}

/**
 * Update the specific child keys to the specified values.
 *
 * @param ref        reference represents a particular location in your database.
 * @param updateData The paths to update and their new values
 * @return a {@link Completable} which is complete when the update children call finish successfully.
 */
@NonNull
public static Completable updateChildren(@NonNull final DatabaseReference ref,
                                         @NonNull final Map<String, Object> updateData) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(final CompletableEmitter emitter) throws Exception {
         ref.updateChildren(updateData, new DatabaseReference.CompletionListener() {
            @Override
            public void onComplete(DatabaseError error, DatabaseReference databaseReference) {
               if (error != null) {
                  emitter.onError(new RxFirebaseDataException(error));
               } else {
                  emitter.onComplete();
               }
            }
         });
      }
   });
}

@Override @NonNull public Completable observeClear() {
  return Completable.create(new CompletableOnSubscribe() {
    @Override public void subscribe(final CompletableEmitter emitter) throws Exception {
      runInWriteLock(readWriteLock, new ThrowingRunnable() {
        @Override public void run() throws Exception {
          if (file.exists() && !file.delete()) {
            throw new IOException("Clear operation on store failed.");
          } else {
            emitter.onComplete();
          }

          updateSubject.onNext(ValueUpdate.<T>empty());
        }
      });
    }
  });
}

public static Completable refreshToken(final Credentials credentials) {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(CompletableEmitter e) throws Exception {
            try {
                AuthenticationManager.get().refreshAccessToken(credentials);
                e.onComplete();
            } catch (Exception ex) {
                e.onError(ex);
            }
        }
    });
}

@Override
public Completable createProfile(final Profile profile) {
    return Completable.create(
            new CompletableOnSubscribe() {
                @Override
                public void subscribe(final CompletableEmitter e) throws Exception {
                    final DatabaseReference rootRef = FirebaseDatabase.getInstance().getReference();
                    final DatabaseReference idRef = rootRef.child(USER_PROFILES).child(profile.getUid());
                    idRef.addListenerForSingleValueEvent(new ValueEventListener() {
                        @Override
                        public void onDataChange(DataSnapshot snapshot) {
                            if (!snapshot.exists()) {
                                idRef.setValue(profile).addOnCompleteListener(new OnCompleteListener<Void>() {
                                    @Override
                                    public void onComplete(@NonNull Task<Void> task) {
                                        if (task.isSuccessful()) {
                                            e.onComplete();
                                        } else {
                                            e.onError(task.getException());
                                        }
                                    }
                                });
                            } else {
                                e.onComplete();
                            }
                        }

                        @Override
                        public void onCancelled(DatabaseError databaseError) {
                            Log.d("FIREBASE", databaseError.toString());
                        }
                    });
                }
            }
    );
}

@Override
public Completable logUserOut() {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(final CompletableEmitter e) throws Exception {
            if (auth == null) {
                auth = FirebaseAuth.getInstance();
            }

            listener = new FirebaseAuth.AuthStateListener() {
                @Override
                public void onAuthStateChanged(@NonNull FirebaseAuth firebaseAuth) {
                    auth.removeAuthStateListener(listener);
                    if (firebaseAuth.getCurrentUser() == null) {
                        e.onComplete();
                    } else {
                        e.onError(new Exception());
                    }
                }
            };

            auth.addAuthStateListener(listener);
            auth.signOut();
        }
    });


}

@Override
public Completable reauthenticateUser(final String password) {
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(final CompletableEmitter e) throws Exception {
            if (auth == null) {
                auth = FirebaseAuth.getInstance();
            }

            final FirebaseUser user = auth.getCurrentUser();
            AuthCredential credential = EmailAuthProvider
                    .getCredential(user.getEmail(), password);

            user.reauthenticate(credential)
                    .addOnCompleteListener(new OnCompleteListener<Void>() {
                        @Override
                        public void onComplete(@NonNull Task<Void> task) {
                            if (task.isSuccessful()) {
                                e.onComplete();
                            } else {
                                e.onError(task.getException());
                            }
                        }
                    });
        }
    });
}

@Override
public void subscribe(CompletableEmitter emitter) throws Exception {
    initGoogleApiClient();
    this.emitter = emitter;
    emitter.setCancellable(() -> {
        AndroidSchedulers.mainThread().scheduleDirect(() -> {
            disconnect();
            activity = null;
        });
    });
}

/**
 * Updates the email address of the user.
 *
 * @param firebaseUser current firebaseUser instance.
 * @param email        new email.
 * @return a {@link Completable} if the task is complete successfully.
 */
@NonNull
public static Completable updateEmail(@NonNull final FirebaseUser firebaseUser,
                                      @NonNull final String email) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(CompletableEmitter emitter) throws Exception {
         RxCompletableHandler.assignOnTask(emitter, firebaseUser.updateEmail(email));
      }
   });
}

/**
 * Updates the password of the user.
 *
 * @param firebaseUser current firebaseUser instance.
 * @param password     new password.
 * @return a {@link Completable} if the task is complete successfully.
 */
@NonNull
public static Completable updatePassword(@NonNull final FirebaseUser firebaseUser,
                                         @NonNull final String password) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(CompletableEmitter emitter) throws Exception {
         RxCompletableHandler.assignOnTask(emitter, firebaseUser.updatePassword(password));
      }
   });
}

/**
 * Updates the user profile information.
 *
 * @param firebaseUser current firebaseUser instance.
 * @param request      {@link UserProfileChangeRequest} request for this user.
 * @return a {@link Completable} if the task is complete successfully.
 */
@NonNull
public static Completable updateProfile(@NonNull final FirebaseUser firebaseUser,
                                        @NonNull final UserProfileChangeRequest request) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(CompletableEmitter emitter) throws Exception {
         RxCompletableHandler.assignOnTask(emitter, firebaseUser.updateProfile(request));
      }
   });
}

/**
 * Deletes the user record from your Firebase project's database.
 *
 * @param firebaseUser current firebaseUser instance.
 * @return a {@link Completable} if the task is complete successfully.
 */
@NonNull
public static Completable delete(@NonNull final FirebaseUser firebaseUser) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(CompletableEmitter emitter) throws Exception {
         RxCompletableHandler.assignOnTask(emitter, firebaseUser.delete());
      }
   });
}

/**
 * Reauthenticates the user with the given credential.
 *
 * @param firebaseUser current firebaseUser instance.
 * @param credential   {@link AuthCredential} to re-authenticate.
 * @return a {@link Completable} if the task is complete successfully.
 */
@NonNull
public static Completable reAuthenticate(@NonNull final FirebaseUser firebaseUser,
                                         @NonNull final AuthCredential credential) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(CompletableEmitter emitter) throws Exception {
         RxCompletableHandler.assignOnTask(emitter, firebaseUser.reauthenticate(credential));
      }
   });
}

/**
 * Manually refreshes the data of the current user (for example, attached providers, display name, and so on).
 *
 * @param firebaseUser current firebaseUser instance.
 * @return a {@link Completable} if the task is complete successfully.
 */
@NonNull
public static Completable reload(@NonNull final FirebaseUser firebaseUser) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(CompletableEmitter emitter) throws Exception {
         RxCompletableHandler.assignOnTask(emitter, firebaseUser.reload());
      }
   });
}

/**
 * Initiates email verification for the user.
 *
 * @param firebaseUser current firebaseUser instance.
 * @return a {@link Completable} if the task is complete successfully.
 */
@NonNull
public static Completable sendEmailVerification(@NonNull final FirebaseUser firebaseUser) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(CompletableEmitter emitter) throws Exception {
         RxCompletableHandler.assignOnTask(emitter, firebaseUser.sendEmailVerification());
      }
   });
}

/**
 * Changes the user's password to newPassword for the account for which the code is valid.
 * Code validity can be checked with {@link FirebaseAuth#verifyPasswordResetCode(String)}.
 * This use case is only valid for signed-out users, and behavior is undefined for signed-in users.
 * Password changes for signed-in users should be made using {@link com.google.firebase.auth.FirebaseUser#updatePassword(String)}.
 *
 * @param firebaseAuth firebaseAuth instance.
 * @param code         generated code by firebase.
 * @param newPassword  new password for the user.
 * @return a {@link Completable} which emits when the action is completed.
 */
@NonNull
public static Completable confirmPasswordReset(@NonNull final FirebaseAuth firebaseAuth,
                                               @NonNull final String code,
                                               @NonNull final String newPassword) {
   return Completable.create(new CompletableOnSubscribe() {
      @Override
      public void subscribe(CompletableEmitter emitter) throws Exception {
         RxCompletableHandler.assignOnTask(emitter, firebaseAuth.confirmPasswordReset(code, newPassword));
      }
   });
}