Fixes #413.

This PR makes sure that the ACK deadline set for the received messages is always consistent with what the leaser uses internally when extending the ACK deadlines for the leased messages.

See the issue description and a [comment](#413 (comment)) explaining a possible sequence of events that lead to a bug.

**PR checklist**
- [x] Make sure to open an issue as a [bug/issue](https://github.com/googleapis/python-pubsub/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [x] Ensure the tests and linter pass
- [x] Code coverage does not decrease (if any source code was changed)
- [x] Appropriate docs were updated (if necessary)

* test: presubmit against Lite samples

* mark new presubmit test required

Fixes #423.

If subscriber client is used as a context manager, we need to block until the shutdown is complete before leaving the `with` block. See the issue description for more details.

**PR checklist:**
- [x] Make sure to open an issue as a [bug/issue](https://github.com/googleapis/python-pubsub/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [x] Ensure the tests and linter pass
- [x] Code coverage does not decrease (if any source code was changed)
- [x] Appropriate docs were updated (if necessary)

…r client (#299)

* feat: allow retry and timeout settings on publisher client

* build: update generated code and update doc

* Propagate publish timeout to the RPC call

* test publisher client

* Fix timeout parameter type in docstrings

* Apply timeout changes to async publisher client

* Introduce TimeoutType type alias

* Update PublisherOptions docs

* Use type alias for timeout in upgrading guide

* Widen timeout types in generated publisher clients

* Use TimeoutType in handwritten code

* Remove redundant backslash

* Use DEFAULT as a defualt timeout in base sequencer

* Do not accept plain None as a valid timeout

Using no timeout is not a good idea, but if one really wants to,
they can pass it in as ConstantTimeout(None).

As a side effect, the logic of converting a constant into a
COnstantTimeout instance can be removed, as this is already handled
in api-core for int and float values.

Co-authored-by: Carlos de la Guardia <cguardia@yahoo.com>

* sample: add publish flow control sample and other nits

* restore changes in untouched files

* address peter's comments

* Add publisher flow controller test for FIFO order

* Simplify _run_in_daemon() test helper

* Fix message slots not acquired in FIFO order

* Unify the logic for distributing any free capacity

* Use OrderedDict for the FIFO queue

This allows to hold the queue of threads and their reservation data
in a single structure, no need for the separate deque and reservations
dict.

Co-authored-by: release-please[bot] <55107282+release-please[bot]@users.noreply.github.com>