Google Cloud Java Client libraries use a ApiFuture type to represent the result of an API call. The calls are asynchronous and the ApiFuture type represents the result once the call is completed.
If you have used Reactive stream based libraries like Project Reactor, a big benefit of using the Reactive types like Mono and Flux is that they provide a rich set of operators that provide a way to transform the data once available from the asynchronous call.
This should become clearer in an example. Consider a Cloud Firestore call to retrieve a ChatRoom entity by id:
public ChatRoom getById(String id) {
ApiFuture<DocumentSnapshot> chatRoomSnapshotFuture =
firestore.collection(ServiceConstants.CHAT_ROOMS).document(id).get();
try {
DocumentSnapshot chatRoomSnapshot = chatRoomSnapshotFuture.get();
return new ChatRoom(chatRoomSnapshot.getId(), chatRoomSnapshot.getString("name"));
} catch (Exception e) {
throw new RuntimeException("Could not retrieve by id", e);
}
}
There are few issues here, the “get()” call is used for blocking and waiting on the response of the async call to come through, which can throw an exception which needs to be accounted for. Then the response is shaped into the ChatRoom type.
Now, look at the same flow with reactive types, assuming that there is a utility available to convert the ApiFuture type to the Mono type:
public Mono<ChatRoom> getById(String id) {
ApiFuture<DocumentSnapshot> chatRoomSnapshotFuture =
firestore.collection(ServiceConstants.CHAT_ROOMS).document(id).get();
Mono<DocumentSnapshot> chatRoomSnapshotMono = ApiFutureUtil.toMono(chatRoomSnapshotFuture);
return chatRoomSnapshotMono.map(chatRoomSnapshot ->
new ChatRoom(chatRoomSnapshot.getId(), chatRoomSnapshot.getString("name")));
}
Here the map operator takes care of transforming the result to the required “ChatRoom” type and any exception is wrapped in Mono type itself.
Alright, so now how can the ApiFutureUtil be implemented, a basic implementation looks like this:
import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutureCallback;
import com.google.api.core.ApiFutures;
import reactor.core.publisher.Mono;
public class ApiFutureUtil {
public static <T> Mono<T> toMono(ApiFuture<T> apiFuture) {
return Mono.create(sink -> {
ApiFutureCallback<T> callback = new ApiFutureCallback<T>() {
@Override
public void onFailure(Throwable t) {
sink.error(t);
}
@Override
public void onSuccess(T result) {
sink.success(result);
}
};
ApiFutures.addCallback(apiFuture, callback, Runnable::run);
});
}
}
This utility serves the purpose of transforming the ApiFuture type, however one catch is that this Mono type is hot. What does this mean — normally reactive streams pipeline(with all the operators chained together) represents the computation, this computation comes alive only when somebody subscribes to this pipeline, with a ApiFuture converted to Mono, even without anybody subscribing, the result will still be emitted. This is okay as the purpose is to use the Mono type for its operators. If “cold” is desired then even the Api call itself can be deferred something like this:
public Mono<ChatRoom> getById(String id) {
return Mono.defer(() -> {
ApiFuture<DocumentSnapshot> chatRoomSnapshotFuture =
firestore.collection(ServiceConstants.CHAT_ROOMS).document(id).get();
Mono<DocumentSnapshot> chatRoomSnapshotMono = ApiFutureUtil.toMono(chatRoomSnapshotFuture);
return chatRoomSnapshotMono.map(chatRoomSnapshot ->
new ChatRoom(chatRoomSnapshot.getId(), chatRoomSnapshot.getString("name")));
});
}
I hope this gives some idea of how Reactive Stream types can be created from ApiFuture. This is far from original though, if you desire a canned approach of doing this, a better solution is to use Spring-Cloud-Gcp Java library which already has these utilities baked in.
Published on Java Code Geeks with permission by Biju Kunjummen, partner at our JCG program. See the original article here: Google Cloud Java Client – ApiFuture to Reactive types Opinions expressed by Java Code Geeks contributors are their own. |
