Skip to content

Commit f033397

Browse files
pongadpongad
authored
---
yaml --- r: 8251 b: refs/heads/master c: 91ad4f4 h: refs/heads/master i: 8249: d063afa 8247: a335894
1 parent e0b19c1 commit f033397

6 files changed

Lines changed: 160 additions & 54 deletions

File tree

‎[refs]‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
---
2-
refs/heads/master: a9099c57595ac5af1ebd6630502b73443ad8800d
2+
refs/heads/master: 91ad4f465ed5cbd2f7853290a33555b00e5c9aed
33
refs/heads/travis: 47e4fee4fd5af9b2a8ce46f23c72ec95f9b195b2
44
refs/heads/gh-pages: 0114589947fea527ea3831330789accedbecf45c
55
refs/tags/0.0.9: 22f1839238f66c39e67ed4dfdcd273b1ae2e8444

‎trunk/google-cloud-datastore/src/main/java/com/google/cloud/datastore/spi/v1/HttpDatastoreRpc.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,9 @@
1919
import com.google.api.client.http.HttpRequest;
2020
import com.google.api.client.http.HttpRequestInitializer;
2121
import com.google.api.client.http.HttpTransport;
22-
import com.google.cloud.http.HttpTransportOptions;
2322
import com.google.cloud.datastore.DatastoreException;
2423
import com.google.cloud.datastore.DatastoreOptions;
24+
import com.google.cloud.http.HttpTransportOptions;
2525
import com.google.datastore.v1.AllocateIdsRequest;
2626
import com.google.datastore.v1.AllocateIdsResponse;
2727
import com.google.datastore.v1.BeginTransactionRequest;

‎trunk/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -173,13 +173,13 @@ private class AckHandler implements FutureCallback<AckReply> {
173173
private final String ackId;
174174
private final int outstandingBytes;
175175
private final AtomicBoolean acked;
176-
private final Instant receivedTime;
176+
private final long receivedTimeMillis;
177177

178178
AckHandler(String ackId, int outstandingBytes) {
179179
this.ackId = ackId;
180180
this.outstandingBytes = outstandingBytes;
181181
acked = new AtomicBoolean(false);
182-
receivedTime = Instant.ofEpochMilli(clock.millisTime());
182+
receivedTimeMillis = clock.millisTime();
183183
}
184184

185185
@Override
@@ -207,7 +207,6 @@ public void onSuccess(AckReply reply) {
207207
pendingAcks.add(ackId);
208208
}
209209
// Record the latency rounded to the next closest integer.
210-
long receivedTimeMillis = TimeUnit.NANOSECONDS.toMillis(receivedTime.getNano());
211210
ackLatencyDistribution.record(
212211
Ints.saturatedCast(
213212
(long) Math.ceil((clock.millisTime() - receivedTimeMillis) / 1000D)));

‎trunk/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java‎

Lines changed: 10 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -29,12 +29,9 @@
2929
import com.google.common.util.concurrent.SettableFuture;
3030
import com.google.pubsub.v1.StreamingPullRequest;
3131
import com.google.pubsub.v1.StreamingPullResponse;
32-
import com.google.pubsub.v1.SubscriberGrpc;
33-
import io.grpc.CallOptions;
34-
import io.grpc.Channel;
32+
import com.google.pubsub.v1.SubscriberGrpc.SubscriberStub;
3533
import io.grpc.Status;
3634
import io.grpc.stub.ClientCallStreamObserver;
37-
import io.grpc.stub.ClientCalls;
3835
import io.grpc.stub.ClientResponseObserver;
3936
import java.util.ArrayList;
4037
import java.util.List;
@@ -55,7 +52,7 @@ final class StreamingSubscriberConnection extends AbstractApiService implements
5552

5653
private Duration channelReconnectBackoff = INITIAL_CHANNEL_RECONNECT_BACKOFF;
5754

58-
private final Channel channel;
55+
private final SubscriberStub asyncStub;
5956

6057
private final String subscription;
6158
private final ScheduledExecutorService executor;
@@ -69,14 +66,14 @@ public StreamingSubscriberConnection(
6966
Duration maxAckExtensionPeriod,
7067
int streamAckDeadlineSeconds,
7168
Distribution ackLatencyDistribution,
72-
Channel channel,
69+
SubscriberStub asyncStub,
7370
FlowController flowController,
7471
ScheduledExecutorService executor,
7572
@Nullable ScheduledExecutorService alarmsExecutor,
7673
ApiClock clock) {
7774
this.subscription = subscription;
7875
this.executor = executor;
79-
this.channel = channel;
76+
this.asyncStub = asyncStub;
8077
this.messageDispatcher =
8178
new MessageDispatcher(
8279
receiver,
@@ -101,8 +98,8 @@ protected void doStart() {
10198
@Override
10299
protected void doStop() {
103100
messageDispatcher.stop();
104-
notifyStopped();
105101
requestObserver.onError(Status.CANCELLED.asException());
102+
notifyStopped();
106103
}
107104

108105
private class StreamingPullResponseObserver
@@ -137,7 +134,6 @@ public void run() {
137134

138135
@Override
139136
public void onError(Throwable t) {
140-
logger.log(Level.WARNING, "Terminated streaming with exception", t);
141137
errorFuture.setException(t);
142138
}
143139

@@ -154,9 +150,7 @@ private void initialize() {
154150
new StreamingPullResponseObserver(errorFuture);
155151
final ClientCallStreamObserver<StreamingPullRequest> requestObserver =
156152
(ClientCallStreamObserver<StreamingPullRequest>)
157-
(ClientCalls.asyncBidiStreamingCall(
158-
channel.newCall(SubscriberGrpc.METHOD_STREAMING_PULL, CallOptions.DEFAULT),
159-
responseObserver));
153+
(asyncStub.streamingPull(responseObserver));
160154
logger.log(
161155
Level.FINER,
162156
"Initializing stream to subscription {0} with deadline {1}",
@@ -173,6 +167,9 @@ private void initialize() {
173167
new FutureCallback<Void>() {
174168
@Override
175169
public void onSuccess(@Nullable Void result) {
170+
if (!isAlive()) {
171+
return;
172+
}
176173
channelReconnectBackoff = INITIAL_CHANNEL_RECONNECT_BACKOFF;
177174
// The stream was closed. And any case we want to reopen it to continue receiving
178175
// messages.
@@ -186,6 +183,7 @@ public void onFailure(Throwable cause) {
186183
logger.log(Level.FINE, "pull failure after service no longer running", cause);
187184
return;
188185
}
186+
logger.log(Level.WARNING, "Terminated streaming with exception", cause);
189187
if (StatusUtil.isRetryable(cause)) {
190188
long backoffMillis = channelReconnectBackoff.toMillis();
191189
channelReconnectBackoff = channelReconnectBackoff.plusMillis(backoffMillis);

0 commit comments

Comments
 (0)