|
36 | 36 | import com.google.api.gax.core.FixedExecutorProvider; |
37 | 37 | import com.google.api.gax.core.InstantiatingExecutorProvider; |
38 | 38 | import com.google.api.gax.grpc.GrpcCallContext; |
| 39 | +import com.google.api.gax.httpjson.HttpJsonCallContext; |
| 40 | +import com.google.api.gax.httpjson.InstantiatingHttpJsonChannelProvider; |
39 | 41 | import com.google.api.gax.retrying.RetrySettings; |
| 42 | +import com.google.api.gax.rpc.ApiCallContext; |
40 | 43 | import com.google.api.gax.rpc.HeaderProvider; |
41 | 44 | import com.google.api.gax.rpc.NoHeaderProvider; |
42 | 45 | import com.google.api.gax.rpc.StatusCode; |
43 | 46 | import com.google.api.gax.rpc.TransportChannelProvider; |
44 | 47 | import com.google.auth.oauth2.GoogleCredentials; |
45 | 48 | import com.google.cloud.pubsub.v1.stub.GrpcPublisherStub; |
| 49 | +import com.google.cloud.pubsub.v1.stub.HttpJsonPublisherStub; |
46 | 50 | import com.google.cloud.pubsub.v1.stub.PublisherStub; |
47 | 51 | import com.google.cloud.pubsub.v1.stub.PublisherStubSettings; |
48 | 52 | import com.google.common.base.Preconditions; |
@@ -120,9 +124,10 @@ public class Publisher implements PublisherInterface { |
120 | 124 |
|
121 | 125 | private final boolean enableCompression; |
122 | 126 | private final long compressionBytesThreshold; |
| 127 | + private final boolean enableRESTJsonTransport; |
123 | 128 |
|
124 | | - private final GrpcCallContext publishContext; |
125 | | - private final GrpcCallContext publishContextWithCompression; |
| 129 | + private final ApiCallContext publishContext; |
| 130 | + private final ApiCallContext publishContextWithCompression; |
126 | 131 |
|
127 | 132 | /** The maximum number of messages in one request. Defined by the API. */ |
128 | 133 | public static long getApiMaxRequestElementCount() { |
@@ -152,6 +157,8 @@ private Publisher(Builder builder) throws IOException { |
152 | 157 | this.messageTransform = builder.messageTransform; |
153 | 158 | this.enableCompression = builder.enableCompression; |
154 | 159 | this.compressionBytesThreshold = builder.compressionBytesThreshold; |
| 160 | + this.enableRESTJsonTransport = |
| 161 | + builder.channelProvider instanceof InstantiatingHttpJsonChannelProvider; |
155 | 162 |
|
156 | 163 | messagesBatches = new HashMap<>(); |
157 | 164 | messagesBatchLock = new ReentrantLock(); |
@@ -199,15 +206,24 @@ private Publisher(Builder builder) throws IOException { |
199 | 206 | StatusCode.Code.UNAVAILABLE) |
200 | 207 | .setRetrySettings(retrySettingsBuilder.build()) |
201 | 208 | .setBatchingSettings(BatchingSettings.newBuilder().setIsEnabled(false).build()); |
202 | | - this.publisherStub = GrpcPublisherStub.create(stubSettings.build()); |
| 209 | + this.publisherStub = |
| 210 | + this.enableRESTJsonTransport |
| 211 | + ? HttpJsonPublisherStub.create(stubSettings.build()) |
| 212 | + : GrpcPublisherStub.create(stubSettings.build()); |
203 | 213 | backgroundResourceList.add(publisherStub); |
204 | 214 | backgroundResources = new BackgroundResourceAggregation(backgroundResourceList); |
205 | 215 | shutdown = new AtomicBoolean(false); |
206 | 216 | messagesWaiter = new Waiter(); |
207 | | - this.publishContext = GrpcCallContext.createDefault(); |
| 217 | + this.publishContext = |
| 218 | + this.enableRESTJsonTransport |
| 219 | + ? HttpJsonCallContext.createDefault() |
| 220 | + : GrpcCallContext.createDefault(); |
208 | 221 | this.publishContextWithCompression = |
209 | | - GrpcCallContext.createDefault() |
210 | | - .withCallOptions(CallOptions.DEFAULT.withCompression(GZIP_COMPRESSION)); |
| 222 | + this.enableRESTJsonTransport |
| 223 | + ? this.publishContext |
| 224 | + : // TODO |
| 225 | + GrpcCallContext.createDefault() |
| 226 | + .withCallOptions(CallOptions.DEFAULT.withCompression(GZIP_COMPRESSION)); |
211 | 227 | } |
212 | 228 |
|
213 | 229 | /** Topic which the publisher publishes to. */ |
@@ -448,7 +464,7 @@ private void publishAllWithoutInflightForKey(final String orderingKey) { |
448 | 464 | } |
449 | 465 |
|
450 | 466 | private ApiFuture<PublishResponse> publishCall(OutstandingBatch outstandingBatch) { |
451 | | - GrpcCallContext context = publishContext; |
| 467 | + ApiCallContext context = publishContext; |
452 | 468 | if (enableCompression && outstandingBatch.batchSizeBytes >= compressionBytesThreshold) { |
453 | 469 | context = publishContextWithCompression; |
454 | 470 | } |
|
0 commit comments