Java 类io.reactivex.MaybeEmitter 实例源码

@Override
public final void subscribe(MaybeEmitter<T> emitter) throws Exception {
    final GoogleApiClient apiClient = createApiClient(new ApiClientConnectionCallbacks(emitter));

    try {
        apiClient.connect();
    } catch (Throwable ex) {
        emitter.onError(ex);
    }

    emitter.setCancellable(new Cancellable() {
        @Override
        public void cancel() throws Exception {
            if (apiClient.isConnected()) {
                RxLocationMaybeOnSubscribe.this.onUnsubscribed(apiClient);
            }

            apiClient.disconnect();
        }
    });
}

/**
 * Listener for a single change in te data at the given query location.
 *
 * @param query reference represents a particular location in your Database and can be used for reading or writing data to that Database location.
 * @return a {@link Maybe} which emits the actual state of the database for the given query.
 */
@NonNull
public static Maybe<DataSnapshot> observeSingleValueEvent(@NonNull final Query query) {
   return Maybe.create(new MaybeOnSubscribe<DataSnapshot>() {
      @Override
      public void subscribe(final MaybeEmitter<DataSnapshot> emitter) throws Exception {
         query.addListenerForSingleValueEvent(new ValueEventListener() {
            @Override
            public void onDataChange(DataSnapshot dataSnapshot) {
               emitter.onSuccess(dataSnapshot);
               emitter.onComplete();
            }

            @Override
            public void onCancelled(DatabaseError error) {
               emitter.onError(new RxFirebaseDataException(error));
            }
         });
      }
   });
}

@Override
public Maybe<Response> download(final Request request) {

    return Maybe.create(new MaybeOnSubscribe<CloseableHttpResponse>(){

        @Override
        public void subscribe(MaybeEmitter emitter) throws Exception {

            emitter.onSuccess(httpManager.getResponse(request));
        }
    }).map(new Function<CloseableHttpResponse, Response>() {

        @Override
        public Response apply(CloseableHttpResponse closeableHttpResponse) throws Exception {

            String html = EntityUtils.toString(closeableHttpResponse.getEntity(), "UTF-8");
            Response response = new Response();
            response.setContent(html);
            response.setStatusCode(closeableHttpResponse.getStatusLine().getStatusCode());
            return response;
        }
    });
}

@Override
public final void subscribe(MaybeEmitter<T> emitter) throws Exception {
    final GoogleApiClient apiClient = createApiClient(new ApiClientConnectionCallbacks(emitter));

    try {
        apiClient.connect();
    } catch (Throwable ex) {
        emitter.onError(ex);
    }

    emitter.setCancellable(() -> {
        if (apiClient.isConnected()) {
            onUnsubscribed(apiClient);
        }

        apiClient.disconnect();
    });
}

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

