GCP Pub / Sub Client, GrpcSubscriberStub, Тайм-аут

Question

Я пытаюсь использовать GrpcSubscriberStub для "Синхронного извлечения", как в примерах https://cloud.google.com/pubsub/docs/pull

SubscriberStubSettings subscriberStubSettings =
    SubscriberStubSettings.newBuilder()
        .setTransportChannelProvider(
            SubscriberStubSettings.defaultGrpcTransportProviderBuilder()
                .setMaxInboundMessageSize(20 << 20) // 20MB
                .build())
        .build();

try (SubscriberStub subscriber = GrpcSubscriberStub.create(subscriberStubSettings)) {
  // String projectId = "my-project-id";
  // String subscriptionId = "my-subscription-id";
  // int numOfMessages = 10;   // max number of messages to be pulled
  String subscriptionName = ProjectSubscriptionName.format(projectId, subscriptionId);
  PullRequest pullRequest =
      PullRequest.newBuilder()
          .setMaxMessages(numOfMessages)
          .setReturnImmediately(false) // return immediately if messages are not available
          .setSubscription(subscriptionName)
          .build();

  // use pullCallable().futureCall to asynchronously perform this operation
  PullResponse pullResponse = subscriber.pullCallable().call(pullRequest);
  List<String> ackIds = new ArrayList<>();
  for (ReceivedMessage message : pullResponse.getReceivedMessagesList()) {
    // handle received message
    // ...
    ackIds.add(message.getAckId());
  }
  // acknowledge received messages
  AcknowledgeRequest acknowledgeRequest =
      AcknowledgeRequest.newBuilder()
          .setSubscription(subscriptionName)
          .addAllAckIds(ackIds)
          .build();
  // use acknowledgeCallable().futureCall to asynchronously perform this operation
  subscriber.acknowledgeCallable().call(acknowledgeRequest);
  return pullResponse.getReceivedMessagesList();
}

Но я не понимаю, как установить тайм-аут для извлечения, с опцией "setReturnImmediately (false)".

Вы можете мне помочь?

Answer 1

Чтобы изменить время ожидания, вам нужно изменить SubscriberStubSettings.

Например, чтобы установить общее время ожидания на 30 секунд:

SubscriberStubSettings.Builder subscriberStubSettingsBuilder =
    SubscriberStubSettings.newBuilder();
subscriberStubSettingsBuilder
    .pullSettings()
    .setRetrySettings(
        subscriberStubSettingsBuilder.pullSettings().getRetrySettings().toBuilder()
            .setTotalTimeout(Duration.ofSeconds(30))
            .build());
SubscriberStubSettings subscriberStubSettings = subscriberStubSettingsBuilder.build();

Это основан на документации SubscriberStubSettings здесь .