How to efficiently have sync and async implementation in a library which uses Guava ListenableFuture?

[shaheen...@gmail.com]

I have a library in which I have provided two methods, sync and async for our customer. They can call whichever method they feel is right for their purpose.

- executeSync() - waits until I have a result, returns the result.
- executeAsync() - returns a Future immediately which can be processed after other things are done, if needed.

They will pass DataKey object which has the user id in it. And we will figure out which machine to call basis on the user id. So we will make http call to the url using AsyncRestTemplate and then send the response back to them basis on whether it is successful or not.

AsyncRestTemplate (http://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/web/client/AsyncRestTemplate.html#exchange-java.net.URI-org.springframework.http.HttpMethod-org.springframework.http.HttpEntity-java.lang.Class-) returns back ListenableFuture.

Below is my interface:

    public interface Client {
        // for synchronous
        public DataResponse executeSync(final DataKey key);
       
        // for asynchronous
        public Future<DataResponse> executeAsync(final DataKey key);
    }

And below is my implementation:

    public class DataClient implements IClient {

        private final AsyncRestTemplate restTemplate = new AsyncRestTemplate();

        @Override
        public DataResponse executeSync(final DataKey keys) {
            Future<DataResponse> responseFuture = executeAsync(keys);
            DataResponse response = null;
            try {
                response = responseFuture.get(keys.getTimeout(), TimeUnit.Milliseconds);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("Interrupted", e);
            } catch (TimeoutException e) {
                DataLogging.logErrors(e.getCause(), DataErrorEnum.TIMEOUT_ON_CLIENT, keys);
                response = new DataResponse(null, DataErrorEnum.TIMEOUT_ON_CLIENT, DataStatusEnum.ERROR);       
            } catch (TimeoutException e) {
                DataLogging.logErrors(e.getCause(), DataErrorEnum.ERROR_CLIENT, keys);
                response = new DataResponse(null, DataErrorEnum.ERROR_CLIENT, DataStatusEnum.ERROR);                   
            }

            return response;
        }

        @Override
        public Future<DataResponse> executeAsync(final DataKey keys) {
            final SettableFuture<DataResponse> responseFuture = SettableFuture.create();
            restTemplate.exchange(createURL(keys), HttpMethod.GET, keys.getEntity(), String.class).addCallback(
                    new ListenableFutureCallback<ResponseEntity<String>>() {
                        @Override
                        public void onSuccess(ResponseEntity<String> result) {
                            responseFuture.set(new DataResponse(result.getBody(), DataErrorEnum.OK,
                                    DataStatusEnum.SUCCESS));
                        }

                        @Override
                        public void onFailure(Throwable ex) {
                            DataLogging.logErrors(ex, DataErrorEnum.ERROR_SERVER, keys);
                            responseFuture.set(new DataResponse(null, DataErrorEnum.ERROR_CLIENT,
                                    DataStatusEnum.ERROR));
                        }
                    });

            return responseFuture;

        }
    }

Now my question is: Is this the right implementation of having sync and async implementation in a library which is using Guava ListenableFuture? Or is there any better way to do this? I wanted to have async non blocking architecture so that's why I went with AsyncRestTemplate which returns back to me as ListenableFuture. I have tried to make the example pretty simple, all the logic of figuring out which machine to call basis on user id is not show but what I have will give you the design flow how I am using it.

I don't wanted to implement sync call as async + waiting along with ExecutorService thread pool since it might be a bad idea because it will consume one thread from the thread pool per a call.
 
Any inputs/suggestions are also welcome on my design for having sync and async implementations in a library which uses Guava ListenableFuture.

[Luke Sandberg]

You should declare your executeAsync() method to return ListenableFuture instead of future so that your clients can use things like Futures.transform on it.

IMHO, you should provide synchronous apis, if your client wants to block they can just call executeAsync().get().  ListenableFuture already is an async/sync api and by providing explicit sync apis you are just creating an attractive nuisance.

your executeAsync method is a little weird.  typically the correct way to represent failure in a settablefuture is to call .setException(), not with an error type.  Also you probably want to do something when future.cancel() is called to cancel the underlying http request

--
guava-...@googlegroups.com
Project site: https://github.com/google/guava
This group: http://groups.google.com/group/guava-discuss
 
This list is for general discussion.
To report an issue: https://github.com/google/guava/issues/new
To get help: http://stackoverflow.com/questions/ask?tags=guava
---
You received this message because you are subscribed to the Google Groups "guava-discuss" group.
To unsubscribe from this group and stop receiving emails from it, send an email to guava-discus...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/guava-discuss/a8164923-f8d3-4d00-a8fd-fc619b3fe6db%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Shaheen Afroz]

Thanks a lot Luke for reviewing my design. Appreciated your help.

I am pretty new to Guava Listenable Future design so that's why I wanted to review that whether I was using the right way or not. Now as per your suggestion, it looks like I was doing something wrong for sure. I implemented both sync and async implementation in our library because by that we can measure how much time our library is taking on sync call end to end. This is for our monitoring purpose to make sure our library is well under SLA which is less than 10 ms at 95th percentile. We have some sort of logging in our company by which we can measure that.

And customer also wanted to have both the sync and async implementation so that's why I provided both. It's up to them what they want to use and depending on their use case, they can use either one.

Now coming back to the design, you wanted to have my interface look like this - Right?

    public interface Client {
        // for synchronous
        public DataResponse executeSync(final DataKey key);
     
        // for asynchronous

        public ListenableFuture<DataResponse> executeAsync(final DataKey key);
    }

