pubsub: enable streaming pull

Moving this to a branch.
If nothing else, we should perf test this before
declaring it release-ready.

This commit re-enables streaming pull.
Unlike previous implementation,
it does not fall back to polling if streaming is unavailable,
since the streaming pull endpoint should be working
by the time this is released.

This commit fixes a bug deadline modification code.
Previously we record an Instant we receive a message,
then call Instant::getNano to get the time in nanoseconds.
This is incorrect since getNano returns the nanosecond
from the beginning of that second, not since an epoch.
The fix is to simply save the time since epoch instead of Instant.

This commit also slightly changes how errors are logged,
so that errors that occur after the service is being shut down
don't spam the console.

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

@@ -173,13 +173,13 @@ private class AckHandler implements FutureCallback<AckReply> {
     private final String ackId;
     private final int outstandingBytes;
     private final AtomicBoolean acked;
-    private final Instant receivedTime;
+    private final long receivedTimeMillis;
 
     AckHandler(String ackId, int outstandingBytes) {
       this.ackId = ackId;
       this.outstandingBytes = outstandingBytes;
       acked = new AtomicBoolean(false);
-      receivedTime = Instant.ofEpochMilli(clock.millisTime());
+      receivedTimeMillis = clock.millisTime();
     }
 
     @Override
@@ -207,7 +207,6 @@ public void onSuccess(AckReply reply) {
             pendingAcks.add(ackId);
           }
           // Record the latency rounded to the next closest integer.
-          long receivedTimeMillis = TimeUnit.NANOSECONDS.toMillis(receivedTime.getNano());
           ackLatencyDistribution.record(
               Ints.saturatedCast(
                   (long) Math.ceil((clock.millisTime() - receivedTimeMillis) / 1000D)));

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

@@ -29,12 +29,9 @@
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.pubsub.v1.StreamingPullRequest;
 import com.google.pubsub.v1.StreamingPullResponse;
-import com.google.pubsub.v1.SubscriberGrpc;
-import io.grpc.CallOptions;
-import io.grpc.Channel;
+import com.google.pubsub.v1.SubscriberGrpc.SubscriberStub;
 import io.grpc.Status;
 import io.grpc.stub.ClientCallStreamObserver;
-import io.grpc.stub.ClientCalls;
 import io.grpc.stub.ClientResponseObserver;
 import java.util.ArrayList;
 import java.util.List;
@@ -55,7 +52,7 @@ final class StreamingSubscriberConnection extends AbstractApiService implements
 
   private Duration channelReconnectBackoff = INITIAL_CHANNEL_RECONNECT_BACKOFF;
 
-  private final Channel channel;
+  private final SubscriberStub asyncStub;
 
   private final String subscription;
   private final ScheduledExecutorService executor;
@@ -69,14 +66,14 @@ public StreamingSubscriberConnection(
       Duration maxAckExtensionPeriod,
       int streamAckDeadlineSeconds,
       Distribution ackLatencyDistribution,
-      Channel channel,
+      SubscriberStub asyncStub,
       FlowController flowController,
       ScheduledExecutorService executor,
       @Nullable ScheduledExecutorService alarmsExecutor,
       ApiClock clock) {
     this.subscription = subscription;
     this.executor = executor;
-    this.channel = channel;
+    this.asyncStub = asyncStub;
     this.messageDispatcher =
         new MessageDispatcher(
             receiver,
@@ -101,8 +98,8 @@ protected void doStart() {
   @Override
   protected void doStop() {
     messageDispatcher.stop();
-    notifyStopped();
     requestObserver.onError(Status.CANCELLED.asException());
+    notifyStopped();
   }
 
   private class StreamingPullResponseObserver
@@ -137,7 +134,6 @@ public void run() {
 
     @Override
     public void onError(Throwable t) {
-      logger.log(Level.WARNING, "Terminated streaming with exception", t);
       errorFuture.setException(t);
     }
 
@@ -154,9 +150,7 @@ private void initialize() {
         new StreamingPullResponseObserver(errorFuture);
     final ClientCallStreamObserver<StreamingPullRequest> requestObserver =
         (ClientCallStreamObserver<StreamingPullRequest>)
-            (ClientCalls.asyncBidiStreamingCall(
-                channel.newCall(SubscriberGrpc.METHOD_STREAMING_PULL, CallOptions.DEFAULT),
-                responseObserver));
+            (asyncStub.streamingPull(responseObserver));
     logger.log(
         Level.FINER,
         "Initializing stream to subscription {0} with deadline {1}",
@@ -173,6 +167,9 @@ private void initialize() {
         new FutureCallback<Void>() {
           @Override
           public void onSuccess(@Nullable Void result) {
+            if (!isAlive()) {
+              return;
+            }
             channelReconnectBackoff = INITIAL_CHANNEL_RECONNECT_BACKOFF;
             // The stream was closed. And any case we want to reopen it to continue receiving
             // messages.
@@ -186,6 +183,7 @@ public void onFailure(Throwable cause) {
               logger.log(Level.FINE, "pull failure after service no longer running", cause);
               return;
             }
+            logger.log(Level.WARNING, "Terminated streaming with exception", cause);
             if (StatusUtil.isRetryable(cause)) {
               long backoffMillis = channelReconnectBackoff.toMillis();
               channelReconnectBackoff = channelReconnectBackoff.plusMillis(backoffMillis);

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java

@@ -38,9 +38,11 @@
 import com.google.pubsub.v1.GetSubscriptionRequest;
 import com.google.pubsub.v1.SubscriberGrpc;
 import com.google.pubsub.v1.SubscriberGrpc.SubscriberFutureStub;
+import com.google.pubsub.v1.SubscriberGrpc.SubscriberStub;
 import com.google.pubsub.v1.Subscription;
 import com.google.pubsub.v1.SubscriptionName;
 import io.grpc.CallCredentials;
+import io.grpc.Channel;
 import io.grpc.ManagedChannel;
 import io.grpc.auth.MoreCallCredentials;
 import java.io.IOException;
@@ -257,22 +259,25 @@ public void close() {
     // same executor, it will deadlock: the thread will be stuck waiting for connections
     // to start but cannot start the connections.
     // For this reason, we spawn a dedicated thread. Starting subscriber should be rare.
-    new Thread(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          startPollingConnections();
-          notifyStarted();
-        } catch (Throwable t) {
-          notifyFailed(t);
-        }
-      }
-    }).start();
+    new Thread(
+            new Runnable() {
+              @Override
+              public void run() {
+                try {
+                  // startPollingConnections();
+                  startStreamingConnections();
+                  notifyStarted();
+                } catch (Throwable t) {
+                  notifyFailed(t);
+                }
+              }
+            })
+        .start();
   }
 
   @Override
   protected void doStop() {
-    // stopAllStreamingConnections();
+    stopAllStreamingConnections();
     stopAllPollingConnections();
     try {
       for (AutoCloseable closeable : closeables) {
@@ -284,6 +289,94 @@ protected void doStop() {
     }
   }
 
+  private void startStreamingConnections() throws IOException {
+    synchronized (streamingSubscriberConnections) {
+      Credentials credentials = credentialsProvider.getCredentials();
+      CallCredentials callCredentials =
+          credentials == null ? null : MoreCallCredentials.from(credentials);
+
+      for (Channel channel : channels) {
+        SubscriberStub stub = SubscriberGrpc.newStub(channel);
+        if (callCredentials != null) {
+          stub = stub.withCallCredentials(callCredentials);
+        }
+        streamingSubscriberConnections.add(
+            new StreamingSubscriberConnection(
+                cachedSubscriptionNameString,
+                receiver,
+                ackExpirationPadding,
+                maxAckExtensionPeriod,
+                streamAckDeadlineSeconds,
+                ackLatencyDistribution,
+                stub,
+                flowController,
+                executor,
+                alarmsExecutor,
+                clock));
+      }
+      startConnections(
+          streamingSubscriberConnections,
+          new Listener() {
+            @Override
+            public void failed(State from, Throwable failure) {
+              // If a connection failed is because of a fatal error, we should fail the
+              // whole subscriber.
+              stopAllStreamingConnections();
+              try {
+                notifyFailed(failure);
+              } catch (IllegalStateException e) {
+                if (isRunning()) {
+                  throw e;
+                }
+                // It could happen that we are shutting down while some channels fail.
+              }
+            }
+          });
+    }
+
+    ackDeadlineUpdater =
+        executor.scheduleAtFixedRate(
+            new Runnable() {
+              @Override
+              public void run() {
+                // It is guaranteed this will be <= MAX_ACK_DEADLINE_SECONDS, the max of the API.
+                long ackLatency =
+                    ackLatencyDistribution.getNthPercentile(PERCENTILE_FOR_ACK_DEADLINE_UPDATES);
+                if (ackLatency > 0) {
+                  long ackExpirationPaddingMillis = ackExpirationPadding.toMillis();
+                  int possibleStreamAckDeadlineSeconds =
+                      Math.max(
+                          MIN_ACK_DEADLINE_SECONDS,
+                          Ints.saturatedCast(
+                              Math.max(
+                                  ackLatency,
+                                  TimeUnit.MILLISECONDS.toSeconds(ackExpirationPaddingMillis))));
+                  if (streamAckDeadlineSeconds != possibleStreamAckDeadlineSeconds) {
+                    streamAckDeadlineSeconds = possibleStreamAckDeadlineSeconds;
+                    logger.log(
+                        Level.FINER,
+                        "Updating stream deadline to {0} seconds.",
+                        streamAckDeadlineSeconds);
+                    for (StreamingSubscriberConnection subscriberConnection :
+                        streamingSubscriberConnections) {
+                      subscriberConnection.updateStreamAckDeadline(streamAckDeadlineSeconds);
+                    }
+                  }
+                }
+              }
+            },
+            ACK_DEADLINE_UPDATE_PERIOD.toMillis(),
+            ACK_DEADLINE_UPDATE_PERIOD.toMillis(),
+            TimeUnit.MILLISECONDS);
+  }
+
+  private void stopAllStreamingConnections() {
+    stopConnections(streamingSubscriberConnections);
+    if (ackDeadlineUpdater != null) {
+      ackDeadlineUpdater.cancel(true);
+    }
+  }
+
   // Starts polling connections. Blocks until all connections declare themselves running.
   private void startPollingConnections() throws IOException {
     synchronized (pollingSubscriberConnections) {

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/SubscriberTest.java

@@ -80,7 +80,7 @@ public class SubscriberTest {
 
   @Parameters
   public static Collection<Object[]> data() {
-    return Arrays.asList(new Object[][] {{false}});
+    return Arrays.asList(new Object[][] {{true}});
   }
 
   static class TestReceiver implements MessageReceiver {
@@ -205,6 +205,9 @@ public void testAckSingleMessage() throws Exception {
 
   @Test
   public void testGetSubscriptionOnce() throws Exception {
+    if (isStreamingTest) {
+      return;
+    }
     Subscriber subscriber = startSubscriber(getTestSubscriberBuilder(testReceiver));
 
     sendMessages(ImmutableList.of("A"));