@Test public void verifyCancellation() {
  final AtomicInteger i = new AtomicInteger();
  //noinspection unchecked because Java
  Maybe<Integer> source = Maybe.create(new MaybeOnSubscribe<Integer>() {
    @Override public void subscribe(MaybeEmitter<Integer> e) {
      e.setCancellable(new Cancellable() {
        @Override public void cancel() {
          i.incrementAndGet();
        }
      });
    }
  });
  MaybeSubject<Integer> lifecycle = MaybeSubject.create();
  source.as(AutoDispose.<Integer>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();
}

/**
 * Listener for a single change in te data at the given query location.
 *
 * @param query reference represents a particular location in your Database and can be used for reading or writing data to that Database location.
 * @return a {@link Maybe} which emits the actual state of the database for the given query. onSuccess will be only call when
 * the given {@link DataSnapshot} exists onComplete will only called when the data doesn't exist.
 */
@NonNull
public static Maybe<DataSnapshot> observeSingleValueEvent(@NonNull final Query query) {
   return Maybe.create(new MaybeOnSubscribe<DataSnapshot>() {
      @Override
      public void subscribe(final MaybeEmitter<DataSnapshot> emitter) throws Exception {
         query.addListenerForSingleValueEvent(new ValueEventListener() {
            @Override
            public void onDataChange(DataSnapshot dataSnapshot) {
               if (dataSnapshot.exists()) {
                  emitter.onSuccess(dataSnapshot);
               }
               else {
                 emitter.onComplete();
               }
            }

            @Override
            public void onCancelled(DatabaseError error) {
               emitter.onError(new RxFirebaseDataException(error));
            }
         });
      }
   });
}

@Override @NonNull public Maybe<T> get() {
  return Maybe.create(new MaybeOnSubscribe<T>() {
    @Override public void subscribe(final MaybeEmitter<T> emitter) throws Exception {
      runInReadLock(readWriteLock, new ThrowingRunnable() {
        @Override public void run() throws Exception {
          if (!file.exists()) {
            emitter.onComplete();
            return;
          }

          T value = converter.read(file, type);
          if (value == null) emitter.onComplete();
          emitter.onSuccess(value);
        }
      });
    }
  });
}

@Override
protected void onGoogleApiClientReady(GoogleApiClient apiClient, MaybeEmitter<Location> emitter) {
    //noinspection MissingPermission
    Location location = LocationServices.FusedLocationApi.getLastLocation(apiClient);

    if (location != null) {
        emitter.onSuccess(location);
    } else {
        emitter.onComplete();
    }
}

@Override
public Maybe<Profile> getProfile(final String uid) {
    return Maybe.create(
            new MaybeOnSubscribe<Profile>() {
                @Override
                public void subscribe(final MaybeEmitter e) throws Exception {
                    final DatabaseReference rootRef = FirebaseDatabase.getInstance().getReference();
                    DatabaseReference idRef = rootRef.child(USER_PROFILES).child(uid);
                    idRef.addListenerForSingleValueEvent(new ValueEventListener() {
                        //does this check node for activeUser exists?
                        @Override
                        public void onDataChange(DataSnapshot snapshot) {
                            if (snapshot.exists()) {
                                //setUpProfilePageComponent(
                                Profile profile = snapshot.getValue(Profile.class);
                                e.onSuccess(profile);
                            } else {
                                e.onComplete();
                            }
                        }

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

@Override
public Maybe<List<Photo>> getThumbnails(final ContentResolver resolver) {
    return Maybe.create(
            new MaybeOnSubscribe<List<Photo>>() {
                @Override
                public void subscribe(final MaybeEmitter<List<Photo>> e) throws Exception {
                    Uri uri;
                    List<Photo> listOfAllPhotos = new ArrayList<Photo>();
                    Cursor cursor;
                    int column_index_data;
                    Uri imageUri;
                    uri = MediaStore.Images.Media.EXTERNAL_CONTENT_URI;

                    String[] projection = {MediaStore.MediaColumns.DATA};

                    cursor = resolver.query(uri, projection, null,
                            null, null);
                    try {
                        column_index_data = cursor.getColumnIndexOrThrow(MediaStore.MediaColumns.DATA);
                        while (cursor.moveToNext()) {
                            imageUri = Uri.parse("file://" + cursor.getString(column_index_data));
                            listOfAllPhotos.add(new Photo(imageUri.toString()));
                        }
                        cursor.close();
                    } catch (Throwable t) {
                        if (cursor != null) {
                            cursor.close();
                        }
                        e.onError(t);
                    }

                    if (listOfAllPhotos.size() == 0) {
                        e.onComplete();
                    } else {
                        e.onSuccess(listOfAllPhotos);
                    }
                }
            }
    );
}

@Override
public Maybe<User> getUser() {
    return Maybe.create(
            new MaybeOnSubscribe<User>() {
                @Override
                public void subscribe(final MaybeEmitter<User> e) throws Exception {
                    if (auth == null) {
                        auth = FirebaseAuth.getInstance();
                    }

                    if (listener != null) {
                        auth.removeAuthStateListener(listener);
                    }

                    listener = new FirebaseAuth.AuthStateListener() {
                        @Override
                        public void onAuthStateChanged(@NonNull FirebaseAuth firebaseAuth) {
                            FirebaseUser firebaseUser = firebaseAuth.getCurrentUser();
                            auth.removeAuthStateListener(listener);
                            if (firebaseUser != null) {
                                User user = new User(
                                        firebaseUser.getEmail(),
                                        firebaseUser.getUid()
                                );
                                Maybe.just(user);
                                e.onSuccess(user);
                            } else {
                                e.onComplete();
                            }
                        }
                    };

                    auth.addAuthStateListener(listener);
                }
            }
    );

}

/**
 * Fetches a Firebase Auth ID Token for the user; useful when authenticating against your own backend.
 *
 * @param firebaseUser current firebaseUser instance.
 * @param forceRefresh force to refresh the token ID.
 * @return a {@link Maybe} which emits an {@link GetTokenResult} if success.
 */
@NonNull
public static Maybe<GetTokenResult> getToken(@NonNull final FirebaseUser firebaseUser,
                                             final boolean forceRefresh) {
   return Maybe.create(new MaybeOnSubscribe<GetTokenResult>() {
      @Override
      public void subscribe(MaybeEmitter<GetTokenResult> emitter) throws Exception {
         RxHandler.assignOnTask(emitter, firebaseUser.getToken(forceRefresh));
      }
   });
}

/**
 * Attaches the given {@link AuthCredential} to the user.
 *
 * @param firebaseUser current firebaseUser instance.
 * @param credential   new {@link AuthCredential} to link.
 * @return a {@link Maybe} which emits an {@link AuthResult} if success.
 */
@NonNull
public static Maybe<AuthResult> linkWithCredential(@NonNull final FirebaseUser firebaseUser,
                                                   @NonNull final AuthCredential credential) {
   return Maybe.create(new MaybeOnSubscribe<AuthResult>() {
      @Override public void subscribe(MaybeEmitter<AuthResult> emitter) throws Exception {
         RxHandler.assignOnTask(emitter, firebaseUser.linkWithCredential(credential));
      }
   });
}

/**
 * Detaches credentials from a given provider type from this user.
 *
 * @param firebaseUser current firebaseUser instance.
 * @param provider     a unique identifier of the type of provider to be unlinked, for example, {@link com.google.firebase.auth.FacebookAuthProvider#PROVIDER_ID}.
 * @return a {@link Maybe} which emits an {@link AuthResult} if success.
 */
@NonNull
public static Maybe<AuthResult> unlink(@NonNull final FirebaseUser firebaseUser,
                                       @NonNull final String provider) {
   return Maybe.create(new MaybeOnSubscribe<AuthResult>() {
      @Override public void subscribe(MaybeEmitter<AuthResult> emitter) throws Exception {
         RxHandler.assignOnTask(emitter, firebaseUser.unlink(provider));
      }
   });
}

/**
 * Asynchronously signs in as an anonymous user.
 * If there is already an anonymous user signed in, that user will be returned; otherwise, a new anonymous user identity will be created and returned.
 *
 * @param firebaseAuth firebaseAuth instance.
 * @return a {@link Maybe} which emits an {@link AuthResult} if success.
 * @see <a href="https://firebase.google.com/docs/reference/android/com/google/firebase/auth/FirebaseAuth">Firebase Auth API</a>
 */
@NonNull
public static Maybe<AuthResult> signInAnonymously(@NonNull final FirebaseAuth firebaseAuth) {
   return Maybe.create(new MaybeOnSubscribe<AuthResult>() {
      @Override
      public void subscribe(MaybeEmitter<AuthResult> emitter) throws Exception {
         RxHandler.assignOnTask(emitter, firebaseAuth.signInAnonymously());
      }
   });
}

/**
 * Asynchronously signs in with the given credentials.
 *
 * @param firebaseAuth firebaseAuth instance.
 * @param credential   The auth credential. Value must not be null.
 * @return a {@link Maybe} which emits an {@link AuthResult} if success.
 * @see <a href="https://firebase.google.com/docs/reference/android/com/google/firebase/auth/FirebaseAuth">Firebase Auth API</a>
 */
@NonNull
public static Maybe<AuthResult> signInWithCredential(@NonNull final FirebaseAuth firebaseAuth,
                                                     @NonNull final AuthCredential credential) {
   return Maybe.create(new MaybeOnSubscribe<AuthResult>() {
      @Override
      public void subscribe(MaybeEmitter<AuthResult> emitter) throws Exception {
         RxHandler.assignOnTask(emitter, firebaseAuth.signInWithCredential(credential));
      }
   });
}

/**
 * Gets the list of provider IDs that can be used to sign in for the given email address. Useful for an "identifier-first" sign-in flow.
 *
 * @param firebaseAuth firebaseAuth instance.
 * @param email        An email address.
 * @return a {@link Maybe} which emits an {@link ProviderQueryResult} if success.
 * @see <a href="https://firebase.google.com/docs/reference/android/com/google/firebase/auth/FirebaseAuth">Firebase Auth API</a>
 */
@NonNull
public static Maybe<ProviderQueryResult> fetchProvidersForEmail(@NonNull final FirebaseAuth firebaseAuth,
                                                                @NonNull final String email) {
   return Maybe.create(new MaybeOnSubscribe<ProviderQueryResult>() {
      @Override
      public void subscribe(MaybeEmitter<ProviderQueryResult> emitter) throws Exception {
         RxHandler.assignOnTask(emitter, firebaseAuth.fetchProvidersForEmail(email));
      }
   });
}

/**
 * Checks that the code given is valid. This code will have been generated
 * by {@link FirebaseAuth#sendPasswordResetEmail(String)} or {@link com.google.firebase.auth.FirebaseUser#sendEmailVerification()} valid for a single use.
 *
 * @param firebaseAuth firebaseAuth instance.
 * @param code         generated code by firebase.
 * @return a {@link Maybe} which emits when the action is completed.
 */
@NonNull
public static Maybe<ActionCodeResult> checkActionCode(@NonNull final FirebaseAuth firebaseAuth,
                                                      @NonNull final String code) {
   return Maybe.create(new MaybeOnSubscribe<ActionCodeResult>() {
      @Override
      public void subscribe(MaybeEmitter<ActionCodeResult> emitter) throws Exception {
         RxHandler.assignOnTask(emitter, firebaseAuth.checkActionCode(code));
      }
   });
}

/**
 * Checks that the code is a valid password reset out of band code.
 * This code will have been generated by a call to {@link FirebaseAuth#sendPasswordResetEmail(String)}, and is valid for a single use.
 *
 * @param firebaseAuth firebaseAuth instance.
 * @param code         generated code by firebase.
 * @return a {@link Maybe} which emits when the action is completed.
 */
@NonNull
public static Maybe<String> verifyPasswordResetCode(@NonNull final FirebaseAuth firebaseAuth,
                                                    @NonNull final String code) {
   return Maybe.create(new MaybeOnSubscribe<String>() {
      @Override
      public void subscribe(MaybeEmitter<String> emitter) throws Exception {
         RxHandler.assignOnTask(emitter, firebaseAuth.verifyPasswordResetCode(code));
      }
   });
}

/**
 * Asynchronously downloads the object from this {@link StorageReference} a byte array will be allocated large enough to hold the entire file in memory.
 *
 * @param storageRef           represents a reference to a Google Cloud Storage object.
 * @param maxDownloadSizeBytes the maximum allowed size in bytes that will be allocated. Set this parameter to prevent out of memory conditions from occurring.
 *                             If the download exceeds this limit, the task will fail and an IndexOutOfBoundsException will be returned.
 * @return a {@link Maybe} which emits an byte[] if success.
 */
@NonNull
public static Maybe<byte[]> getBytes(@NonNull final StorageReference storageRef,
                                     final long maxDownloadSizeBytes) {
    return Maybe.create(new MaybeOnSubscribe<byte[]>() {
        @Override
        public void subscribe(MaybeEmitter<byte[]> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.getBytes(maxDownloadSizeBytes));
        }
    });
}

/**
 * Asynchronously retrieves a long lived download URL with a revocable token.
 *
 * @param storageRef represents a reference to a Google Cloud Storage object.
 * @return a {@link Maybe} which emits an {@link Uri} if success.
 */
@NonNull
public static Maybe<Uri> getDownloadUrl(@NonNull final StorageReference storageRef) {
    return Maybe.create(new MaybeOnSubscribe<Uri>() {
        @Override
        public void subscribe(MaybeEmitter<Uri> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.getDownloadUrl());
        }
    });
}

/**
 * Asynchronously downloads the object at this {@link StorageReference} to a specified system filepath.
 *
 * @param storageRef      represents a reference to a Google Cloud Storage object.
 * @param destinationFile a File representing the path the object should be downloaded to.
 * @return a {@link Maybe} which emits an {@link FileDownloadTask.TaskSnapshot} if success.
 */
@NonNull
public static Maybe<FileDownloadTask.TaskSnapshot> getFile(@NonNull final StorageReference storageRef,
                                                           @NonNull final File destinationFile) {
    return Maybe.create(new MaybeOnSubscribe<FileDownloadTask.TaskSnapshot>() {
        @Override
        public void subscribe(MaybeEmitter<FileDownloadTask.TaskSnapshot> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.getFile(destinationFile));
        }
    });
}

/**
 * Asynchronously downloads the object at this {@link StorageReference} to a specified system filepath.
 *
 * @param storageRef     represents a reference to a Google Cloud Storage object.
 * @param destinationUri a file system URI representing the path the object should be downloaded to.
 * @return a {@link Maybe} which emits an {@link FileDownloadTask.TaskSnapshot} if success.
 */
@NonNull
public static Maybe<FileDownloadTask.TaskSnapshot> getFile(@NonNull final StorageReference storageRef,
                                                           @NonNull final Uri destinationUri) {
    return Maybe.create(new MaybeOnSubscribe<FileDownloadTask.TaskSnapshot>() {
        @Override
        public void subscribe(MaybeEmitter<FileDownloadTask.TaskSnapshot> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.getFile(destinationUri));
        }
    });
}

/**
 * Retrieves metadata associated with an object at this {@link StorageReference}.
 *
 * @param storageRef represents a reference to a Google Cloud Storage object.
 * @return a {@link Maybe} which emits an {@link StorageMetadata} if success.
 */
@NonNull
public static Maybe<StorageMetadata> getMetadata(@NonNull final StorageReference storageRef) {
    return Maybe.create(new MaybeOnSubscribe<StorageMetadata>() {
        @Override
        public void subscribe(MaybeEmitter<StorageMetadata> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.getMetadata());
        }
    });
}

/**
 * Asynchronously downloads the object at this {@link StorageReference} via a InputStream.
 *
 * @param storageRef represents a reference to a Google Cloud Storage object.
 * @return a {@link Maybe} which emits an {@link StreamDownloadTask.TaskSnapshot} if success.
 */
@NonNull
public static Maybe<StreamDownloadTask.TaskSnapshot> getStream(@NonNull final StorageReference storageRef) {
    return Maybe.create(new MaybeOnSubscribe<StreamDownloadTask.TaskSnapshot>() {
        @Override
        public void subscribe(MaybeEmitter<StreamDownloadTask.TaskSnapshot> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.getStream());
        }
    });
}

/**
 * Asynchronously downloads the object at this {@link StorageReference} via a InputStream.
 *
 * @param storageRef represents a reference to a Google Cloud Storage object.
 * @param processor  A StreamDownloadTask.StreamProcessor that is responsible for reading data from the InputStream.
 *                   The StreamDownloadTask.StreamProcessor is called on a background thread and checked exceptions thrown
 *                   from this object will be returned as a failure to the OnFailureListener registered on the StreamDownloadTask.
 * @return a {@link Maybe} which emits an {@link StreamDownloadTask.TaskSnapshot} if success.
 */
@NonNull
public static Maybe<StreamDownloadTask.TaskSnapshot> getStream(@NonNull final StorageReference storageRef,
                                                               @NonNull final StreamDownloadTask.StreamProcessor processor) {
    return Maybe.create(new MaybeOnSubscribe<StreamDownloadTask.TaskSnapshot>() {
        @Override
        public void subscribe(MaybeEmitter<StreamDownloadTask.TaskSnapshot> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.getStream(processor));
        }
    });
}

