Adding stream commands #299
Conversation
Codecov Report
@@ Coverage Diff @@
## master #299 +/- ##
==========================================
- Coverage 96.85% 96.71% -0.15%
==========================================
Files 54 56 +2
Lines 7447 7825 +378
Branches 525 546 +21
==========================================
+ Hits 7213 7568 +355
- Misses 171 185 +14
- Partials 63 72 +9
Continue to review full report at Codecov.
|
|
It appears to me that the |
aioredis/commands/streams.py
Outdated
| # TODO: Add the MAXLEN parameter | ||
| flattened = [] | ||
| for k, v in fields.items(): | ||
| flattened += [k, v] |
There was a problem hiding this comment.
Ah yes. Changed.
|
Great @adamcharnock! Do you know where I can find the implementation details related to Redis? Did you use this: https://github.com/redis/redis-rcp/blob/master/RCP11.md? |
|
Hi @barrachri, I think that is a different kind of stream, one more to do with how the redis protocol encodes data on the wire. I used the following: The branch can be found here: |
|
Thanks. Is using I was trying to think about a use case where the order of the key=value might be important after you used |
|
I used an OrderedDict simply because the ordering of the key/value pairs is explicitly respected and maintained by Redis Streams. I therefore wanted to respect this in the implementation. I expect someone will find a use for it, even if I'm not sure what it would be. Perhaps one reads a stream record then writes it back to another stream. The new record has a different field order. As Redis guarantees order, other clients could legitimately decide that this copied record was different to the original, despite there being no intention to modify it during the copying process. As a secondary reason, I've occasionally seen hard-to-diagnose bugs crop you due to the non-deterministic ordering of dictionaries. This normally occurs when the developer assumes a regular dict has a particular ordering, which then sporadically changes under different environments and python versions. Given that the key/values do in fact have ordering in this case, it seemed like an easy way to avoid this issue. Also, I suspect ignoring the ordering provided & documented by Redis would probably cause developers to walk right into this kind of bug. |
|
Clarified, thanks! |
|
|
|
Should we wait for redis stable with streaming feature? |
|
@barrachri That is great to hear. @asvetlov Not sure if that was directed at me – but I have no view on the matter. I'll see about getting |
|
My question is addressed mostly to @popravich |
… unstable branch. Improving stream test skipping logic.
|
Update: Redis 5 RC3 is now out. What is the view on merging at this time? |
|
Ok, travis is happy now. The code coverage diff seems to show lack of coverage on sentinel as the sticking point, but as I haven't touched that I'm going to assuming is not my fault. |
aioredis/commands/streams.py
Outdated
| PY_VER = sys.version_info | ||
|
|
||
| if PY_VER < (3, 0): | ||
| from itertools import izip as zip |
There was a problem hiding this comment.
These lines don't make sense. We don't support Python 2.x.
There was a problem hiding this comment.
Yep, not sure what I was thinking here
aioredis/commands/streams.py
Outdated
|
|
||
| def parse_lists_to_dicts(lists): | ||
| """ Convert [[a, 1, b, 2], ...] into [{a:1, b: 2}, ...]""" | ||
| def _list_to_dict(list_): |
There was a problem hiding this comment.
This function is almost the same as fields_to_dict except return type,
It is better to make a single function and pass OrderedDict/dict` types as argument.
| fut = self.execute(b'XREAD', *args) | ||
| return wait_convert(fut, parse_messages_by_stream) | ||
|
|
||
| def xread_group(self, group_name, consumer_name, streams, timeout=0, |
There was a problem hiding this comment.
...and to 12 more methods below
aioredis/commands/streams.py
Outdated
| consumer=None): | ||
| # Returns: total pel messages, min id, max id, count | ||
| ssc = [start, stop, count] | ||
| if any(ssc) and not all(ssc): |
There was a problem hiding this comment.
This check may trigger false-positive exceptions in case when start=0.
Please add test for this case.
There was a problem hiding this comment.
Good catch! Will add test and fix
tests/stream_commands_test.py
Outdated
| "unstable branch") | ||
| async def test_xadd_maxlen_exact(redis, server_bin): | ||
| message_id1 = await redis.xadd('test_stream', {'f1': 'v1'}) # noqa | ||
| sleep(0.001) # Ensure the millisecond-based message ID increments |
There was a problem hiding this comment.
Do you really need a thread-blocking sleep?
There was a problem hiding this comment.
Ha, I certainly don't :-)
|
Thank you for the comments @popravich, I'll take a look at making the changes now/shortly. |
|
I've just pushed changes for everything bar the doc strings, which I'll take a look at now. |
aioredis/commands/streams.py
Outdated
| def fields_to_dict(fields, type_=OrderedDict): | ||
| """Convert a flat list of key/values into an OrderedDict""" | ||
| fields_iterator = iter(fields) | ||
| return OrderedDict(zip(fields_iterator, fields_iterator)) |
There was a problem hiding this comment.
type_ argument is not used
There was a problem hiding this comment.
Urgh, thanks. Fixed.
|
AFAIK I've now responded to all your change requests. Do let me know if there is anything else though 👍 |
|
Lets wait for CI tasks to complete. |
|
Merged! Thanks |
|
That's great to hear, thank you very much for taking the time to review this. I'll be happy to revisit this if this API changes before Redis 5.0 stable. Once this sees a release I'll update lightbus to point to the particular aioredis version. |
* Adding XADD/XRANGE/XREAD stream commands * Fixing flake errors. Mostly long lines. * Work on CI detection of streams availability * Fixing test error caused by undefined ordering of fields * Removing stray print statement * Removing stray print statement * Making using of .extend() rather than list += [], as per @barrachri's suggestion. * Streams PR: Fixing use of asyncio in tests for recent changes in aioredis project * Adding MAXLEN support to XADD command * Steams PR: Adding XREVRANGE command * Streams: Updating travis config to match that on master. Building for unstable branch. Improving stream test skipping logic. * Flake fixes * Adding message_id parameter to XADD, plus test * Streams: Moving default value for xadd's message_id parameter onto method signature (as per @hoh's suggestion) * Initial untested implementation of consumer group commands (tests coming) * Work on tests from streams consumer group commands * Adding tests for remaining consumer group commands * Fixing flake errors * Testing use of xgroup_create() when group already exists * Adding test for xpending() using start/stop/count args * Fixing flake error * Fixing flake error (again) * Removing unnecessary python 2 compatability * Refactoring parse_lists_to_dicts() at @popravich's request * Fixing error in xpending sanity check * Replacing blocking sleep with asyncio sleep * Adding docstrings * Enabling testing of xgroup_setid, xgroup_destroy, xinfo_consumers (as these are now present in redis) * Fixing error in fields_to_dict() refactoring (type_ was not used) * Fixing flake errors
@antirez is in the process of adding streams to Redis. This PR adds initial support for the
XADD/XRANGE/XREADcommands in aioredis.Tests
I've written tests and added the streams Redis branch to the travis config. In theory, the tests will only be run when the streams branch is available.
The logic is a bit hacky, but I figured it will only be until streams makes it into 4.0(now improved as the hacky version didn't actually work).Style
I'm very open to feedback on my implementation style for the new commands. I've tried to stick to the aioredis style, but I think I will have probably wandered here or there. Happy to make any corrections or alterations.
Reason
This is my first time using aioredis. I'm working on another project in which I would like to experiment with Redis streams, plus I also need asyncio support. I therefore thought I'd contribute this here.
As I said, happy to make any changes etc.
Remaining work
MAXLENparameter inxadd()xread()