Add Redis Streams option for job delivery

[ajac-zero]

This pull request adds a basic implementation of Redis Streams, in order to avoid polling for new jobs in the worker and reduce latency, in accordance with objective 4 of issue #437.

To create a worker that listens to a Redis Stream, we can use the cli or specify it in the code directly.

CLI: arq worker.WorkerSettings --stream

Code:

class WorkerSettings:
    functions = [...]
    stream = True
    ...

On the client, they must specify that they want to deliver a job to a worker through a Redis Stream.

redis = await create_pool(RedisSettings())
await redis.enqueue_job('hello_world', _use_stream=True)

Here are the results of a very simple benchmark that showcases the potential of using Redis Streams for improved latency.

Polling:

Average time: 0.268s

Streaming:

Average time: 0.012s

[codecov[bot]]

Codecov Report

Attention: Patch coverage is 50.00000% with 17 lines in your changes are missing coverage. Please review.

Project coverage is 95.00%. Comparing base (94cd878) to head (1b44875). Report is 11 commits behind head on main.

Additional details and impacted files

```diff @@ Coverage Diff @@ ## main #451 +/- ## ========================================== - Coverage 96.27% 95.00% -1.28% ========================================== Files 11 11 Lines 1074 1100 +26 Branches 209 197 -12 ========================================== + Hits 1034 1045 +11 - Misses 19 31 +12 - Partials 21 24 +3 ``` | [Files](https://app.codecov.io/gh/samuelcolvin/arq/pull/451?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin) | Coverage Δ | | |---|---|---| | [arq/constants.py](https://app.codecov.io/gh/samuelcolvin/arq/pull/451?src=pr&el=tree&filepath=arq%2Fconstants.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin#diff-YXJxL2NvbnN0YW50cy5weQ==) | `100.00% <100.00%> (ø)` | | | [arq/cli.py](https://app.codecov.io/gh/samuelcolvin/arq/pull/451?src=pr&el=tree&filepath=arq%2Fcli.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin#diff-YXJxL2NsaS5weQ==) | `96.49% <60.00%> (-3.51%)` | :arrow_down: | | [arq/connections.py](https://app.codecov.io/gh/samuelcolvin/arq/pull/451?src=pr&el=tree&filepath=arq%2Fconnections.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin#diff-YXJxL2Nvbm5lY3Rpb25zLnB5) | `88.81% <33.33%> (-1.25%)` | :arrow_down: | | [arq/worker.py](https://app.codecov.io/gh/samuelcolvin/arq/pull/451?src=pr&el=tree&filepath=arq%2Fworker.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin#diff-YXJxL3dvcmtlci5weQ==) | `94.88% <43.47%> (-2.29%)` | :arrow_down: | ------ [Continue to review full report in Codecov by Sentry](https://app.codecov.io/gh/samuelcolvin/arq/pull/451?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin). > **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin) > `Δ = absolute (impact)`, `ø = not affected`, `? = missing data` > Powered by [Codecov](https://app.codecov.io/gh/samuelcolvin/arq/pull/451?dropdown=coverage&src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin). Last update [1315583...1b44875](https://app.codecov.io/gh/samuelcolvin/arq/pull/451?dropdown=coverage&src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin).

[gaby]

@ajac-zero This may need a unit-test with stream enabled.

[ajac-zero]

Sure thing @gaby . I was wondering how I should go about that...

What I started doing was add a stream parameter to the worker tests, and then wrap them so they run twice, once with stream and once without.

But I feel this might not be the best way to do things, maybe I should focus on some vital tests? What do you suggest?

[gaby]

@ajac-zero That's probably a great starting point, running current tests with "stream" set to false. Then running the test suite with "stream" set to True. This will require setting the source of the data to use Streams.