Handling Google API Java Client metadata request and response with RxJava

GoogleUtils.java

/**
 * Sends the {@code AbstractGoogleJsonClientRequest} to the server and returns an {@code Observable}
 * that emits the parsed {@code GenericJson} response.
 *
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code executeAsObservable} does not operate by default on a particular
 *  {@link rx.Scheduler}.</dd>
 * </dl>
 *
 * @param request    {@code AbstractGoogleJsonClientRequest} to be sent to the server
 * @param <T>        {@code GenericJson} that this {@code Observable} emits
 * @return An {@code Observable} that emits the parsed {@code GenericJson} response.
 */
public static <T extends GenericJson> Observable<T> executeAsObservable(
        @Nonnull final AbstractGoogleJsonClientRequest<T> request) {
    return Observable.create((subscriber) -> {
            try {
                if (!subscriber.isUnsubscribed()) {
                    subscriber.onNext(request.execute());
                    subscriber.onCompleted();
                }
            } catch (IOException e) {
                subscriber.onError(e);
            }
    });
}

/**
 * Sends a {@code batchRequest} composed of {@code requests} to the server and returns
 * an {@code Observable} that emits the parsed {@code GenericJson} responses.
 *
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code batchExecuteAsObservable} does not operate by default on a particular
 *  {@link rx.Scheduler}.</dd>
 * </dl>
 *
 * @param batchRequest    {@code BatchRequest} to be sent to the server
 * @param requests        {@code AbstractGoogleJsonClientRequest}s to queue
 * @param <T>             {@code GenericJson} that this {@code Observable} emits
 * @return An {@code Observable} that emits the parsed {@code GenericJson} responses.
 */
public static <T extends GenericJson> Observable<T> batchExecuteAsObservable(
        @Nonnull final BatchRequest batchRequest,
        @Nonnull final Collection<? extends AbstractGoogleJsonClientRequest<T>> requests) {
    if (requests.isEmpty()) {
        return Observable.empty();
    } else if (requests.size() == 1) {
        return executeAsObservable(requests.iterator().next());
    }

    return Observable.create((subscriber) -> {
            if (subscriber.isUnsubscribed()) {
                return;
            }

            try {
                final List<JsonBatchCallback<T>> callbacks = Collections
                        .synchronizedList(new ArrayList<JsonBatchCallback<T>>(requests.size()));
                for (AbstractGoogleJsonClientRequest<T> request : requests) {
                    JsonBatchCallback<T> batchCallback = new JsonBatchCallback<T>() {
                        @Override
                        public void onFailure(GoogleJsonError e, HttpHeaders responseHeaders)
                                throws IOException {
                            System.err.println("There was a JSON error: " + e.getMessage());
                        }

                        @Override
                        public void onSuccess(T t, HttpHeaders responseHeaders)
                                throws IOException {
                            callbacks.remove(this);
                            if (!subscriber.isUnsubscribed()) {
                                subscriber.onNext(t);
                                if (callbacks.isEmpty()) {
                                    subscriber.onCompleted();
                                }
                            }
                        }
                    };
                    callbacks.add(batchCallback);
                    request.queue(batchRequest, batchCallback);
                }
                batchRequest.execute();
            } catch (IOException e) {
                subscriber.onError(e);
            }
    });
}