Skip to content

Commit 7c4678d

Browse files
mziccardmziccard
authored
Refactor API call and connection settings for gRPC services (#1163)
1 parent 36ba7c3 commit 7c4678d

7 files changed

Lines changed: 96 additions & 85 deletions

File tree

‎gcloud-java-core/src/main/java/com/google/cloud/ByteArray.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -135,30 +135,30 @@ public final void copyTo(byte[] target) {
135135
/**
136136
* Creates a {@code ByteArray} object given an array of bytes. The bytes are copied.
137137
*/
138-
public final static ByteArray copyFrom(byte[] bytes) {
138+
public static final ByteArray copyFrom(byte[] bytes) {
139139
return new ByteArray(ByteString.copyFrom(bytes));
140140
}
141141

142142
/**
143143
* Creates a {@code ByteArray} object given a string. The string is encoded in {@code UTF-8}. The
144144
* bytes are copied.
145145
*/
146-
public final static ByteArray copyFrom(String string) {
146+
public static final ByteArray copyFrom(String string) {
147147
return new ByteArray(ByteString.copyFrom(string, StandardCharsets.UTF_8));
148148
}
149149

150150
/**
151151
* Creates a {@code ByteArray} object given a {@link ByteBuffer}. The bytes are copied.
152152
*/
153-
public final static ByteArray copyFrom(ByteBuffer bytes) {
153+
public static final ByteArray copyFrom(ByteBuffer bytes) {
154154
return new ByteArray(ByteString.copyFrom(bytes));
155155
}
156156

157157
/**
158158
* Creates a {@code ByteArray} object given an {@link InputStream}. The stream is read into the
159159
* created object.
160160
*/
161-
public final static ByteArray copyFrom(InputStream input) throws IOException {
161+
public static final ByteArray copyFrom(InputStream input) throws IOException {
162162
return new ByteArray(ByteString.readFrom(input));
163163
}
164164
}

‎gcloud-java-core/src/main/java/com/google/cloud/GrpcServiceOptions.java‎

Lines changed: 41 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,20 @@
1818

1919
import static com.google.common.base.MoreObjects.firstNonNull;
2020

21+
import com.google.api.gax.core.ConnectionSettings;
22+
import com.google.api.gax.core.RetrySettings;
23+
import com.google.api.gax.grpc.ApiCallSettings;
24+
import com.google.auth.oauth2.GoogleCredentials;
2125
import com.google.cloud.spi.ServiceRpcFactory;
2226
import com.google.common.annotations.VisibleForTesting;
2327
import com.google.common.base.Preconditions;
28+
import com.google.common.net.HostAndPort;
2429

2530
import io.grpc.internal.SharedResourceHolder;
2631
import io.grpc.internal.SharedResourceHolder.Resource;
2732

33+
import org.joda.time.Duration;
34+
2835
import java.io.IOException;
2936
import java.io.ObjectInputStream;
3037
import java.util.Objects;
@@ -160,8 +167,8 @@ public B executorFactory(ExecutorFactory<ScheduledExecutorService> executorFacto
160167
* Sets the timeout for the initial RPC, in milliseconds. Subsequent calls will use this value
161168
* adjusted according to {@link #timeoutMultiplier(double)}. Default value is 20000.
162169
*
163-
* @throws IllegalArgumentException if the provided timeout is &lt; 0
164170
* @return the builder
171+
* @throws IllegalArgumentException if the provided timeout is &lt; 0
165172
*/
166173
public B initialTimeout(int initialTimeout) {
167174
Preconditions.checkArgument(initialTimeout > 0, "Initial timeout must be > 0");
@@ -173,8 +180,8 @@ public B initialTimeout(int initialTimeout) {
173180
* Sets the timeout multiplier. This value is used to compute the timeout for a retried RPC.
174181
* Timeout is computed as {@code timeoutMultiplier * previousTimeout}. Default value is 1.5.
175182
*
176-
* @throws IllegalArgumentException if the provided timeout multiplier is &lt; 0
177183
* @return the builder
184+
* @throws IllegalArgumentException if the provided timeout multiplier is &lt; 0
178185
*/
179186
public B timeoutMultiplier(double timeoutMultiplier) {
180187
Preconditions.checkArgument(timeoutMultiplier >= 1.0, "Timeout multiplier must be >= 1");
@@ -216,6 +223,38 @@ protected ExecutorFactory<ScheduledExecutorService> executorFactory() {
216223
return executorFactory;
217224
}
218225

226+
/**
227+
* Returns a builder for API call settings.
228+
*/
229+
protected ApiCallSettings.Builder apiCallSettings() {
230+
// todo(mziccard): specify timeout these settings:
231+
// retryParams().retryMaxAttempts(), retryParams().retryMinAttempts()
232+
final RetrySettings.Builder builder = RetrySettings.newBuilder()
233+
.setTotalTimeout(Duration.millis(retryParams().totalRetryPeriodMillis()))
234+
.setInitialRpcTimeout(Duration.millis(initialTimeout()))
235+
.setRpcTimeoutMultiplier(timeoutMultiplier())
236+
.setMaxRpcTimeout(Duration.millis(maxTimeout()))
237+
.setInitialRetryDelay(Duration.millis(retryParams().initialRetryDelayMillis()))
238+
.setRetryDelayMultiplier(retryParams().retryDelayBackoffFactor())
239+
.setMaxRetryDelay(Duration.millis(retryParams().maxRetryDelayMillis()));
240+
return ApiCallSettings.newBuilder().setRetrySettingsBuilder(builder);
241+
}
242+
243+
/**
244+
* Returns a builder for connection-related settings.
245+
*/
246+
protected ConnectionSettings.Builder connectionSettings() {
247+
HostAndPort hostAndPort = HostAndPort.fromString(host());
248+
ConnectionSettings.Builder builder = ConnectionSettings.newBuilder()
249+
.setServiceAddress(hostAndPort.getHostText())
250+
.setPort(hostAndPort.getPort());
251+
GoogleCredentials credentials = authCredentials().credentials();
252+
if (credentials != null) {
253+
builder.provideCredentialsWith(credentials.createScoped(scopes()));
254+
}
255+
return builder;
256+
}
257+
219258
/**
220259
* Returns the timeout for the initial RPC, in milliseconds. Subsequent calls will use this value
221260
* adjusted according to {@link #timeoutMultiplier()}. Default value is 20000.

‎gcloud-java-core/src/main/java/com/google/cloud/MonitoredResourceDescriptor.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -39,13 +39,13 @@ public class MonitoredResourceDescriptor implements Serializable {
3939
private static final long serialVersionUID = -3702077512777687441L;
4040
public static final Function<com.google.api.MonitoredResourceDescriptor,
4141
MonitoredResourceDescriptor> FROM_PB_FUNCTION =
42-
new Function<com.google.api.MonitoredResourceDescriptor, MonitoredResourceDescriptor>() {
43-
@Override
44-
public MonitoredResourceDescriptor apply(
45-
com.google.api.MonitoredResourceDescriptor pb) {
46-
return fromPb(pb);
47-
}
48-
};
42+
new Function<com.google.api.MonitoredResourceDescriptor, MonitoredResourceDescriptor>() {
43+
@Override
44+
public MonitoredResourceDescriptor apply(
45+
com.google.api.MonitoredResourceDescriptor pb) {
46+
return fromPb(pb);
47+
}
48+
};
4949

5050
private final String type;
5151
private final String name;

‎gcloud-java-logging/src/main/java/com/google/cloud/logging/LoggingOptions.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import com.google.cloud.logging.spi.DefaultLoggingRpc;
2121
import com.google.cloud.logging.spi.LoggingRpc;
2222
import com.google.cloud.logging.spi.LoggingRpcFactory;
23+
import com.google.cloud.logging.spi.v2.LoggingServiceV2Settings;
2324
import com.google.common.collect.ImmutableSet;
2425

2526
import java.io.IOException;
@@ -31,7 +32,8 @@ public class LoggingOptions extends GrpcServiceOptions<Logging, LoggingRpc, Logg
3132
private static final long serialVersionUID = -2996451684945061075L;
3233
private static final String LOGGING_SCOPE = "https://www.googleapis.com/auth/logging.admin";
3334
private static final Set<String> SCOPES = ImmutableSet.of(LOGGING_SCOPE);
34-
private static final String DEFAULT_HOST = "https://logging.googleapis.com";
35+
private static final String DEFAULT_HOST = LoggingServiceV2Settings.getDefaultServiceAddress()
36+
+ ':' + LoggingServiceV2Settings.getDefaultServicePort();
3537

3638
public static class DefaultLoggingFactory implements LoggingFactory {
3739
private static final LoggingFactory INSTANCE = new DefaultLoggingFactory();

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

Lines changed: 21 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,11 @@
1818

1919
import static com.google.common.base.MoreObjects.firstNonNull;
2020

21-
import com.google.api.gax.core.RetrySettings;
21+
import com.google.api.gax.core.ConnectionSettings;
2222
import com.google.api.gax.grpc.ApiCallSettings;
2323
import com.google.api.gax.grpc.ApiException;
24-
import com.google.auth.oauth2.GoogleCredentials;
2524
import com.google.cloud.AuthCredentials;
2625
import com.google.cloud.GrpcServiceOptions.ExecutorFactory;
27-
import com.google.cloud.RetryParams;
2826
import com.google.cloud.logging.LoggingException;
2927
import com.google.cloud.logging.LoggingOptions;
3028
import com.google.cloud.logging.spi.v2.ConfigServiceV2Api;
@@ -65,8 +63,6 @@
6563
import io.grpc.netty.NegotiationType;
6664
import io.grpc.netty.NettyChannelBuilder;
6765

68-
import org.joda.time.Duration;
69-
7066
import java.io.IOException;
7167
import java.util.Set;
7268
import java.util.concurrent.Future;
@@ -94,10 +90,21 @@ private InternalLoggingOptions(LoggingOptions options) {
9490
protected ExecutorFactory<ScheduledExecutorService> executorFactory() {
9591
return super.executorFactory();
9692
}
93+
94+
@Override
95+
protected ApiCallSettings.Builder apiCallSettings() {
96+
return super.apiCallSettings();
97+
}
98+
99+
@Override
100+
protected ConnectionSettings.Builder connectionSettings() {
101+
return super.connectionSettings();
102+
}
97103
}
98104

99105
public DefaultLoggingRpc(LoggingOptions options) throws IOException {
100-
executorFactory = new InternalLoggingOptions(options).executorFactory();
106+
InternalLoggingOptions internalOptions = new InternalLoggingOptions(options);
107+
executorFactory = internalOptions.executorFactory();
101108
executor = executorFactory.get();
102109
String libraryName = options.libraryName();
103110
String libraryVersion = firstNonNull(options.libraryVersion(), "");
@@ -121,18 +128,15 @@ public DefaultLoggingRpc(LoggingOptions options) throws IOException {
121128
logBuilder.provideChannelWith(channel, true);
122129
metricsBuilder.provideChannelWith(channel, true);
123130
} else {
124-
GoogleCredentials credentials = options.authCredentials().credentials();
125-
confBuilder.provideChannelWith(
126-
credentials.createScoped(ConfigServiceV2Settings.getDefaultServiceScopes()));
127-
logBuilder.provideChannelWith(
128-
credentials.createScoped(LoggingServiceV2Settings.getDefaultServiceScopes()));
129-
metricsBuilder.provideChannelWith(
130-
credentials.createScoped(MetricsServiceV2Settings.getDefaultServiceScopes()));
131+
ConnectionSettings connectionSettings = internalOptions.connectionSettings().build();
132+
confBuilder.provideChannelWith(connectionSettings);
133+
logBuilder.provideChannelWith(connectionSettings);
134+
metricsBuilder.provideChannelWith(connectionSettings);
131135
}
132-
ApiCallSettings.Builder callBuilder = apiCallSettings(options);
133-
confBuilder.applyToAllApiMethods(callBuilder);
134-
logBuilder.applyToAllApiMethods(callBuilder);
135-
metricsBuilder.applyToAllApiMethods(callBuilder);
136+
ApiCallSettings.Builder callSettingsBuilder = internalOptions.apiCallSettings();
137+
confBuilder.applyToAllApiMethods(callSettingsBuilder);
138+
logBuilder.applyToAllApiMethods(callSettingsBuilder);
139+
metricsBuilder.applyToAllApiMethods(callSettingsBuilder);
136140
configApi = ConfigServiceV2Api.create(confBuilder.build());
137141
loggingApi = LoggingServiceV2Api.create(logBuilder.build());
138142
metricsApi = MetricsServiceV2Api.create(metricsBuilder.build());
@@ -141,21 +145,6 @@ public DefaultLoggingRpc(LoggingOptions options) throws IOException {
141145
}
142146
}
143147

144-
private static ApiCallSettings.Builder apiCallSettings(LoggingOptions options) {
145-
// todo(mziccard): specify timeout these settings:
146-
// retryParams.retryMaxAttempts(), retryParams.retryMinAttempts()
147-
RetryParams retryParams = options.retryParams();
148-
final RetrySettings.Builder builder = RetrySettings.newBuilder()
149-
.setTotalTimeout(Duration.millis(retryParams.totalRetryPeriodMillis()))
150-
.setInitialRpcTimeout(Duration.millis(options.initialTimeout()))
151-
.setRpcTimeoutMultiplier(options.timeoutMultiplier())
152-
.setMaxRpcTimeout(Duration.millis(options.maxTimeout()))
153-
.setInitialRetryDelay(Duration.millis(retryParams.initialRetryDelayMillis()))
154-
.setRetryDelayMultiplier(retryParams.retryDelayBackoffFactor())
155-
.setMaxRetryDelay(Duration.millis(retryParams.maxRetryDelayMillis()));
156-
return ApiCallSettings.newBuilder().setRetrySettingsBuilder(builder);
157-
}
158-
159148
private static <V> Future<V> translate(ListenableFuture<V> from, final boolean idempotent,
160149
int... returnNullOn) {
161150
final Set<Integer> returnNullOnSet = Sets.newHashSetWithExpectedSize(returnNullOn.length);

‎gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/PubSubOptions.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,8 @@ public class PubSubOptions extends GrpcServiceOptions<PubSub, PubSubRpc, PubSubO
3232
private static final long serialVersionUID = 5640180400046623305L;
3333
private static final String PUBSUB_SCOPE = "https://www.googleapis.com/auth/pubsub";
3434
private static final Set<String> SCOPES = ImmutableSet.of(PUBSUB_SCOPE);
35-
private static final String DEFAULT_HOST = PublisherSettings.getDefaultServiceAddress();
35+
private static final String DEFAULT_HOST = PublisherSettings.getDefaultServiceAddress()
36+
+ ':' + PublisherSettings.getDefaultServicePort();
3637

3738
public static class DefaultPubSubFactory implements PubSubFactory {
3839
private static final PubSubFactory INSTANCE = new DefaultPubSubFactory();

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

Lines changed: 18 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -19,13 +19,10 @@
1919
import static com.google.common.base.MoreObjects.firstNonNull;
2020

2121
import com.google.api.gax.core.ConnectionSettings;
22-
import com.google.api.gax.core.RetrySettings;
2322
import com.google.api.gax.grpc.ApiCallSettings;
2423
import com.google.api.gax.grpc.ApiException;
25-
import com.google.auth.oauth2.GoogleCredentials;
2624
import com.google.cloud.AuthCredentials;
2725
import com.google.cloud.GrpcServiceOptions.ExecutorFactory;
28-
import com.google.cloud.RetryParams;
2926
import com.google.cloud.pubsub.PubSubException;
3027
import com.google.cloud.pubsub.PubSubOptions;
3128
import com.google.cloud.pubsub.spi.v1.PublisherApi;
@@ -64,8 +61,6 @@
6461
import io.grpc.netty.NegotiationType;
6562
import io.grpc.netty.NettyChannelBuilder;
6663

67-
import org.joda.time.Duration;
68-
6964
import java.io.IOException;
7065
import java.util.Set;
7166
import java.util.concurrent.Future;
@@ -92,6 +87,16 @@ private InternalPubSubOptions(PubSubOptions options) {
9287
protected ExecutorFactory<ScheduledExecutorService> executorFactory() {
9388
return super.executorFactory();
9489
}
90+
91+
@Override
92+
protected ApiCallSettings.Builder apiCallSettings() {
93+
return super.apiCallSettings();
94+
}
95+
96+
@Override
97+
protected ConnectionSettings.Builder connectionSettings() {
98+
return super.connectionSettings();
99+
}
95100
}
96101

97102
private static final class PullFutureImpl
@@ -119,7 +124,8 @@ public void onFailure(Throwable error) {
119124
}
120125

121126
public DefaultPubSubRpc(PubSubOptions options) throws IOException {
122-
executorFactory = new InternalPubSubOptions(options).executorFactory();
127+
InternalPubSubOptions internalOptions = new InternalPubSubOptions(options);
128+
executorFactory = internalOptions.executorFactory();
123129
executor = executorFactory.get();
124130
String libraryName = options.libraryName();
125131
String libraryVersion = firstNonNull(options.libraryVersion(), "");
@@ -139,46 +145,20 @@ public DefaultPubSubRpc(PubSubOptions options) throws IOException {
139145
pubBuilder.provideChannelWith(channel, true);
140146
subBuilder.provideChannelWith(channel, true);
141147
} else {
142-
GoogleCredentials credentials = options.authCredentials().credentials();
143-
ConnectionSettings pubConnectionSettings = ConnectionSettings.newBuilder()
144-
.setServiceAddress(options.host())
145-
.setPort(PublisherSettings.getDefaultServicePort())
146-
.provideCredentialsWith(
147-
credentials.createScoped(PublisherSettings.getDefaultServiceScopes()))
148-
.build();
149-
ConnectionSettings subConnectionSettings = ConnectionSettings.newBuilder()
150-
.setServiceAddress(options.host())
151-
.setPort(SubscriberSettings.getDefaultServicePort())
152-
.provideCredentialsWith(
153-
credentials.createScoped(SubscriberSettings.getDefaultServiceScopes()))
154-
.build();
155-
pubBuilder.provideChannelWith(pubConnectionSettings);
156-
subBuilder.provideChannelWith(subConnectionSettings);
148+
ConnectionSettings connectionSettings = internalOptions.connectionSettings().build();
149+
pubBuilder.provideChannelWith(connectionSettings);
150+
subBuilder.provideChannelWith(connectionSettings);
157151
}
158-
pubBuilder.applyToAllApiMethods(apiCallSettings(options));
159-
subBuilder.applyToAllApiMethods(apiCallSettings(options));
152+
ApiCallSettings.Builder callSettingsBuilder = internalOptions.apiCallSettings();
153+
pubBuilder.applyToAllApiMethods(callSettingsBuilder);
154+
subBuilder.applyToAllApiMethods(callSettingsBuilder);
160155
publisherApi = PublisherApi.create(pubBuilder.build());
161156
subscriberApi = SubscriberApi.create(subBuilder.build());
162157
} catch (Exception ex) {
163158
throw new IOException(ex);
164159
}
165160
}
166161

167-
private static ApiCallSettings.Builder apiCallSettings(PubSubOptions options) {
168-
// TODO: specify timeout these settings:
169-
// retryParams.retryMaxAttempts(), retryParams.retryMinAttempts()
170-
RetryParams retryParams = options.retryParams();
171-
final RetrySettings.Builder builder = RetrySettings.newBuilder()
172-
.setTotalTimeout(Duration.millis(retryParams.totalRetryPeriodMillis()))
173-
.setInitialRpcTimeout(Duration.millis(options.initialTimeout()))
174-
.setRpcTimeoutMultiplier(options.timeoutMultiplier())
175-
.setMaxRpcTimeout(Duration.millis(options.maxTimeout()))
176-
.setInitialRetryDelay(Duration.millis(retryParams.initialRetryDelayMillis()))
177-
.setRetryDelayMultiplier(retryParams.retryDelayBackoffFactor())
178-
.setMaxRetryDelay(Duration.millis(retryParams.maxRetryDelayMillis()));
179-
return ApiCallSettings.newBuilder().setRetrySettingsBuilder(builder);
180-
}
181-
182162
private static <V> ListenableFuture<V> translate(ListenableFuture<V> from,
183163
final boolean idempotent, int... returnNullOn) {
184164
final Set<Integer> returnNullOnSet = Sets.newHashSetWithExpectedSize(returnNullOn.length);

0 commit comments

Comments
 (0)