/**
 * Asynchronously uploads byte data to this {@link StorageReference}.
 *
 * @param storageRef represents a reference to a Google Cloud Storage object.
 * @param bytes      The byte[] to upload.
 * @return a {@link Maybe} which emits an {@link UploadTask.TaskSnapshot} if success.
 */
@NonNull
public static Maybe<UploadTask.TaskSnapshot> putBytes(@NonNull final StorageReference storageRef,
                                                      @NonNull final byte[] bytes) {
    return Maybe.create(new MaybeOnSubscribe<UploadTask.TaskSnapshot>() {
        @Override
        public void subscribe(MaybeEmitter<UploadTask.TaskSnapshot> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.putBytes(bytes));
        }
    });
}

/**
 * Asynchronously uploads byte data to this {@link StorageReference}.
 *
 * @param storageRef represents a reference to a Google Cloud Storage object.
 * @param bytes      The byte[] to upload.
 * @param metadata   {@link StorageMetadata} containing additional information (MIME type, etc.) about the object being uploaded.
 * @return a {@link Maybe} which emits an {@link UploadTask.TaskSnapshot} if success.
 */
@NonNull
public static Maybe<UploadTask.TaskSnapshot> putBytes(@NonNull final StorageReference storageRef,
                                                      @NonNull final byte[] bytes,
                                                      @NonNull final StorageMetadata metadata) {
    return Maybe.create(new MaybeOnSubscribe<UploadTask.TaskSnapshot>() {
        @Override
        public void subscribe(MaybeEmitter<UploadTask.TaskSnapshot> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.putBytes(bytes, metadata));
        }
    });
}

