use RpcFuture and remove old BundlingSettings

google-cloud-core/pom.xml

@@ -111,7 +111,7 @@
     <dependency>
       <groupId>com.google.api</groupId>
       <artifactId>gax</artifactId>
-      <version>0.0.25</version>
+      <version>0.0.28-SNAPSHOT</version>
       <exclusions>
         <exclusion>
           <groupId>io.grpc</groupId>

google-cloud-logging/src/main/java/com/google/cloud/logging/LoggingImpl.java

@@ -40,7 +40,6 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.Uninterruptibles;
 import com.google.logging.v2.CreateLogMetricRequest;
 import com.google.logging.v2.CreateSinkRequest;
@@ -107,9 +106,6 @@ private static <V> V get(Future<V> future) {
 
   private static <I, O> Future<O> transform(Future<I> future,
       Function<? super I, ? extends O> function) {
-    if (future instanceof ListenableFuture) {
-      return Futures.transform((ListenableFuture<I>) future, function);
-    }
     return Futures.lazyTransform(future, function);
   }
 

google-cloud-logging/src/main/java/com/google/cloud/logging/spi/DefaultLoggingRpc.java

@@ -22,6 +22,7 @@
 import com.google.api.gax.grpc.FixedChannelProvider;
 import com.google.api.gax.grpc.FixedExecutorProvider;
 import com.google.api.gax.grpc.ProviderManager;
+import com.google.api.gax.grpc.RpcFuture;
 import com.google.api.gax.grpc.UnaryCallSettings;
 import com.google.cloud.GrpcServiceOptions.ExecutorFactory;
 import com.google.cloud.NoCredentials;
@@ -33,10 +34,6 @@
 import com.google.cloud.logging.spi.v2.LoggingServiceV2Settings;
 import com.google.cloud.logging.spi.v2.MetricsServiceV2Client;
 import com.google.cloud.logging.spi.v2.MetricsServiceV2Settings;
-import com.google.common.base.Function;
-import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
 import com.google.logging.v2.CreateLogMetricRequest;
 import com.google.logging.v2.CreateSinkRequest;
 import com.google.logging.v2.DeleteLogMetricRequest;
@@ -59,13 +56,13 @@
 import com.google.logging.v2.WriteLogEntriesRequest;
 import com.google.logging.v2.WriteLogEntriesResponse;
 import com.google.protobuf.Empty;
-
 import io.grpc.ManagedChannel;
 import io.grpc.Status.Code;
 import io.grpc.netty.NegotiationType;
 import io.grpc.netty.NettyChannelBuilder;
-
 import java.io.IOException;
+import java.util.Collections;
+import java.util.EnumSet;
 import java.util.Set;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
@@ -148,21 +145,25 @@ public DefaultLoggingRpc(LoggingOptions options) throws IOException {
     }
   }
 
