original_docstring: Maintain all of the leases being managed.
  
          This method modifies the ack deadline for all of the managed
          ack IDs, then waits for most of that time (but with jitter), and
          repeats.
          
docstring: Maintain all of the leases being managed.
  This method modifies the ack deadline for all of the managed
  ack IDs, then waits for most of that time (but with jitter), and
  repeats.
docstring_tokens: Maintain all of the leases being managed . This method modifies the ack deadline for all of the managed ack IDs then waits for most of that time ( but with jitter ) and repeats .
code: def maintain_leases(self) -> None:
          """Maintain all of the leases being managed.
  
          This method modifies the ack deadline for all of the managed
          ack IDs, then waits for most of that time (but with jitter), and
          repeats.
          """
          while not self._stop_event.is_set():
              # Determine the appropriate duration for the lease. This is
              # based off of how long previous messages have taken to ack, with
              # a sensible default and within the ranges allowed by Pub/Sub.
              # Also update the deadline currently used if enough new ACK data has been
              # gathered since the last deadline update.
              deadline = self._manager._obtain_ack_deadline(maybe_update=True)
              _LOGGER.debug("The current deadline value is %d seconds.", deadline)
  
              # Make a copy of the leased messages. This is needed because it's
              # possible for another thread to modify the dictionary while
              # we're iterating over it.
              leased_messages = copy.copy(self._leased_messages)
  
              # Drop any leases that are beyond the max lease time. This ensures
              # that in the event of a badly behaving actor, we can drop messages
              # and allow the Pub/Sub server to resend them.
              cutoff = time.time() - self._manager.flow_control.max_lease_duration
              to_drop = [
                  requests.DropRequest(ack_id, item.size, item.ordering_key)
                  for ack_id, item in leased_messages.items()
                  if item.sent_time < cutoff
              ]
  
              if to_drop:
                  _LOGGER.warning(
                      "Dropping %s items because they were leased too long.", len(to_drop)
                  )
                  assert self._manager.dispatcher is not None
                  self._manager.dispatcher.drop(to_drop)
  
              # Remove dropped items from our copy of the leased messages (they
              # have already been removed from the real one by
              # self._manager.drop(), which calls self.remove()).
              for item in to_drop:
                  leased_messages.pop(item.ack_id)
  
              # Create a streaming pull request.
              # We do not actually call `modify_ack_deadline` over and over
              # because it is more efficient to make a single request.
              ack_ids = leased_messages.keys()
              if ack_ids:
                  _LOGGER.debug("Renewing lease for %d ack IDs.", len(ack_ids))
  
                  # NOTE: This may not work as expected if ``consumer.active``
                  #       has changed since we checked it. An implementation
                  #       without any sort of race condition would require a
                  #       way for ``send_request`` to fail when the consumer
                  #       is inactive.
                  assert self._manager.dispatcher is not None
                  self._manager.dispatcher.modify_ack_deadline(
                      [requests.ModAckRequest(ack_id, deadline) for ack_id in ack_ids]
                  )
  
              # Now wait an appropriate period of time and do this again.
              #
              # We determine the appropriate period of time based on a random
              # period between 0 seconds and 90% of the lease. This use of
              # jitter (http://bit.ly/2s2ekL7) helps decrease contention in cases
              # where there are many clients.
              snooze = random.uniform(0.0, deadline * 0.9)
              _LOGGER.debug("Snoozing lease management for %f seconds.", snooze)
              self._stop_event.wait(timeout=snooze)
  
          _LOGGER.info("%s exiting.", _LEASE_WORKER_NAME)
code_tokens: def maintain_leases ( self ) -> None : while not self . _stop_event . is_set ( ) : deadline = self . _manager . _obtain_ack_deadline ( maybe_update = True ) _LOGGER . debug ( "The current deadline value is %d seconds." , deadline ) leased_messages = copy . copy ( self . _leased_messages ) cutoff = time . time ( ) - self . _manager . flow_control . max_lease_duration to_drop = [ requests . DropRequest ( ack_id , item . size , item . ordering_key ) for ack_id , item in leased_messages . items ( ) if item . sent_time < cutoff ] if to_drop : _LOGGER . warning ( "Dropping %s items because they were leased too long." , len ( to_drop ) ) assert self . _manager . dispatcher is not None self . _manager . dispatcher . drop ( to_drop ) for item in to_drop : leased_messages . pop ( item . ack_id ) ack_ids = leased_messages . keys ( ) if ack_ids : _LOGGER . debug ( "Renewing lease for %d ack IDs." , len ( ack_ids ) ) assert self . _manager . dispatcher is not None self . _manager . dispatcher . modify_ack_deadline ( [ requests . ModAckRequest ( ack_id , deadline ) for ack_id in ack_ids ] ) snooze = random . uniform ( 0.0 , deadline * 0.9 ) _LOGGER . debug ( "Snoozing lease management for %f seconds." , snooze ) self . _stop_event . wait ( timeout = snooze ) _LOGGER . info ( "%s exiting." , _LEASE_WORKER_NAME )
short_docstring: Maintain all of the leases being managed.
short_docstring_tokens: Maintain all of the leases being managed .
comment: """Maintain all of the leases being managed.
  
          This method modifies the ack deadline for all of the managed
          ack IDs, then waits for most of that time (but with jitter), and
          repeats.
          """$SEP$# Determine the appropriate duration for the lease. This is$SEP$# based off of how long previous messages have taken to ack, with$SEP$# a sensible default and within the ranges allowed by Pub/Sub.$SEP$# Also update the deadline currently used if enough new ACK data has been$SEP$# gathered since the last deadline update.$SEP$# Make a copy of the leased messages. This is needed because it's$SEP$# possible for another thread to modify the dictionary while$SEP$# we're iterating over it.$SEP$# Drop any leases that are beyond the max lease time. This ensures$SEP$# that in the event of a badly behaving actor, we can drop messages$SEP$# and allow the Pub/Sub server to resend them.$SEP$# Remove dropped items from our copy of the leased messages (they$SEP$# have already been removed from the real one by$SEP$# self._manager.drop(), which calls self.remove()).$SEP$# Create a streaming pull request.$SEP$# We do not actually call `modify_ack_deadline` over and over$SEP$# because it is more efficient to make a single request.$SEP$# NOTE: This may not work as expected if ``consumer.active``$SEP$#       has changed since we checked it. An implementation$SEP$#       without any sort of race condition would require a$SEP$#       way for ``send_request`` to fail when the consumer$SEP$#       is inactive.$SEP$# Now wait an appropriate period of time and do this again.$SEP$#$SEP$# We determine the appropriate period of time based on a random$SEP$# period between 0 seconds and 90% of the lease. This use of$SEP$# jitter (http://bit.ly/2s2ekL7) helps decrease contention in cases$SEP$# where there are many clients.