/**
 * Asynchronously uploads from a content URI to this {@link StorageReference}.
 *
 * @param storageRef represents a reference to a Google Cloud Storage object.
 * @param uri        The source of the upload. This can be a file:// scheme or any content URI. A content resolver will be used to load the data.
 * @return a {@link Maybe} which emits an {@link UploadTask.TaskSnapshot} if success.
 */
@NonNull
public static Maybe<UploadTask.TaskSnapshot> putFile(@NonNull final StorageReference storageRef,
                                                     @NonNull final Uri uri) {
    return Maybe.create(new MaybeOnSubscribe<UploadTask.TaskSnapshot>() {
        @Override
        public void subscribe(MaybeEmitter<UploadTask.TaskSnapshot> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.putFile(uri));
        }
    });
}

/**
 * Asynchronously uploads from a content URI to this {@link StorageReference}.
 *
 * @param storageRef represents a reference to a Google Cloud Storage object.
 * @param uri        The source of the upload. This can be a file:// scheme or any content URI. A content resolver will be used to load the data.
 * @param metadata   {@link StorageMetadata} containing additional information (MIME type, etc.) about the object being uploaded.
 * @return a {@link Maybe} which emits an {@link UploadTask.TaskSnapshot} if success.
 */
@NonNull
public static Maybe<UploadTask.TaskSnapshot> putFile(@NonNull final StorageReference storageRef,
                                                     @NonNull final Uri uri,
                                                     @NonNull final StorageMetadata metadata) {
    return Maybe.create(new MaybeOnSubscribe<UploadTask.TaskSnapshot>() {
        @Override
        public void subscribe(MaybeEmitter<UploadTask.TaskSnapshot> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.putFile(uri, metadata));
        }
    });
}

