Improving the ThreadManager and TokenRefresher APIs

src/main/java/com/google/firebase/FirebaseApp.java

@@ -33,11 +33,14 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.io.BaseEncoding;
 import com.google.firebase.internal.FirebaseAppStore;
-import com.google.firebase.internal.FirebaseExecutors;
 import com.google.firebase.internal.FirebaseService;
+import com.google.firebase.internal.GaeThreadFactory;
 import com.google.firebase.internal.NonNull;
 import com.google.firebase.internal.Nullable;
 
+import com.google.firebase.internal.RevivingScheduledExecutor;
+import com.google.firebase.tasks.Task;
+import com.google.firebase.tasks.Tasks;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -47,9 +50,14 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -84,10 +92,14 @@ public class FirebaseApp {
  private final String name;
  private final FirebaseOptions options;
  private final TokenRefresher tokenRefresher;
+ private final ThreadManager threadManager;
+ private final ThreadManager.FirebaseExecutor executor;
 
  private final AtomicBoolean deleted = new AtomicBoolean();
  private final Map<String, FirebaseService> services = new HashMap<>();
 
+ private volatile ScheduledExecutorService scheduledExecutor;
+
  /**
  * Per application lock for synchronizing all internal FirebaseApp state changes.
  */
@@ -99,6 +111,8 @@ private FirebaseApp(String name, FirebaseOptions options, TokenRefresher.Factory
  this.name = name;
  this.options = checkNotNull(options);
  this.tokenRefresher = checkNotNull(factory).create(this);
+ this.threadManager = options.getThreadManager();
+ this.executor = this.threadManager.getFirebaseExecutor(this);
  }
 
  /** Returns a list of all FirebaseApps. */
@@ -242,7 +256,6 @@ private static String normalize(@NonNull String name) {
  /** Returns the unique name of this app. */
  @NonNull
  public String getName() {
- checkNotDeleted();
  return name;
  }
 
@@ -315,7 +328,14 @@ public void delete() {
  service.destroy();
  }
  services.clear();
- tokenRefresher.cleanup();
+ tokenRefresher.stop();
+
+ // Clean up and terminate the thread pools
+ threadManager.releaseFirebaseExecutor(this, executor);
+ if (scheduledExecutor != null) {
+ scheduledExecutor.shutdownNow();
+ scheduledExecutor = null;
+ }
  }
 
  synchronized (appsLock) {
@@ -332,6 +352,47 @@ private void checkNotDeleted() {
  checkState(!deleted.get(), "FirebaseApp was deleted %s", this);
  }
 
+ private ScheduledExecutorService ensureScheduledExecutorService() {
+ if (scheduledExecutor == null) {
+ synchronized (lock) {
+ checkNotDeleted();
+ if (scheduledExecutor == null) {
+ scheduledExecutor = new RevivingScheduledExecutor(threadManager.getThreadFactory(),
+ "firebase-scheduled-worker", GaeThreadFactory.isAvailable());
+ }
+ }
+ }
+ return scheduledExecutor;
+ }
+
+ ThreadFactory getThreadFactory() {
+ return threadManager.getThreadFactory();
+ }
+
+ // TODO: Return an ApiFuture once Task API is fully removed.
+ <T> Task<T> submit(Callable<T> command) {
+ checkNotNull(command);
+ return Tasks.call(executor.getListeningExecutor(), command);
+ }
+
+ <T> ScheduledFuture<T> schedule(Callable<T> command, long delayMillis) {
+ checkNotNull(command);
+ try {
+ return ensureScheduledExecutorService().schedule(command, delayMillis, TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
+ // This may fail if the underlying ThreadFactory does not support long-lived threads.
+ throw new UnsupportedOperationException("Scheduled tasks not supported", e);
+ }
+ }
+
+ void startTokenRefresher() {
+ synchronized (lock) {
+ checkNotDeleted();
+ // TODO: Provide an option to disable this altogether.
+ tokenRefresher.start();
+ }
+ }
+
  boolean isDefaultApp() {
  return DEFAULT_APP_NAME.equals(getName());
  }
@@ -362,24 +423,30 @@ FirebaseService getService(String id) {
  */
  static class TokenRefresher implements CredentialsChangedListener {
 
+ private static final int STATE_READY = 0;
+ private static final int STATE_STARTED = 1;
+ private static final int STATE_STOPPED = 2;
+
+ private final FirebaseApp firebaseApp;
  private final GoogleCredentials credentials;
- private ScheduledFuture<Void> future;
- private boolean closed;
+ private final AtomicInteger state;
+
+ private Future<Void> future;
 
- TokenRefresher(FirebaseApp app) {
- this.credentials = app.getOptions().getCredentials();
- this.credentials.addChangeListener(this);
+ TokenRefresher(FirebaseApp firebaseApp) {
+ this.firebaseApp = checkNotNull(firebaseApp);
+ this.credentials = firebaseApp.getOptions().getCredentials();
+ this.state = new AtomicInteger(STATE_READY);
  }
 
  @Override
  public final synchronized void onChanged(OAuth2Credentials credentials) throws IOException {
- if (closed) {
+ if (state.get() != STATE_STARTED) {
  return;
  }
 
  AccessToken accessToken = credentials.getAccessToken();
- long refreshDelay = accessToken.getExpirationTime().getTime()
- - System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(5);
+ long refreshDelay = getRefreshDelay(accessToken);
  if (refreshDelay > 0) {
  scheduleRefresh(refreshDelay);
  } else {
@@ -388,26 +455,6 @@ public final synchronized void onChanged(OAuth2Credentials credentials) throws I
  }
  }
 
- /**
- * Schedule a forced token refresh to be executed after a specified duration.
- *
- * @param delayMillis Duration in milliseconds, after which the token should be forcibly
- * refreshed.
- */
- private void scheduleRefresh(final long delayMillis) {
- cancelPrevious();
- scheduleNext(
- new Callable<Void>() {
- @Override
- public Void call() throws Exception {
- logger.debug("Refreshing OAuth2 credential");
- credentials.refresh();
- return null;
- }
- },
- delayMillis);
- }
-
  protected void cancelPrevious() {
  if (future != null) {
  future.cancel(true);
@@ -417,17 +464,71 @@ protected void cancelPrevious() {
  protected void scheduleNext(Callable<Void> task, long delayMillis) {
  logger.debug("Scheduling next token refresh in {} milliseconds", delayMillis);
  try {
- future =
- FirebaseExecutors.DEFAULT_SCHEDULED_EXECUTOR.schedule(
- task, delayMillis, TimeUnit.MILLISECONDS);
- } catch (UnsupportedOperationException ignored) {
+ future = firebaseApp.schedule(task, delayMillis);
+ } catch (UnsupportedOperationException e) {
  // Cannot support task scheduling in the current runtime.
+ logger.debug("Failed to schedule token refresh event", e);
  }
  }
 
- protected synchronized void cleanup() {
+ /**
+ * Starts the TokenRefresher if not already started. Starts listening to credentials changed
+ * events, and schedules refresh events every time the OAuth2 token changes. If no active
+ * token is present, or if the available token is set to expire soon, this will also schedule
+ * a refresh event to be executed immediately.
+ *
+ * <p>This operation is idempotent. Calling it multiple times, or calling it after the
+ * refresher has been stopped has no effect.
+ */
+ final synchronized void start() {
+ // Allow starting only from the ready state.
+ if (!state.compareAndSet(STATE_READY, STATE_STARTED)) {
+ return;
+ }
+
+ logger.debug("Starting the proactive token refresher");
+ credentials.addChangeListener(this);
+ AccessToken accessToken = credentials.getAccessToken();
+ long refreshDelay = 0L;
+ if (accessToken != null) {
+ refreshDelay = Math.max(getRefreshDelay(accessToken), refreshDelay);
+ }
+ // If the access token is null, or is about to expire (i.e. expires in less than 5 minutes),
+ // schedule a refresh event with 0 delay. Otherwise schedule a refresh event at the token
+ // expiry time, minus 5 minutes.
+ scheduleRefresh(refreshDelay);
+ }
+
+ final synchronized void stop() {
+ // Allow stopping from any state.
+ int previous = state.getAndSet(STATE_STOPPED);
+ if (previous == STATE_STARTED) {
+ cancelPrevious();
+ logger.debug("Stopped the proactive token refresher");
+ }
+ }
+
+ /**
+ * Schedule a forced token refresh to be executed after a specified duration.
+ *
+ * @param delayMillis Duration in milliseconds, after which the token should be forcibly
+ * refreshed.
+ */
+ private void scheduleRefresh(final long delayMillis) {
  cancelPrevious();
- closed = true;
+ scheduleNext(new Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ logger.debug("Refreshing OAuth2 credential");
+ credentials.refresh();
+ return null;
+ }
+ }, delayMillis);
+ }
+
+ private long getRefreshDelay(AccessToken accessToken) {
+ return accessToken.getExpirationTime().getTime() - System.currentTimeMillis()
+ - TimeUnit.MINUTES.toMillis(5);
  }
 
  static class Factory {

src/main/java/com/google/firebase/FirebaseOptions.java

@@ -28,6 +28,7 @@
 import com.google.firebase.auth.FirebaseCredentials;
 import com.google.firebase.auth.internal.BaseCredential;
 import com.google.firebase.auth.internal.FirebaseCredentialsAdapter;
+import com.google.firebase.internal.FirebaseExecutors;
 import com.google.firebase.internal.NonNull;
 import com.google.firebase.internal.Nullable;
 
@@ -46,6 +47,7 @@ public final class FirebaseOptions {
  private final String projectId;
  private final HttpTransport httpTransport;
  private final JsonFactory jsonFactory;
+ private final ThreadManager threadManager;
 
  private FirebaseOptions(@NonNull FirebaseOptions.Builder builder) {
  this.credentials = checkNotNull(builder.credentials,
@@ -59,6 +61,8 @@ private FirebaseOptions(@NonNull FirebaseOptions.Builder builder) {
  "FirebaseOptions must be initialized with a non-null HttpTransport.");
  this.jsonFactory = checkNotNull(builder.jsonFactory,
  "FirebaseOptions must be initialized with a non-null JsonFactory.");
+ this.threadManager = checkNotNull(builder.threadManager,
+ "FirebaseOptions must be initialized with a non-null ThreadManager");
  }
 
  /**
@@ -118,7 +122,12 @@ public JsonFactory getJsonFactory() {
  return jsonFactory;
  }
 
- /** 
+ @NonNull
+ ThreadManager getThreadManager() {
+ return threadManager;
+ }
+
+ /**
  * Builder for constructing {@link FirebaseOptions}. 
  */
  public static final class Builder {
@@ -130,6 +139,7 @@ public static final class Builder {
  private String projectId;
  private HttpTransport httpTransport = Utils.getDefaultTransport();
  private JsonFactory jsonFactory = Utils.getDefaultJsonFactory();
+ private ThreadManager threadManager = FirebaseExecutors.DEFAULT_THREAD_MANAGER;
 
  /** Constructs an empty builder. */
  public Builder() {}
@@ -148,6 +158,7 @@ public Builder(FirebaseOptions options) {
  projectId = options.projectId;
  httpTransport = options.httpTransport;
  jsonFactory = options.jsonFactory;
+ threadManager = options.threadManager;
  }
 
  /**
@@ -263,6 +274,18 @@ public Builder setJsonFactory(JsonFactory jsonFactory) {
  return this;
  }
 
+ /**
+ * Sets the <code>ThreadManager</code> used to initialize thread pools and thread factories
+ * for Firebase apps.
+ *
+ * @param threadManager A <code>ThreadManager</code> instance.
+ * @return This <code>Builder</code> instance is returned so subsequent calls can be chained.
+ */
+ public Builder setThreadManager(ThreadManager threadManager) {
+ this.threadManager = threadManager;
+ return this;
+ }
+
  /**
  * Builds the {@link FirebaseOptions} instance from the previously set options.
  *

src/main/java/com/google/firebase/ImplFirebaseTrampolines.java

@@ -20,6 +20,10 @@
 import com.google.firebase.internal.FirebaseService;
 import com.google.firebase.internal.NonNull;
 
+import com.google.firebase.tasks.Task;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ThreadFactory;
+
 /**
  * Provides trampolines into package-private APIs used by components of Firebase. Intentionally
  * scarily-named to dissuade people from actually trying to use the class and to make it less likely
@@ -61,4 +65,16 @@ public static <T extends FirebaseService> T addService(
  app.addService(service);
  return service;
  }
+
+ public static ThreadFactory getThreadFactory(@NonNull FirebaseApp app) {
+ return app.getThreadFactory();
+ }
+
+ public static <T> Task<T> submitCallable(@NonNull FirebaseApp app, @NonNull Callable<T> command) {
+ return app.submit(command);
+ }
+
+ public static void startTokenRefresher(@NonNull FirebaseApp app) {
+ app.startTokenRefresher();
+ }
 }