-  private static <V> Future<V> translate(ListenableFuture<V> from, final boolean idempotent,
-      int... returnNullOn) {
-    final Set<Integer> returnNullOnSet = Sets.newHashSetWithExpectedSize(returnNullOn.length);
-    for (int value : returnNullOn) {
-      returnNullOnSet.add(value);
+  private static <V> Future<V> translate(
+      RpcFuture<V> from, final boolean idempotent, Code... returnNullOn) {
+    final Set<Code> returnNullOnSet;
+    if (returnNullOn.length > 0) {
+      returnNullOnSet = EnumSet.of(returnNullOn[0], returnNullOn);
+    } else {
+      returnNullOnSet = Collections.<Code>emptySet();
     }
-    return Futures.catching(from, ApiException.class, new Function<ApiException, V>() {
-      @Override
-      public V apply(ApiException exception) {
-        if (returnNullOnSet.contains(exception.getStatusCode().value())) {
-          return null;
-        }
-        throw new LoggingException(exception, idempotent);
-      }
-    });
+    return from.catching(
+        ApiException.class,
+        new RpcFuture.Function<ApiException, V>() {
+          @Override
+          public V apply(ApiException exception) {
+            if (returnNullOnSet.contains(exception.getStatusCode().value())) {
+              return null;
+            }
+            throw new LoggingException(exception, idempotent);
+          }
+        });
   }
 
   @Override
@@ -177,7 +178,7 @@ public Future<LogSink> update(UpdateSinkRequest request) {
 
   @Override
   public Future<LogSink> get(GetSinkRequest request) {
-    return translate(configClient.getSinkCallable().futureCall(request), true, Code.NOT_FOUND.value());
+    return translate(configClient.getSinkCallable().futureCall(request), true, Code.NOT_FOUND);
   }
 
   @Override
@@ -187,14 +188,12 @@ public Future<ListSinksResponse> list(ListSinksRequest request) {
 
   @Override
   public Future<Empty> delete(DeleteSinkRequest request) {
-    return translate(configClient.deleteSinkCallable().futureCall(request), true,
-        Code.NOT_FOUND.value());
+    return translate(configClient.deleteSinkCallable().futureCall(request), true, Code.NOT_FOUND);
   }
 
   @Override
   public Future<Empty> delete(DeleteLogRequest request) {
-    return translate(loggingClient.deleteLogCallable().futureCall(request), true,
-        Code.NOT_FOUND.value());
+    return translate(loggingClient.deleteLogCallable().futureCall(request), true, Code.NOT_FOUND);
   }
 
   @Override
@@ -226,8 +225,8 @@ public Future<LogMetric> update(UpdateLogMetricRequest request) {
 
   @Override
   public Future<LogMetric> get(GetLogMetricRequest request) {
-    return translate(metricsClient.getLogMetricCallable().futureCall(request), true,
-        Code.NOT_FOUND.value());
+    return translate(
+        metricsClient.getLogMetricCallable().futureCall(request), true, Code.NOT_FOUND);
   }
 
   @Override
@@ -237,8 +236,8 @@ public Future<ListLogMetricsResponse> list(ListLogMetricsRequest request) {
 
   @Override
   public Future<Empty> delete(DeleteLogMetricRequest request) {
-    return translate(metricsClient.deleteLogMetricCallable().futureCall(request), true,
-        Code.NOT_FOUND.value());
+    return translate(
+        metricsClient.deleteLogMetricCallable().futureCall(request), true, Code.NOT_FOUND);
   }
 
   @Override

google-cloud-logging/src/main/java/com/google/cloud/logging/spi/v2/LoggingServiceV2Settings.java

@@ -546,11 +546,8 @@ private static Builder createDefault() {
           .writeLogEntriesSettings()
           .getBundlingSettingsBuilder()
           .setElementCountThreshold(1)
-          .setElementCountLimit(1000)
           .setRequestByteThreshold(1024)
-          .setRequestByteLimit(10485760)
-          .setDelayThreshold(Duration.millis(10))
-          .setBlockingCallCountThreshold(1);
+          .setDelayThreshold(Duration.millis(10));
       builder
           .writeLogEntriesSettings()
           .setRetryableCodes(RETRYABLE_CODE_DEFINITIONS.get("non_idempotent"))

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/PubSubImpl.java

@@ -41,7 +41,6 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.Uninterruptibles;
 import com.google.iam.v1.SetIamPolicyRequest;
 import com.google.iam.v1.TestIamPermissionsRequest;
@@ -222,9 +221,6 @@ private static <V> V get(Future<V> future) {
 
   private static <I, O> Future<O> transform(Future<I> future,
       Function<? super I, ? extends O> function) {
-    if (future instanceof ListenableFuture) {
-      return Futures.transform((ListenableFuture<I>) future, function);
-    }
     return Futures.lazyTransform(future, function);
   }
 

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/DefaultPubSubRpc.java

@@ -22,6 +22,8 @@
 import com.google.api.gax.grpc.FixedChannelProvider;
 import com.google.api.gax.grpc.FixedExecutorProvider;
 import com.google.api.gax.grpc.ProviderManager;
+import com.google.api.gax.grpc.RpcFuture;
+import com.google.api.gax.grpc.RpcFutureCallback;
 import com.google.api.gax.grpc.UnaryCallSettings;
 import com.google.cloud.GrpcServiceOptions.ExecutorFactory;
 import com.google.cloud.NoCredentials;
@@ -31,12 +33,7 @@
 import com.google.cloud.pubsub.spi.v1.PublisherSettings;
 import com.google.cloud.pubsub.spi.v1.SubscriberClient;
 import com.google.cloud.pubsub.spi.v1.SubscriberSettings;
-import com.google.common.base.Function;
 import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.ForwardingListenableFuture;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
 import com.google.iam.v1.GetIamPolicyRequest;
 import com.google.iam.v1.Policy;
 import com.google.iam.v1.SetIamPolicyRequest;
@@ -62,18 +59,18 @@
 import com.google.pubsub.v1.PullResponse;
 import com.google.pubsub.v1.Subscription;
 import com.google.pubsub.v1.Topic;
-
 import io.grpc.ManagedChannel;
 import io.grpc.Status.Code;
 import io.grpc.netty.NegotiationType;
 import io.grpc.netty.NettyChannelBuilder;
-
-import org.joda.time.Duration;
-
 import java.io.IOException;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.joda.time.Duration;
 
 public class DefaultPubSubRpc implements PubSubRpc {
 
@@ -111,26 +108,54 @@ protected ChannelProvider getChannelProvider() {
   }
 
   private static final class PullFutureImpl
-      extends ForwardingListenableFuture.SimpleForwardingListenableFuture<PullResponse>
       implements PullFuture {
 
-    PullFutureImpl(ListenableFuture<PullResponse> delegate) {
-      super(delegate);
+    private final RpcFuture<PullResponse> delegate;
+
+    PullFutureImpl(RpcFuture<PullResponse> delegate) {
+      this.delegate = delegate;
+    }
+
+    @Override
+    public boolean cancel(boolean mayInterruptIfRunning) {
+      return delegate.cancel(mayInterruptIfRunning);
+    }
+
+    @Override
+    public PullResponse get() throws InterruptedException, ExecutionException {
+      return delegate.get();
+    }
+
+    @Override
+    public PullResponse get(long timeout, TimeUnit unit)
+        throws InterruptedException, ExecutionException, TimeoutException {
+      return delegate.get(timeout, unit);
+    }
+
+    @Override
+    public boolean isCancelled() {
+      return delegate.isCancelled();
+    }
+
+    @Override
+    public boolean isDone() {
+      return delegate.isDone();
     }
 
     @Override
     public void addCallback(final PullCallback callback) {
-      Futures.addCallback(delegate(), new FutureCallback<PullResponse>() {
-        @Override
-        public void onSuccess(PullResponse result) {
-          callback.success(result);
-        }
-
-        @Override
-        public void onFailure(Throwable error) {
-          callback.failure(error);
-        }
-      });
+      delegate.addCallback(
+          new RpcFutureCallback<PullResponse>() {
+            @Override
+            public void onSuccess(PullResponse result) {
+              callback.success(result);
+            }
+
+            @Override
+            public void onFailure(Throwable error) {
+              callback.failure(error);
+            }
+          });
     }
   }
 
@@ -178,21 +203,23 @@ public DefaultPubSubRpc(PubSubOptions options) throws IOException {
     }
   }
 
-  private static <V> ListenableFuture<V> translate(ListenableFuture<V> from,
-      final boolean idempotent, int... returnNullOn) {
+  private static <V> RpcFuture<V> translate(
+      RpcFuture<V> from, final boolean idempotent, int... returnNullOn) {
     final Set<Integer> returnNullOnSet = Sets.newHashSetWithExpectedSize(returnNullOn.length);
     for (int value : returnNullOn) {
       returnNullOnSet.add(value);
     }
-    return Futures.catching(from, ApiException.class, new Function<ApiException, V>() {
-      @Override
-      public V apply(ApiException exception) {
-        if (returnNullOnSet.contains(exception.getStatusCode().value())) {
-          return null;
-        }
-        throw new PubSubException(exception, idempotent);
-      }
-    });
+    return from.catching(
+        ApiException.class,
+        new RpcFuture.Function<ApiException, V>() {
+          @Override
+          public V apply(ApiException exception) {
+            if (returnNullOnSet.contains(exception.getStatusCode().value())) {
+              return null;
+            }
+            throw new PubSubException(exception, idempotent);
+          }
+        });
   }
 
   @Override

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PublisherSettings.java

@@ -527,11 +527,8 @@ private static Builder createDefault() {
           .publishSettings()
           .getBundlingSettingsBuilder()
           .setElementCountThreshold(10)
-          .setElementCountLimit(1000)
           .setRequestByteThreshold(1024)
-          .setRequestByteLimit(10485760)
-          .setDelayThreshold(Duration.millis(10))
-          .setBlockingCallCountThreshold(1);
+          .setDelayThreshold(Duration.millis(10));
       builder
           .publishSettings()
           .setRetryableCodes(RETRYABLE_CODE_DEFINITIONS.get("one_plus_delivery"))