/**
 * Asynchronously uploads from a content URI to this {@link StorageReference}.
 *
 * @param storageRef        represents a reference to a Google Cloud Storage object.
 * @param uri               The source of the upload. This can be a file:// scheme or any content URI. A content resolver will be used to load the data.
 * @param metadata          {@link StorageMetadata} containing additional information (MIME type, etc.) about the object being uploaded.
 * @param existingUploadUri If set, an attempt is made to resume an existing upload session as defined by getUploadSessionUri().
 * @return a {@link Maybe} which emits an {@link UploadTask.TaskSnapshot} if success.
 */
@NonNull
public static Maybe<UploadTask.TaskSnapshot> putFile(@NonNull final StorageReference storageRef,
                                                     @NonNull final Uri uri,
                                                     @NonNull final StorageMetadata metadata,
                                                     @NonNull final Uri existingUploadUri) {
    return Maybe.create(new MaybeOnSubscribe<UploadTask.TaskSnapshot>() {
        @Override
        public void subscribe(MaybeEmitter<UploadTask.TaskSnapshot> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.putFile(uri, metadata, existingUploadUri));
        }
    });
}

/**
 * @param storageRef represents a reference to a Google Cloud Storage object.
 * @param stream     The InputStream to upload.
 * @param metadata   {@link StorageMetadata} containing additional information (MIME type, etc.) about the object being uploaded.
 * @return a {@link Maybe} which emits an {@link UploadTask.TaskSnapshot} if success.
 */