Now I was not able to understand other part which you were suggesting in executeAsync method. Can you provide an example how will you design executeAsync and executeSync method if you were planning to do it. This will help me to understand better how to use ListenableFuture and SettableFuture efficiently in this scenario.

[Luke Sandberg]

For monitoring you can always just perform it in your future callback.  That is the most common strategy i see.  So i think you would implement it something like:

public interface Client {
    // for asynchronous
    ListenableFuture<DataResponse> executeAsync(DataKey key);

}

  public class DataClient implements IClient {

        private final AsyncRestTemplate restTemplate = new AsyncRestTemplate();

        @Override
        public ListenableFuture<DataResponse> executeAsync(final DataKey keys) {

            final SettableFuture<DataResponse> responseFuture = SettableFuture.create();

            final org.springframework.util.concurrent.ListenableFuture orig = 

                restTemplate.exchange(createURL(keys), HttpMethod.GET, keys.getEntity(), String.class);

            orig.addCallback(

                    new ListenableFutureCallback<ResponseEntity<String>>() {
                        @Override
                        public void onSuccess(ResponseEntity<String> result) {

                            responseFuture.set(new DataResponse(result.getBody(), DataErrorEnum.OK,
                                    DataStatusEnum.SUCCESS));

                           // log your latency here

                        }

                        @Override
                        public void onFailure(Throwable ex) {
                            DataLogging.logErrors(ex, DataErrorEnum.ERROR_SERVER, keys);

                            // use an appropriate exception type?

                            responseFuture.set(new DataResponse(null, DataErrorEnum.ERROR_CLIENT,
                                    DataStatusEnum.ERROR));
                        }
                    });

            // propagate cancellation back to the original request
            responseFuture.addListener(new Runnable() {

              @Override public void run() {

                 if (responseFuture.isCancelled()) {

                   orig.cancel(false);

                 }

              }

            }, MoreExecutors.directExecutor());
            return responseFuture;
        }

If you find yourself dealing with spring futures a lot, i would consider creating a standard utility that adapts a springfuture to a guava listenablefuture.

For sync clients they can just call .get() on the future (or .get(1, TimeUnit.MILLISECONDS)),  as part of your documentation you should make sure to document your error conditions.  If you really have to provide a sync interface, you could do this pretty much what you have above.  Though again, exceptions are generally preferred to error types unless you have a good reason to not use exceptions.

[Shaheen Afroz]

Thanks a lot Luke. Appreciated. I understood lot of things now basis on your example. I want to ask few things on your suggestion for my understanding -

1) // use an appropriate exception type? So by this you mean to say, I should return back the exact exception type like if server returned back HttpStatus.BAD_REQUEST or HttpStatus.UNAUTHORIZED then I should return those, instead of returning generalize error "DataErrorEnum.ERROR_CLIENT". Right? If yes, that's what I am already doing but for simplicity case, I made the "executeAsync" method very simple so that people can understand clearly what I am trying to do.

2) At the very end of "executeAsync" method, we are returning "return responseFuture;" but "responseFuture" is of type "SettableFuture" and "executeAsync" return type is "ListenableFuture" so I am seeing compilcation error. To fix, I just need to modify the "executeAsync" implementation datatype from "ListenableFuture<DataResponse>" to "SettableFuture<DataResponse>" right?

                             // something like this?

                              public SettableFuture<DataResponse> executeAsync(final DataKey keys) {

                                  // code here

                              }

or I need to modify both the interface and implementation of executeAsync like this -

                              public interface Client {
                                  // for asynchronous

                                 SettableFuture<DataResponse> executeAsync(DataKey key);
                              }


                             // something like this?
                              public SettableFuture<DataResponse> executeAsync(final DataKey keys) {

                                    // code here

                              }

3) Thirdly, if cancel is called on future, then I am seeing that we are also calling "cancel" on "orig" like this orig.cancel(false);

Question is -  why we are passing "false" in this case? Shouldn't this be true? Or what is the need of having this listener again?

                        // propagate cancellation back to the original request
                        responseFuture.addListener(new Runnable() {
                            @Override
                            public void run() {
                                if (responseFuture.isCancelled()) {

                                    orig.cancel(false); // why we have false here?
                                }
                            }
                        }, MoreExecutors.directExecutor());


If my understanding is right - The reason for adding the listener and passing false is to make sure that we can propagate this back to the caller? Meaning if we don't do that, it will just log the exception in onFailure future callback method?

[Luke Sandberg]

1)

I just mean that in java the standard way to signal errors is via exceptions.  So if your request has failed, you should probably be calling setException() on the future with an appropriate exception type

2).  you are probably getting confused by the fact that you have two classes called ListenableFuture on your classpath, there is com.google.common.util.concurrent.ListenableFuture and there is org.springframework.util.concurrent.ListenableFuture, SettableFuture implements the former (so make sure you either use a fully qualified class name or don't import the springframework version)

3)

This is all about propagating cancellation, the choice of passing true or false often doesn't matter very much and i often choose false as a default because people don't tend to handle interruption gracefully.  If you want to be accurate, you can propagate the original setting but to do that you can't use SettableFuture, you need to create a custom com.google.common.util.concurrent.AbstractFuture subclass.  For example, you can see how futures.transform does cancellation propagation: https://github.com/google/guava/blob/master/guava/src/com/google/common/util/concurrent/Futures.java#L1437

If you don't propagate cancellation back to the original future, you won't actually cancel the corresponding http request (which you probably want to do to save resources).