@NonNull
public static Maybe<UploadTask.TaskSnapshot> putStream(@NonNull final StorageReference storageRef,
                                                       @NonNull final InputStream stream,
                                                       @NonNull final StorageMetadata metadata) {
    return Maybe.create(new MaybeOnSubscribe<UploadTask.TaskSnapshot>() {
        @Override
        public void subscribe(MaybeEmitter<UploadTask.TaskSnapshot> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.putStream(stream, metadata));
        }
    });
}

/**
 * Asynchronously uploads a stream of data to this {@link StorageReference}.
 *
 * @param storageRef represents a reference to a Google Cloud Storage object.
 * @param stream     The InputStream to upload.
 * @return a {@link Maybe} which emits an {@link UploadTask.TaskSnapshot} if success.
 */
@NonNull
public static Maybe<UploadTask.TaskSnapshot> putStream(@NonNull final StorageReference storageRef,
                                                       @NonNull final InputStream stream) {
    return Maybe.create(new MaybeOnSubscribe<UploadTask.TaskSnapshot>() {
        @Override
        public void subscribe(MaybeEmitter<UploadTask.TaskSnapshot> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.putStream(stream));
        }
    });
}

/**
 * Asynchronously uploads a stream of data to this {@link StorageReference}.
 *
 * @param storageRef represents a reference to a Google Cloud Storage object.
 * @param metadata   {@link StorageMetadata} containing additional information (MIME type, etc.) about the object being uploaded.
 * @return a {@link Maybe} which emits an {@link StorageMetadata} if success.
 */
@NonNull
public static Maybe<StorageMetadata> updateMetadata(@NonNull final StorageReference storageRef,
                                                    @NonNull final StorageMetadata metadata) {
    return Maybe.create(new MaybeOnSubscribe<StorageMetadata>() {
        @Override
        public void subscribe(MaybeEmitter<StorageMetadata> emitter) throws Exception {
            RxHandler.assignOnTask(emitter, storageRef.updateMetadata(metadata));
        }
    });
}

public static <T> void assignOnTask(MaybeEmitter<? super T> emitter, Task<T> task) {
    RxHandler handler = new RxHandler(emitter);
    task.addOnSuccessListener(handler);
    task.addOnFailureListener(handler);
    try {
        task.addOnCompleteListener(handler);
    } catch (Throwable t) {
        // ignore
    }
}

@Override
public Maybe<Response> download(Request request) {

    okhttp3.Request okrequest = new okhttp3.Request.Builder()
            .url(request.getUrl())
            .build();

    return Maybe.create(new MaybeOnSubscribe<okhttp3.Response>(){

        @Override
        public void subscribe(MaybeEmitter emitter) throws Exception {

            emitter.onSuccess(client.newCall(okrequest).execute());
        }
    }).map(new Function<okhttp3.Response, Response>() {

        @Override
        public Response apply(okhttp3.Response resp) throws Exception {

            String html = resp.body().string();
            Response response = new Response();
            response.setContent(html);
            response.setStatusCode(resp.code());
            return response;
        }
    });
}

@Override
protected void onGoogleApiClientReady(GoogleApiClient apiClient, MaybeEmitter<Location> emitter) {
    //noinspection MissingPermission
    Location location = LocationServices.FusedLocationApi.getLastLocation(apiClient);

    if (location != null) {
        emitter.onSuccess(location);
    } else {
        emitter.onComplete();
    }
}

/**
 * @param task
 * @param <R>
 * @return
 */
@CheckReturnValue
@NonNull
public static <R> Maybe<R> maybe(@NonNull final Task<R> task) {
    return Maybe.create(new MaybeOnSubscribe<R>() {
        @Override
        public void subscribe(@NonNull final MaybeEmitter<R> emit) throws Exception {
            task.addOnCompleteListener(listener(emit));
        }
    });
}

@Override public void subscribe(MaybeEmitter<FirebaseUser> emitter) {
  if (!emitter.isDisposed()) {
    FirebaseUser currentUser = instance.getCurrentUser();
    if (currentUser != null) {
      emitter.onSuccess(currentUser);
    } else {
      emitter.onComplete();
    }
  }
}